c# - 如何防止 CPU : Synchronous method calling multiple workers asynchronously & throttling using SemaphoreSlim? 的 "maxing out"

Question

我目前正在优化现有的、非常缓慢且超时的生产应用程序。 没有重写它的选项。

简而言之，它是一个 WCF 服务，当前依次调用其他 4 个“worker”WCF 服务。任何工作人员服务都不依赖于另一个工作人员的结果。 所以我们希望它一次调用它们(而不是顺序调用)。我会重申，我们没有重写它的奢侈。

优化涉及让它一次调用所有工作人员服务。这就是想到异步的地方。

我在异步编程方面的经验有限，但我已经尽可能广泛地阅读了关于我的解决方案的主题。

问题是，在测试中，它可以工作，但会耗尽我的 CPU。感谢您的帮助

以下是主要 WCF 服务中基本代码的简化版本

// The service operation belonging to main WCF Service
public void ProcessAllPendingWork()
{
    var workerTasks = new List<Task<bool>>();
    foreach(var workerService in _workerServices)
    {
        //DoWorkAsync is the worker method with the following signature:
        // Task<bool> DoWorkAsync()

        var workerTask = workerService.DoWorkAsync()
        workerTasks.Add(workerTask);
    }

    var task = Task.Run(async ()=>
    {
        await RunWorkerTasks(workerTasks);
    });
    task.Wait();


}

private async RunWorkerTasks(IEnumerable<Tast<bool>> workerTasks)
{
    using(var semaphore = new SemaphoreSlim(initialCount:3))
    {

        foreach (var workerTask in workerTasks)
        {
            await semaphore.WaitAsync();
            try
            {
                await workerTask;
            }
            catch (System.Exception)
            {
                //assume 'Log' is a predefined logging service
                Log.Error(ex);
            }
        }
    }
} 

我读过的内容:

Multiple ways how to limit parallel tasks processing

How to limit the amount of concurrent async I/O operations?

Approaches for throttling asynchronous methods in C#

Constraining Concurrent Threads in C#

Limiting Number of Concurrent Threads With SemaphoresSlim

Async WCF call with ChannelFactory and CreateChannel

Answer 1

你没有解释你想如何限制并发调用。您是想要运行 30 个并发工作任务，还是想要 30 个 WCF 调用，每个调用都同时运行其所有工作任务，或者您是否希望并发 WCF 调用每个都有自己的并发工作任务限制？鉴于您说过每个 WCF 调用只有 4 个辅助任务，并且查看您的示例代码，我假设您希望全局限制 30 个并发辅助任务。

首先，正如@mjwills 暗示的那样，您需要使用 SemaphoreSlim 来限制对 workerService.DoWorkAsync() 的调用。您的代码当前启动了所有这些，并且只试图限制您等待完成的数量。我认为这就是为什么您会最大化 CPU 的原因。启动的工作任务数保持无限。但是请注意，您还需要在持有信号量时等待工作任务，否则您只会限制创建任务的速度，而不是并发运行的数量。

其次，您要为每个 WCF 请求创建一个新的 SemaphoreSlim。因此，我的第一段提出了问题。这会限制任何事情的唯一方法是，如果你有比初始计数更多的 worker 服务，在你的样本中是 30，但你说只有 4 个 worker 。要具有“全局”限制，您需要使用单例 SemaphoreSlim。

第三，您永远不会在 SemaphoreSlim 上调用 .Release()，因此，如果您确实将其设为单例，则一旦进程启动 30 个工作线程，您的代码就会挂起。确保在 try-finally block 中执行此操作，这样如果 worker 崩溃，它仍会被释放。

下面是一些草草写的示例代码:

public async Task ProcessAllPendingWork()
{
    var workerTasks = new List<Task<bool>>();
    foreach(var workerService in _workerServices)
    {
        var workerTask = RunWorker(workerService);
        workerTasks.Add(workerTask);
    }

    await Task.WhenAll(workerTasks);
}

private async Task<bool> RunWorker(Func<bool> workerService)
{
    // use singleton semaphore.
    await _semaphore.WaitAsync();
    try
    {
        return await workerService.DoWorkAsync();
    }
    catch (System.Exception)
    {
        //assume error is a predefined logging service
        Log.Error(ex);
        return false; // ??
    }
    finally
    {
        _semaphore.Release();
    }
}