如何在C#多字面量搜索TPL DataFlow流水线中动态批处理小工作项

核心逻辑 #

自定义TPL DataFlow块替代固定批量大小的BatchBlock，核心逻辑为累计流入工作项的总工作量，达到阈值时输出一批，流结束时将剩余未达阈值的工作项合并为最后一批输出，允许不同文件的小工作项聚合为同一批次，同时保留大文件拆分出的独立大工作项单独处理的规则。

具体实现步骤 #

• 首先扩展你的工作项数据结构，预存每个工作项的工作量值（即文件行数 × 待搜索字面量总数的计算结果），避免后续重复计算
• 实现自定义动态批量块，代码示例如下：

public class DynamicBatchBlock<T> : IPropagatorBlock<T, T[]>
{
    private readonly long _workloadThreshold;
    private readonly BufferBlock<T[]> _outputBuffer;
    private readonly List<T> _currentBatch = new();
    private long _currentTotalWorkload = 0;
    private readonly Func<T, long> _getWorkloadFunc;

    public DynamicBatchBlock(long workloadThreshold, Func<T, long> getWorkloadFunc, DataflowBlockOptions? options = null)
    {
        _workloadThreshold = workloadThreshold;
        _getWorkloadFunc = getWorkloadFunc;
        options ??= new DataflowBlockOptions();
        _outputBuffer = new BufferBlock<T[]>(options);

        var inputProcessor = new ActionBlock<T>(item =>
        {
            var itemWorkload = _getWorkloadFunc(item);
            // 单个工作项直接超过阈值，单独成批
            if (itemWorkload >= _workloadThreshold)
            {
                if (_currentBatch.Any())
                {
                    _outputBuffer.Post(_currentBatch.ToArray());
                    _currentBatch.Clear();
                    _currentTotalWorkload = 0;
                }
                _outputBuffer.Post(new[] { item });
                return;
            }
            // 累加后超过阈值则输出当前批次
            if (_currentTotalWorkload + itemWorkload > _workloadThreshold)
            {
                _outputBuffer.Post(_currentBatch.ToArray());
                _currentBatch.Clear();
                _currentTotalWorkload = 0;
            }
            _currentBatch.Add(item);
            _currentTotalWorkload += itemWorkload;
        }, new ExecutionDataflowBlockOptions
        {
            BoundedCapacity = options.BoundedCapacity,
            CancellationToken = options.CancellationToken
        });

        // 输入结束时输出剩余批次
        inputProcessor.Completion.ContinueWith(t =>
        {
            if (_currentBatch.Any())
            {
                _outputBuffer.Post(_currentBatch.ToArray());
            }
            _outputBuffer.Complete();
        }, TaskContinuationOptions.OnlyOnRanToCompletion);

        // 异常传播逻辑
        inputProcessor.Completion.ContinueWith(t =>
        {
            if (t.IsFaulted) ((ITargetBlock<T[]>)_outputBuffer).Fault(t.Exception!);
        }, TaskContinuationOptions.OnlyOnFaulted);

        Source = _outputBuffer;
        Target = inputProcessor;
    }

    public ITargetBlock<T> Target { get; }
    public ISourceBlock<T[]> Source { get; }
    public Task Completion => Source.Completion;
    public void Complete() => Target.Complete();
    public void Fault(Exception exception) => Target.Fault(exception);
    public T[] ConsumeMessage(DataflowMessageHeader messageHeader, ITargetBlock<T[]> target, out bool messageConsumed)
        => Source.ConsumeMessage(messageHeader, target, out messageConsumed);
    public bool ReserveMessage(DataflowMessageHeader messageHeader, ITargetBlock<T[]> target)
        => Source.ReserveMessage(messageHeader, target);
    public void ReleaseReservation(DataflowMessageHeader messageHeader, ITargetBlock<T[]> target)
        => Source.ReleaseReservation(messageHeader, target);
    public DataflowMessageStatus OfferMessage(DataflowMessageHeader messageHeader, T messageValue, ISourceBlock<T>? source, bool consumeToAccept)
        => Target.OfferMessage(messageHeader, messageValue, source, consumeToAccept);
}

• 修改现有流水线：将原本输出单个工作项的节点，先连接到上述自定义的DynamicBatchBlock，配置阈值为100万，工作量计算逻辑直接读取工作项预存的工作量值即可，后续再接处理批量工作项的执行块，批量块会自动将小工作项聚合为符合阈值要求的批次，大幅减少调度开销。

可选优化点 #

• 可增加最大等待时长参数，避免长时间没有足够工作项凑够阈值导致的任务延迟，比如设置最长500ms就算未达阈值也输出当前批次，你的离线搜索场景如果无实时性要求可忽略该配置
• 批量处理执行块可将同批次的多个工作项合并为单次任务执行，进一步降低TPL DataFlow的调度开销，性能收益会更明显

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