从Windows服务调用异步方法

Question

我有一个用C#编写的Windows服务,它会定期激活后台作业.通常,在任何给定时间,几十个重度I/O绑定任务(下载大文件等)并行运行.该服务在一个相对繁忙的Web服务器上运行(目前是必需的),我认为在线程保护方面可以尽可能地使用异步API.

大部分工作都已完成.所有作业现在完全异步(利用HttpClient等),主要作业循环(使用大量Task.Delay)也是如此.剩下的就是弄清楚如何从服务的OnStart正确安全地启动主循环.实际上,这是一个备受关注的呼叫异步同步困境.以下是我到目前为止(非常简化).

在Program.cs中:

static void Main(string[] args) {
    TaskScheduler.UnobservedTaskException += (sender, e) => {
        // log & alert!
        e.SetObserved();
    };
    ServiceBase.Run(new MyService());
}

在MyService.cs中:

protected override void OnStart(string[] args) {
    _scheduler.StartLoopAsync(); // fire and forget! will this get me into trouble?
}

这StartLoopAsync是对我的呼吁.我不能简单地Wait()在返回的Task上,因为OnStart需要相对快速地返回.(作业循环需要在一个单独的线程上运行.)想到几个想法:

• 通过将该处理程序放在Main中,我是否覆盖了未被观察到的异常？
• 使用Task.Run会有什么好处Task.Run(() => _scheduler.StartLoopAsync().Wait());吗？
• _scheduler.StartLoopAsync().ConfigureAwait(false)在这里打电话会有什么好处吗？(我怀疑它,因为这里没有await.)
• 在这种情况下使用Stephen Cleary的AsyncContextThread会有什么好处吗？我还没有看到任何使用它的例子,因为我开始一个无限循环,我不知道同步回到某些上下文甚至是相关的.

Answer 1

UnobservedTaskException将被调用所有未被观察到的Task异常,因此这是一个像这样记录的好地方.但是,它并不好,因为根据您的程序逻辑,您可能会看到虚假消息; 例如,如果您Task.WhenAny然后忽略较慢的任务,则应忽略该较慢任务的任何异常,但它们会被发送到UnobservedTaskException.作为替代方案,请考虑ContinueWith在您的顶级任务(从中返回的任务)上放置一个StartLoopAsync.

StartLoopAsync假设它是正确的异步,你的呼叫看起来很好.您可以使用TaskRun(例如,Task.Run(() => _scheduler.StartLoopAsync())- 没有Wait必要),但唯一的好处是,如果StartLoopAsync它本身可以引发异常(而不是故障返回任务)或者如果它在第一个之前花了太长时间await.

ConfigureAwait(false)await正如你所推测的那样,只有在做一个时才有用.

我AsyncContextThread是为这种情况而设计的,但它的设计也非常简单.:) AsyncContextThread提供了一个主循环类似于您的调度,完成一个独立的线程TaskScheduler,TaskFactory和SynchronizationContext.但是,它很简单:它只使用一个线程,并且所有调度/上下文都指向同一个线程.我喜欢它,因为它极大地简化了线程安全问题,同时也允许并发异步操作 - 但它没有充分利用线程池,因此,例如,CPU绑定的工作会阻塞主循环(类似于UI线程场景).

在您的情况下,听起来AsyncContextThread可能会让您删除/简化您已编写的一些代码.但另一方面,它不像你的解决方案那样多线程.

Answer 2

本身不是答案,但在发布此问题一年后,我们将此服务移至Azure云服务.我发现Azure SDK的Worker Role模板是从同步中正确调用async,提供取消支持,处理异常等等的一个很好的例子.它与Windows服务并不完全相同,后者没有提供与Run方法等效的方法(你需要开始工作OnStart并立即返回),但是对于它的价值,这里是:

public class WorkerRole : RoleEntryPoint
{
    private readonly CancellationTokenSource cancellationTokenSource = new CancellationTokenSource();
    private readonly ManualResetEvent runCompleteEvent = new ManualResetEvent(false);

    public override void Run() {
        Trace.TraceInformation("WorkerRole1 is running");

        try {
            this.RunAsync(this.cancellationTokenSource.Token).Wait();
        }
        finally {
            this.runCompleteEvent.Set();
        }
    }

    public override bool OnStart() {
        // Set the maximum number of concurrent connections
        ServicePointManager.DefaultConnectionLimit = 12;

        // For information on handling configuration changes
        // see the MSDN topic at http://go.microsoft.com/fwlink/?LinkId=166357.

        bool result = base.OnStart();

        Trace.TraceInformation("WorkerRole1 has been started");

        return result;
    }

    public override void OnStop() {
        Trace.TraceInformation("WorkerRole1 is stopping");

        this.cancellationTokenSource.Cancel();
        this.runCompleteEvent.WaitOne();

        base.OnStop();

        Trace.TraceInformation("WorkerRole1 has stopped");
    }

    private async Task RunAsync(CancellationToken cancellationToken) {
        // TODO: Replace the following with your own logic.
        while (!cancellationToken.IsCancellationRequested) {
            Trace.TraceInformation("Working");
            await Task.Delay(1000);
        }
    }
}