.NET Core Windows 服务中的并行任务在几秒钟后挂起

Question

我正在尝试运行 Windows 服务。该服务应使用工作对象来生成多个任务。

我在工作对象和每个任务中都使用SemaphoreSlim 来等待事件完成，如下所示：

public static IHostBuilder ConfigureServices(this IHostBuilder builder)
{
    builder.ConfigureServices((hostContext, services) =>
    {
        services.AddHostedService<WorkerService>();
        services.AddSingleton<WorkerClient>();
    });

    return builder;
}

工人服务

public WorkerService(ILogger<WorkerService> logger, WorkerClient workerClient)
{
    _logger = logger;
    _workerClient = workerClient;
    _bleClient.OnValuesReceived += _bleClient_OnValuesReceived;
}

protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
    while (!stoppingToken.IsCancellationRequested)
    {
        try
        {
            await _workerClient.Run();
        }
        catch(Exception ex)
        {
            _logger.LogCritical(ex, "Error while running worker client.");
        }

        await Task.Delay(TimeSpan.FromSeconds(_scanDelay), stoppingToken);
    }
}

工人客户端

public class WorkerClient
{
    private Scanner _scanner;
    private SemaphoreSlim _lock;

    public WorkerClient()
    {
        _lock = new SemaphoreSlim(0, 1);
        _scanner = new Scanner();
        _scanner.OnScanFinished += scanner_ScanFinished;
    }

    public async Task Run()
    {
        _scanner.Scan();
        await _lock.WaitAsync();
    }

    private void scanner_ScanFinished(object sender, string[] macs)
    {
        var tasks = new List<Task>();
        foreach(var mac in macs)
        {   
            var client = new TaskRunner(mac);
            tasks.Add(client.Run());
        }
        if(tasks.Count > 0)
        {
            try
            {
                var task = Task.WhenAll(tasks.ToArray());
                await task;
            }
            catch(Exception ex)
            {
                _logger.LogError(ex, ex.Message);
            }
        }
        _lock.Release();
    }
}

任务运行器

public class TaskRunner
{
    private SemaphoreSlim _lock;
    private Client _client;

    public TaskRunner(string mac)
    {
        _lock = new SemaphoreSlim(0, 1);
        _client = new Client(mac);
        _client.OnWorkFinished += client_WorkFinished;
    }

    public async Task Run()
    {
        _client.DoWork();
        await _lock.WaitAsync();
    }

    private void client_WorkFinished(object sender, EventArgs args)
    {
        _lock.Release();
    }
}

当我在控制台或 VS 中启动它时，整个构造运行良好。但是当我使用sc 实用程序创建服务并启动它时，它会在运行 1-2 次后挂起。

我不知道自己做错了什么，因为我对 Windows 服务和多线程非常陌生。

Answer 1

SemaphoreSlim 可能不是将事件转换为Task 的适当机制，因为它不能传播异常。 TaskCompletionSource 类是更适合此目的的机制。此外，在订阅事件时，如果我们不想收到任何进一步的通知，最好取消订阅。取消订阅是使用-= 操作符实现的。

下面是 Scanner 和 Client 类的两个扩展方法，它们允许订阅它们的特定事件以获得单个通知，并将此通知作为 Task 传播。

public static class ScannerExtensions
{
    public static Task<string[]> ScanAsync(this Scanner source)
    {
        var tcs = new TaskCompletionSource<string[]>();
        Action<object, string[]> evenHandler = null;
        evenHandler = (s, macs) =>
        {
            source.OnScanFinished -= evenHandler;
            tcs.TrySetResult(macs);
        };
        source.OnScanFinished += evenHandler;
        try
        {
            source.Scan();
        }
        catch (Exception ex)
        {
            source.OnScanFinished -= evenHandler;
            tcs.SetException(ex);
        }
        return tcs.Task;
    }
}

public static class ClientExtensions
{
    public static Task DoWorkAsync(this Client source)
    {
        var tcs = new TaskCompletionSource<object>();
        EventHandler evenHandler = null;
        evenHandler = (s, e) =>
        {
            source.OnWorkFinished -= evenHandler;
            tcs.TrySetResult(null);
        };
        source.OnWorkFinished += evenHandler;
        try
        {
            source.DoWork();
        }
        catch (Exception ex)
        {
            source.OnWorkFinished -= evenHandler;
            tcs.SetException(ex);
        }
        return tcs.Task;
    }
}

您可以使用扩展方法 Scanner.ScanAsync 和 Client.DoWorkAsync 重构您的服务的 ExecuteAsync 方法，如下所示：

private Scanner _scanner = new Scanner();

protected override async Task ExecuteAsync(CancellationToken token)
{
    while (true)
    {
        Task delayTask = Task.Delay(TimeSpan.FromSeconds(_scanDelay), token);
        try
        {
            string[] macs = await _scanner.ScanAsync();
            Task[] doWorktasks = macs.Select(mac =>
            {
                var client = new Client(mac);
                return client.DoWorkAsync();
            }).ToArray();
            await Task.WhenAll(doWorktasks);
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, ex.Message);
        }
        await delayTask;
    }
}

不确定这是否能解决您的问题，但我认为这是朝着正确方向的转变。

如果问题仍然存在，您可以尝试一次创建并等待一个 client.DoWorkAsync 任务（而不是同时启动所有任务），看看是否有什么不同。