等待foreach（）启动后重叠任务

Question

这是我的情况。在iHostedService中，我需要使用等待foreach（）订阅GRPC频道和处理消息。

    var channel = GrpcChannel.ForAddress("https://localhost:5001");
    var client = new Messenger.MessengerClient(channel);

    var messages= client.SubscribeToMessages();

    await foreach (var message in messages.ResponseStream.ReadAllAsync())
    {
        // do something with message
    }

由于我正在订阅流，因此我有可能错过了服务器在客户端订阅之前发送的消息。 Messengerclient具有一种方法，可以在当天获得所有消息的快照：

  var snapshotMessages = client.GetTodaysMessages();

我想在订阅流以这种方式订阅流之后致电客户端。

我正在考虑在我等待foreach（...）之前启动一个计时器，并在计时器的回调中调用client.getTodaySmessages（）。您如何看待这种方法？

一项要求是，如果我与GRPC流断开连接，则需要重新订阅并再次致电client.getTodaySmessages（）。

我愿意提出您可能有的建议

Answer 1

Microsoft有一个名为 asynchronous编程和等待async和等等， 在这里可能会对您有所帮助。

本质上，您可以将呼叫分配给task，而不是调用等待，它可以在您能够执行其他操作的同时运行。然后，您可以等待任务准备好后完成。 任务 S还具有结果属性，其中包含您所久的代码的结果。

通过调用task.whenany（）在您正在等待的一系列任务上，有一种非常有用的方法来等待多个任务。这是Microsoft中的一些代码（来自上面链接的异步编程指南），它们能够创建几个相关的早餐任务，然后等待 它们：

var breakfastTasks = new List<Task> { eggsTask, baconTask, toastTask };
while (breakfastTasks.Count > 0)
{
    Task finishedTask = await Task.WhenAny(breakfastTasks);
    if (finishedTask == eggsTask)
    {
        Console.WriteLine("Eggs are ready");
    }
    else if (finishedTask == baconTask)
    {
        Console.WriteLine("Bacon is ready");
    }
    else if (finishedTask == toastTask)
    {
        Console.WriteLine("Toast is ready");
    }
    breakfastTasks.Remove(finishedTask);
}

您可以尝试使您的foreach a <<<代码> task like（这是注销袖口，我没有尝试编译，但可能会对您有所帮助）：

var task = messages.ResponseStream.ReadAllAsync();

// do other work here, while the task runs
var snapshotMessages = client.GetTodaysMessages();

// and then deal with the result of the message stream

await task;

foreach(var message in task.Result){ 

  // process message
}

Answer 2

XY问题。

解决问题的解决方案是没有问题。

“由于我正在订阅流，因此有可能错过消息”。

这是GRPC的功能。

选择一项技术，其中“自从使用X，我没有机会丢失消息”。您想使用消息队列而不是可观察的模式。查看

• MSMQ
• Rabbit MQ
• MQTT
• MassTransit
• Redis
• Tibco
• Simpleemq
• 等，

消息代理将使用零代码解决您的问题。

Answer 3

我认为您会做这样的事情：

var channel = GrpcChannel.ForAddress("https://localhost:5001");
var client = new Messenger.MessengerClient(channel);

var messages = client.SubscribeToMessages();

// Get the stream going so that you don't miss any
var messagesEnumerableTask = messages.ResponseStream.ReadAllAsync();

// Get today's messages
var snapshotMessages = await client.GetTodaysMessages(); // I'm asuming this is actually awaitable?

// Process today's messages
foreach (var message in snapshotMessages)
{
    if (!IsProcessed(message))
    {
        Process(message);
    } 
}

// Process the stream
await foreach (var message in messagesEnumerableTask)
{
    if (!IsProcessed(message))
    {
        Process(message);
    }
}

我想您将有某种方法来确定是否已经看到消息（例如一种称为Isprocessed（Message））的方法。