多个工作线程与一个具有异步/等待功能的工作线程

debugcn 发表于 Dev

Hengyi

我的代码当前具有以下10个工作线程。每个工作线程继续从队列中轮询作业，然后处理长时间运行的作业。

for (int k=0; k<10; k++)
{
  Task.Factory.StartNew(() => DoPollingThenWork(), TaskCreationOptions.LongRunning);
}

void DoPollingThenWork()
{
  while (true)
  {
    var msg = Poll();
    if (msg != null)
    {
      Thread.Sleep(3000); // process the I/O bound job
    }
  }
}

我正在重构底层代码以使用异步/等待模式。我想我可以将上面的代码重写为以下代码。它使用一个主线程继续创建异步任务，并使用SemaphoreSlim将并发任务的数量限制为10。

Task.Factory.StartNew(() => WorkerMainAsync(), TaskCreationOptions.LongRunning);

async Task WorkerMainAsync()
{
  SemaphoreSlim ss = new SemaphoreSlim(10);
  while (true)
  {
    await ss.WaitAsync();
    Task.Run(async () => 
              {
                await DoPollingThenWorkAsync();
                ss.Release();
              });
  }
}

async Task DoPollingThenWorkAsync()
{
  var msg = Poll();
  if (msg != null)
  {
    await Task.Delay(3000); // process the I/O-bound job
  }
}

两者的行为应相同。但是我认为第二种选择似乎更好，因为它不会阻塞线程。但不利的一面是我无法等待（优雅地停止任务），因为任务就像是一发不可收拾。第二种选择是否是替代传统工作线程模式的正确方法？

斯威克

当您拥有异步代码时，通常没有理由使用Task.Run()（甚至更糟的是Task.Factory.StartNew()）。这意味着您可以将代码更改为以下内容：

await WorkerMainAsync();

async Task WorkerMainAsync()
{
  SemaphoreSlim ss = new SemaphoreSlim(10);
  while (true)
  {
    await ss.WaitAsync();
    // you should probably store this task somewhere and then await it
    var task = DoPollingThenWorkAsync();
  }
}

async Task DoPollingThenWorkAsync(SemaphoreSlim semaphore)
{
  var msg = Poll();
  if (msg != null)
  {
    await Task.Delay(3000); // process the I/O-bound job
  }

  // this assumes you don't have to worry about exceptions
  // otherwise consider try-finally
  semaphore.Release();
}

本文收集自互联网，转载请注明来源。

如有侵权，请联系[email protected] 删除。