Parallel.ForEach MaxDegreeOfParallelism 随“块化”加剧的异常行为

问题根因 #

你遇到的现象是两个典型的误用导致的：

1. Parallel.ForEach不支持异步委托：你传入的async report => { ... }实际是async void签名，Parallel不会等待异步操作完成，只会认为任务启动就处理完毕，不仅会导致并发数远超你设置的MaxDegreeOfParallelism=8，还会出现异常无法捕获、执行状态混乱的问题。
2. Parallel.ForEach默认分区策略不适配消费队列场景：Parallel默认采用分块预取策略，会一次性从枚举中加载一批数据到本地缓存再分配给线程处理，这个设计是为了降低CPU密集型任务的调度开销，但对BlockingCollection的消费枚举来说，就会导致大量消息卡在分区器缓存里，只有积压到块大小阈值才会被批量处理，就是你观察到的块大小按2的倍数增长的现象。

Parallel本身是为CPU密集型同步任务设计的，完全不适合你的IO密集型异步处理场景。

推荐方案1：TPL Dataflow（最优适配） #

这是微软官方为异步并行生产者消费者场景设计的类库，天然支持异步执行、严格并发度控制、无预取分区，完全匹配你的需求。

1. 先安装NuGet包：System.Threading.Tasks.Dataflow
2. 改造你的服务代码：

public sealed class GpsReportService : IHostedService
{
    private readonly GpsReportBuffer _buffer;
    private readonly Config _config;
    private readonly GpsReportHandler _handler;
    private readonly ILogger _logger;
    private CancellationTokenSource? _cts;
    private Task? _processTask;

    public GpsReportService(
        GpsReportBuffer buffer,
        Config config,
        GpsReportHandler handler,
        ILogger<GpsReportService> logger)
    {
        _buffer = buffer;
        _config = config;
        _handler = handler;
        _logger = logger;
    }

    public Task StartAsync(CancellationToken cancellationToken)
    {
        _logger.LogInformation("GPS Report Service => starting");
        _cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);

        // 配置处理块
        var blockOptions = new ExecutionDataflowBlockOptions
        {
            MaxDegreeOfParallelism = 8, // 严格控制并行度
            CancellationToken = _cts.Token,
            BoundedCapacity = 200 // 可选，限制块缓存大小，避免内存溢出
        };

        // 定义处理逻辑
        var processorBlock = new ActionBlock<GpsReport>(async report =>
        {
            try
            {
                await _handler.ProcessAsync(report).ConfigureAwait(false);
            }
            catch (Exception e)
            {
                if (_config.IsDevelopment)
                    throw;
                _logger.LogError(e, "GPS Report Service：处理报告失败");
            }
        }, blockOptions);

        // 启动消费任务，将缓冲区数据喂给处理块
        _processTask = Task.Run(async () =>
        {
            foreach (var report in _buffer.GetConsumingEnumerable(_cts.Token))
            {
                await processorBlock.SendAsync(report, _cts.Token).ConfigureAwait(false);
            }
            processorBlock.Complete();
            // 等待所有剩余报告处理完成
            await processorBlock.Completion.ConfigureAwait(false);
        }, _cts.Token);

        _logger.LogInformation("GPS Report Service => started");
        return Task.CompletedTask;
    }

    public async Task StopAsync(CancellationToken cancellationToken)
    {
        _logger.LogInformation("GPS Report Service => stopping");
        _buffer.CompleteAdding();
        try
        {
            if (_processTask != null)
                await _processTask.WaitAsync(cancellationToken).ConfigureAwait(false);
        }
        catch (OperationCanceledException)
        {
            // 正常停机取消，忽略
        }
        finally
        {
            _cts?.Dispose();
        }
        _logger.LogInformation("GPS Report Service => stopped");
    }
}

该方案完全解决分块预取问题，报告进入缓冲区就会被立即处理，停机时还会等待已接收的报告全部处理完成，降低数据丢失风险。

方案2：SemaphoreSlim手动控制并发 #

如果不想引入额外NuGet包，可以用信号量自己控制并行度，实现也很简单：

public sealed class GpsReportService : IHostedService
{
    private readonly GpsReportBuffer _buffer;
    private readonly Config _config;
    private readonly GpsReportHandler _handler;
    private readonly ILogger _logger;
    private CancellationTokenSource? _cts;
    private Task? _processTask;
    private readonly SemaphoreSlim _concurrencySemaphore = new SemaphoreSlim(8, 8); // 并行度设为8

    public GpsReportService(
        GpsReportBuffer buffer,
        Config config,
        GpsReportHandler handler,
        ILogger<GpsReportService> logger)
    {
        _buffer = buffer;
        _config = config;
        _handler = handler;
        _logger = logger;
    }

    public Task StartAsync(CancellationToken cancellationToken)
    {
        _logger.LogInformation("GPS Report Service => starting");
        _cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
        _processTask = ProcessAsync(_cts.Token);
        _logger.LogInformation("GPS Report Service => started");
        return Task.CompletedTask;
    }

    private async Task ProcessAsync(CancellationToken cancellationToken)
    {
        var runningTasks = new List<Task>();
        try
        {
            foreach (var report in _buffer.GetConsumingEnumerable(cancellationToken))
            {
                await _concurrencySemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
                var task = Task.Run(async () =>
                {
                    try
                    {
                        await _handler.ProcessAsync(report).ConfigureAwait(false);
                    }
                    catch (Exception e)
                    {
                        if (_config.IsDevelopment)
                            throw;
                        _logger.LogError(e, "GPS Report Service：处理报告失败");
                    }
                    finally
                    {
                        _concurrencySemaphore.Release();
                    }
                }, cancellationToken);
                runningTasks.Add(task);
                // 定期清理已完成的任务，避免内存泄漏
                if (runningTasks.Count > 100)
                    runningTasks.RemoveAll(t => t.IsCompleted);
            }
            // 缓冲区结束后等待所有剩余任务处理完成
            await Task.WhenAll(runningTasks).ConfigureAwait(false);
        }
        catch (OperationCanceledException)
        {
            // 正常取消忽略
        }
    }

    public async Task StopAsync(CancellationToken cancellationToken)
    {
        _logger.LogInformation("GPS Report Service => stopping");
        _buffer.CompleteAdding();
        try
        {
            if (_processTask != null)
                await _processTask.WaitAsync(cancellationToken).ConfigureAwait(false);
        }
        catch (OperationCanceledException)
        {
            // 正常取消忽略
        }
        finally
        {
            _cts?.Dispose();
            _concurrencySemaphore.Dispose();
        }
        _logger.LogInformation("GPS Report Service => stopped");
    }
}

额外问题解答 #

你注释中问的Task.Run(Process, cancellationToken).ConfigureAwait(false);这里的ConfigureAwait完全没用，因为你没有await这个Task返回值，写不写效果都一样。如果要避免内存泄漏和正确控制生命周期，必须把这个Task存起来，停机时等待执行完成，就是上面示例的写法。

如果要彻底解决重启丢数据的问题，可以把内存缓冲区替换为持久化队列（比如RabbitMQ、Redis List等），服务重启后可以继续消费未处理的报告。

内容的提问来源于stack exchange，提问作者Gup3rSuR4c