在Windows服务始终运行的线程 [英] Always Running Threads on Windows Service

问题描述

我正在写一个Windows服务，将开球多个工作线程将侦听到Amazon SQS队列和处理消息。将有大约20个线程听10的队列。

I'm writing a Windows Service that will kick off multiple worker threads that will listen to Amazon SQS queues and process messages. There will be about 20 threads listening to 10 queues.

该线程将必须始终在运行，这就是为什么我倾向于以实际使用的实际线程工人循环，而不是线程池线程。

The threads will have to be always running and that's why I'm leaning towards to actually using actual threads for the worker loops rather than threadpool threads.

下面是一个顶级的实现。 Windows服务将揭开序幕多个工作线程，每个会听它的队列和处理消息。

Here is a top level implementation. Windows service will kick off multiple worker threads and each will listen to it's queue and process messages.

protected override void OnStart(string[] args)
{
     for (int i = 0; i < _workers; i++)
   {
      new Thread(RunWorker).Start();
   }
}

下面是工作的落实

public async void RunWorker()
{
  while(true)
  {
    // .. get message from amazon sqs sync.. about 20ms
    var message = sqsClient.ReceiveMessage();

    try
    {
       await PerformWebRequestAsync(message);
       await InsertIntoDbAsync(message);
    }
    catch(SomeExeception)
    {
       // ... log
       //continue to retry
       continue;
    }
    sqsClient.DeleteMessage();
  }
}

我知道我可以Task.Run执行相同的操作和执行它的线程池线程上，而不是个人开始线程，但我没有看到一个原因，因为每个线程都始终处于运行状态。

I know I can perform the same operation with Task.Run and execute it on the threadpool thread rather than starting individual thread, but I don't see a reason for that since each thread will always be running.

你看到这个实现什么问题？这将是多么可靠是离开线程总是以这种方式运行，我能做些什么，以确保每个线程始终运行？

Do you see any problems with this implementation? How reliable would it be to leave threads always running in this fashion and what can I do to make sure that each thread is always running?

推荐答案

与现有的解决方案的一个问题是，你打电话给你的 RunWorker 在发射后不管的方式，尽管上一个新的线程（即新的Thread（RunWorker）。开始（））。

One problem with your existing solution is that you call your RunWorker in a fire-and-forget manner, albeit on a new thread (i.e., new Thread(RunWorker).Start()).

RunWorker 是异步方法，它将返回给调用者的时候执行点命中第一个等待（即等待PerformWebRequestAsync（消息））。如果 PerformWebRequestAsync 返回一个未决的任务， RunWorker 收益和新的线程，你刚开始终止。

RunWorker is an async method, it will return to the caller when the execution point hits the first await (i.e. await PerformWebRequestAsync(message)). If PerformWebRequestAsync returns a pending task, RunWorker returns and the new thread you just started terminates.

我不认为你需要一个新的线程在这里的所有，只需使用 AmazonSQSClient.ReceiveMessageAsync 和等待其结果。另一件事是，你不应该使用异步无效方法，除非你真的不在乎跟踪异步任务的状态。使用异步任务代替。

I don't think you need a new thread here at all, just use AmazonSQSClient.ReceiveMessageAsync and await its result. Another thing is that you shouldn't be using async void methods unless you really don't care about tracking the state of the asynchronous task. Use async Task instead.

您code可能是这样的：

Your code might look like this:

List<Task> _workers = new List<Task>();
CancellationTokenSource _cts = new CancellationTokenSource();

protected override void OnStart(string[] args)
{
  for (int i = 0; i < _MAX_WORKERS; i++)
  {
    _workers.Add(RunWorkerAsync(_cts.Token)); 
  }
}

public async Task RunWorkerAsync(CancellationToken token)
{
  while(true)
  {
    token.ThrowIfCancellationRequested();

    // .. get message from amazon sqs sync.. about 20ms
    var message = await sqsClient.ReceiveMessageAsync().ConfigureAwait(false);

    try
    {
       await PerformWebRequestAsync(message);
       await InsertIntoDbAsync(message);
    }
    catch(SomeExeception)
    {
       // ... log
       //continue to retry
       continue;
    }
    sqsClient.DeleteMessage();
  }
}

现在，停止所有未决的工人，你可以简单的做到这一点（从主请求分发线程）：

Now, to stop all pending workers, you could simple do this (from the main "request dispatcher" thread):

_cts.Cancel();
try
{
    Task.WaitAll(_workers.ToArray()); 
}
catch (AggregateException ex) 
{
    ex.Handle(inner => inner is OperationCanceledException);
}

请注意， ConfigureAwait（假）是可选的Windows服务，因为有最初的线程上没有同步方面，默认情况下。不过，我会继续保持这种方式使code独立于执行环境（对于那些有同步上下文的情况下）。

Note, ConfigureAwait(false) is optional for Windows Service, because there's no synchronization context on the initial thread, by default. However, I'd keep it that way to make the code independent of the execution environment (for cases where there is synchronization context).

最后，如果由于某种原因，你不能使用 ReceiveMessageAsync ，或者你需要调用其他拦截API，或者干脆办了一张CPU密集型工作的的在开始 的RunWorkerAsync 的，只要用 Task.Run 把它包（而不是包裹整个的RunWorkerAsync ）

Finally, if for some reason you cannot use ReceiveMessageAsync, or you need to call another blocking API, or simply do a piece of CPU intensive work at the beginning of RunWorkerAsync, just wrap it with Task.Run (as opposed to wrapping the whole RunWorkerAsync):

var message = await Task.Run(
    () => sqsClient.ReceiveMessage()).ConfigureAwait(false);

这篇关于在Windows服务始终运行的线程的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！