作为控制台应用程序,您的应用程序在默认的
同步上下文上运行,在此上下文中,
await
继续回调将在等待任务已完成的相同线程上调用。如果您想在
await SendAwaitResponse
之后切换线程,可以使用
await Task.Yield()
来实现:
await SendAwaitResponse("first message");
await Task.Yield();
// will be continued on a pool thread
// ...
SendAwaitResponse("second message").Wait(); // so no deadlock
您可以通过将
Thread.CurrentThread.ManagedThreadId
存储在
Task.Result
中,并在
await
之后将其与当前线程的id进行比较,从而进一步改进此功能。如果仍然在同一线程上,则执行
await Task.Yield()
。
虽然我理解SendAwaitResponse
是您实际代码的简化版本(就像您在问题中展示的那样),但它仍然完全在内部同步。为什么您希望在其中进行任何线程切换呢?
无论如何,您可能应该重新设计逻辑,使其不会对您当前所在的线程做出任何假设。避免混合使用await
和Task.Wait()
,并使所有代码都异步化。通常,在顶级别(例如在Main
内部)只需要使用一个Wait()
。
[编辑] 在ReceiverRun
中调用task.SetResult(msg)
实际上将控制流转移到您在task
上await
的点,由于默认同步上下文的行为,没有线程切换。因此,处理实际消息的代码接管了ReceiverRun
线程。最终,在同一线程上调用SendAwaitResponse("second message").Wait()
,导致死锁。
以下是一个控制台应用程序代码,模拟您的示例。它在ProcessAsync
内使用await Task.Yield()
来在单独的线程上调度继续,因此控制流返回到ReceiverRun
,并且没有死锁。
using System;
using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Tasks;
namespace ConsoleApplication
{
class Program
{
class Worker
{
public struct Response
{
public string message;
public int threadId;
}
CancellationToken _token;
readonly ConcurrentQueue<string> _messages = new ConcurrentQueue<string>();
readonly ConcurrentDictionary<string, TaskCompletionSource<Response>> _requests = new ConcurrentDictionary<string, TaskCompletionSource<Response>>();
public Worker(CancellationToken token)
{
_token = token;
}
string ReadNextMessage()
{
string message;
while (!_messages.TryDequeue(out message))
{
Thread.Sleep(100);
_token.ThrowIfCancellationRequested();
}
return message;
}
public void ReceiverRun()
{
LogThread("Enter ReceiverRun");
while (true)
{
var msg = ReadNextMessage();
LogThread("ReadNextMessage: " + msg);
var tcs = _requests[msg];
tcs.SetResult(new Response { message = msg, threadId = Thread.CurrentThread.ManagedThreadId });
_token.ThrowIfCancellationRequested();
}
}
Task<Response> SendAwaitResponse(string msg)
{
LogThread("SendAwaitResponse: " + msg);
var tcs = new TaskCompletionSource<Response>();
_requests.TryAdd(msg, tcs);
_messages.Enqueue(msg);
return tcs.Task;
}
public async Task ProcessAsync()
{
LogThread("Enter Worker.ProcessAsync");
var task1 = SendAwaitResponse("first message");
await task1;
LogThread("result1: " + task1.Result.message);
if (task1.Result.threadId == Thread.CurrentThread.ManagedThreadId)
await Task.Yield();
var task2 = SendAwaitResponse("second message");
task2.Wait();
LogThread("result2: " + task2.Result.message);
var task3 = SendAwaitResponse("third message");
task3.Wait();
LogThread("result3: " + task3.Result.message);
var task4 = SendAwaitResponse("fourth message");
await task4;
LogThread("result4: " + task4.Result.message);
if (task4.Result.threadId == Thread.CurrentThread.ManagedThreadId)
await Task.Yield();
var task5 = SendAwaitResponse("fifth message");
task5.Wait();
LogThread("result5: " + task5.Result.message);
LogThread("Leave Worker.ProcessAsync");
}
public static void LogThread(string message)
{
Console.WriteLine("{0}, thread: {1}", message, Thread.CurrentThread.ManagedThreadId);
}
}
static void Main(string[] args)
{
Worker.LogThread("Enter Main");
var cts = new CancellationTokenSource(5000);
var worker = new Worker(cts.Token);
Task receiver = Task.Run(() => worker.ReceiverRun());
Task main = worker.ProcessAsync();
try
{
Task.WaitAll(main, receiver);
}
catch (Exception e)
{
Console.WriteLine("Exception: " + e.Message);
}
Worker.LogThread("Leave Main");
Console.ReadLine();
}
}
}
这与在
ReceiverRun
内执行
Task.Run(() => task.SetResult(msg))
没有太大区别。我能想到的唯一优点是你可以明确控制何时切换线程。这样,你可以尽可能长时间地保持在同一线程上(例如,对于
task2
、
task3
、
task4
),但在
task5.Wait()
上避免死锁仍需要另一个线程切换。
两种解决方案最终都会使线程池增长,这对性能和可扩展性来说是不好的。
现在,如果我们在上面的代码中
ProcessAsync
内部到处使用
await task
替换
task.Wait()
,我们将不必使用
await Task.Yield
,并且仍然不会出现死锁。但是,在
ProcessAsync
内部的第一个
await task1
之后的整个
await
调用链实际上将在
ReceiverRun
线程上执行。只要我们不使用其他
Wait()
风格的调用阻塞此线程,并且在处理消息时不进行大量的CPU绑定工作,这种方法可能可以正常工作(异步IO-bound
await
风格的调用仍然应该可以正常工作,并且它们可能实际上会触发隐式线程切换)。
说到这里,我认为你需要一个单独的线程,并安装一个串行同步上下文来处理消息(类似于
WindowsFormsSynchronizationContext
)。这就是包含
awaits
的异步代码应该运行的地方。你仍然需要避免在该线程上使用
Task.Wait
。如果个别消息处理需要大量的CPU绑定工作,则应该使用
Task.Run
来进行此类工作。对于异步IO绑定调用,可以保持在同一线程上。
你可能需要查看
@StephenCleary的
Nito Asynchronous Library中的
ActionDispatcher
/
ActionDispatcherSynchronizationContext
,以获取异步消息处理逻辑。希望Stephen能够提供更好的答案。
Task.Run(() => task.SetResult(msg));
(类似地使用SetException()
)。但我希望有更好的选择。 - svick