线程Join()导致Task.RunSynchronously不完成

Question

调用_thread.Join()会导致GetConsumingEnumerable循环停留在最后一个元素上.为什么会出现这种情况？

  public abstract class ActorBase : IDisposable
  {
    private readonly BlockingCollection<Task> _queue = new BlockingCollection<Task>(new ConcurrentQueue<Task>());
    private readonly Thread _thread;
    private bool _isDisposed;

    protected ActorBase()
    {
      _thread = new Thread(ProcessMessages);
      _thread.Start();
    }

    protected void QueueTask(Task task)
    {
      if (_isDisposed)
      {
        throw new Exception("Actor was disposed, cannot queue task.");
      }
      _queue.Add(task);
    }

    private void ProcessMessages()
    {
      foreach (var task in _queue.GetConsumingEnumerable())
      {
        task.RunSynchronously();
      }
    }

    public void Dispose()
    {
      _isDisposed = true;
      _queue.CompleteAdding();
      _thread.Join();
    }
  }

  public class SampleActor : ActorBase
  {
    private string GetThreadStatus()
    {
      Thread.Sleep(500);
      return string.Format("Running on thread {0}", Thread.CurrentThread.ManagedThreadId);
    }

    public async Task<string> GetThreadStatusAsync()
    {
      var task = new Task<string>(GetThreadStatus);
      QueueTask(task);
      return await task;
    }
  }

  class Program
  {
    public static async Task Run()
    {
      using (var sa = new SampleActor())
      {
        for (int i = 0; i < 3; i++)
        {
          Console.WriteLine(await sa.GetThreadStatusAsync());
        }
      }
    }

    public static void Main(string[] args)
    {
      Console.WriteLine("Main thread id {0}", Thread.CurrentThread.ManagedThreadId);
      var task = Task.Run(async ()=> { await Run(); });
      task.Wait();
    }
  }

这种方法的上下文是我需要确保所有操作都在一个OS线程上执行,这将允许应用程序的一部分使用与主线程不同的凭据.

Answer 1

async-await适用于延续.为了提高效率并减少调度,这些延续通常在完成上一个任务的同一个线程上运行.

这意味着在您的情况下,您的特殊线程不仅运行任务,它还在这些任务之后运行所有延续(for循环本身).您可以通过打印线程ID来看到:

using (var sa = new SampleActor())
{
    for (int i = 0; i < 3; i++)
    {
        Console.WriteLine(await sa.GetThreadStatusAsync());
        Console.WriteLine("Continue on thread :" + Thread.CurrentThread.ManagedThreadId);
    }
}

当for循环完成并且SampleActor正在处理时,Thread.Join您正在尝试加入同一个线程,这样您就会遇到死锁.你的情况归结为:

public static void Main()
{
    Thread thread = null;
    thread = new Thread(() =>
    {
        Thread.Sleep(100);
        thread.Join();
        Console.WriteLine("joined");
    });
    thread.Start();
}

在.Net 4.6中,您可以解决此问题,TaskCreationOptions.RunContinuationsAsynchronously但在当前版本中,您可以指定默认值TaskScheduler:

public Task<string> GetThreadStatusAsync()
{
    var task = new Task<string>(GetThreadStatus);
    QueueTask(task);
    return task.ContinueWith(task1 => task1.GetAwaiter().GetResult(), TaskScheduler.Default);
}