c# - 异步等待另一个请求而不阻塞

Question

我有一个 websocket 应用程序，它是一个 OWIN 中间件。当请求传入时，将启动 websocket 处理程序的新实例，然后在循环中等待传入消息，如下所示:

var buffer = new byte[1024*64];
Tuple<ArraySegment<byte>, WebSocketMessageType> received;
do
{
    received = await _webSocket.ReceiveMessage(buffer, _cancellationToken.Token);
    if (received.Item1.Count > 0 && someConditionForCallingDoSomething)
    {
        await DoSomething(received.Item1);
    }
    else if(isAnswer)
    {
        QueueAnswer(received.Item1);
    }
} while (received.Item2 != WebSocketMessageType.Close);

从 _webSocket.ReceiveMessage 返回的任务将在数据可用时完成。

DoSomething 处理其数据，然后通过 websocket 连接发送一些内容。然后它应该通过 websocket 连接等待消息。处理完此消息后，它应该做一些工作并返回(任务)。 也许这个小图可以更容易地解释它:

_______________
| DoSomething |
|-------------|
|    Work  ---------> Send WS message
|             |
|     ??      |
|             |
|  More Work <------- Receive WS message
|   |         |
|   V         |
|  return;    |
|_____________|

Do some work with the data
          |
          V
     Send a message
          |
          V
   Wait for an answer
          |
          V
    Process answer
          |
          V
        finish

我试图等待答案:

var answerCancel = new CancellationTokenSource();
answerCancel.CancelAfter(30 * 1000);

var answer = await Task.Run(async () => 
    {
        string tmpAnswer = null;

        while (!_concurrentAnswerDict.TryGetValue(someKey, out tmpAnswer)) {
            await Task.Delay(150, answerCancel.Token);
        }

        return tmpAnswer;
    }, answerCancel.Token);

但这似乎会阻塞，直到任务被取消。当我调试程序时，我在 30 秒后看到 QueueAnswer 的调用。我认为，Task.Run将在新线程中运行该函数，但似乎并非如此。从 Task.Run 阻塞的角度来看，对我来说，它不起作用似乎是合乎逻辑的，因为我等待 DoSomething 的执行，因此接收新消息将被阻塞也是。

我的问题是:如何实现这样的行为？如何让 DoSomething 在完成之前等待另一条 websocket 消息？

预先感谢您的每一个提示

卢卡斯

Answer 1

首先，我建议使用 SignalR，因为它们会为您处理很多此类困难的事情。但如果您想自己做，请继续阅读...

另外，我假设“do work”和“answer”消息可以以任何顺序到达同一个网络套接字，并且您正在使用 _concurrentAnswerDict协调来自 DoSomething 的传出“问题”消息带有传入的“应答”消息。

在这种情况下，您将需要一个独立于 DoSomething 的“websocket reader”任务;你不能拥有你的阅读器 await DoSomething因为这会妨碍阅读答案。我认为这是您遇到的主要问题。

这是极少数可以接受的情况之一，不 await一个任务。假设DoSomething将捕获它自己的异常并处理日志记录等，然后我们可以将它视为一个独立的“main”并忽略它返回的任务:

var buffer = new byte[1024*64];
Tuple<ArraySegment<byte>, WebSocketMessageType> received;
do
{
  received = await _webSocket.ReceiveMessage(buffer, _cancellationToken.Token);
  if (received.Item1.Count > 0 && someConditionForCallingDoSomething)
  {
    var _ = DoSomething(received.Item1);
  }
  else if(isAnswer)
  {
    QueueAnswer(received.Item1);
  }
} while (received.Item2 != WebSocketMessageType.Close);

这应该允许 QueueAnswer运行时 DoSomething尚未完成。

I thought, Task.Run will run the function in a new thread, but it seems like it does not. From the view that Task.Run is blocking, it seems logical to me that it does not work, because I await the execution of DoSomething, therefore receiving new messages will be blocked too.

Task.Run 正在在另一个线程中运行。但是DoSomething正在(异步)等待它完成，并且读取循环正在(异步)等待 DoSomething在读取下一条消息之前完成。

其他说明:

while (!_concurrentAnswerDict.TryGetValue(someKey, out tmpAnswer)) {
  await Task.Delay(150, answerCancel.Token);
}

这对我来说似乎很奇怪。我建议使用 TaskCompletionSource<Answer> 的键字典而不是Answer 。然后，QueueAnswer会打电话 TaskCompletionSource<Answer>.SetResult ，此代码将等待 TaskCompletionSource<Answer>.Task (如果需要超时，则与 Task.Delay 一起)。