TPL Dataflow在BoundedCapacity满时如何丢弃最早消息保留最新N条数据

BufferBlock的BoundedCapacity达到上限后只会阻塞新数据写入，不会自动淘汰旧数据；BroadcastBlock的设计本身仅保留最新1条消息，设置BoundedCapacity也无法实现保留多条最新消息的需求，你可以通过自定义封装TPL Dataflow的现有块来实现需求，以下是可直接使用的实现方案：

方案1：单消费者场景 #

适用于每条数据仅需被一个消费端读取一次的场景
封装滑动缓冲区块，核心逻辑是每次新数据写入前如果缓冲区已满，主动弹出最早的旧数据：

public class SlidingBufferBlock<T> : IPropagatorBlock<T, T>, IReceivableSourceBlock<T>
{
    private readonly BufferBlock<T> _innerBuffer;
    private readonly int _maxCapacity;
    private readonly object _lockObj = new();

    public SlidingBufferBlock(int maxCapacity)
    {
        _maxCapacity = maxCapacity;
        // 内部缓冲区不设上限，由自定义逻辑控制容量
        _innerBuffer = new BufferBlock<T>(new DataflowBlockOptions
        {
            BoundedCapacity = DataflowBlockOptions.Unbounded
        });
    }

    public DataflowMessageStatus OfferMessage(DataflowMessageHeader messageHeader, T messageValue, ISourceBlock<T>? source, bool consumeToAccept)
    {
        lock (_lockObj)
        {
            // 容量已满时先淘汰最旧的数据
            while (_innerBuffer.Count >= _maxCapacity)
            {
                _innerBuffer.TryReceive(out _);
            }
            return ((ITargetBlock<T>)_innerBuffer).OfferMessage(messageHeader, messageValue, source, consumeToAccept);
        }
    }

    // 其余接口成员直接转发给内部BufferBlock实现
    public T Receive() => _innerBuffer.Receive();
    public T Receive(TimeSpan timeout) => _innerBuffer.Receive(timeout);
    public bool TryReceive(Predicate<T>? filter, out T? item) => _innerBuffer.TryReceive(filter, out item);
    public bool TryReceiveAll(out IList<T>? items) => _innerBuffer.TryReceiveAll(out items);
    public Task Completion => _innerBuffer.Completion;
    public void Complete() => _innerBuffer.Complete();
    void IDataflowBlock.Fault(Exception exception) => ((IDataflowBlock)_innerBuffer).Fault(exception);
    public IDisposable LinkTo(ITargetBlock<T> target, DataflowLinkOptions linkOptions) => _innerBuffer.LinkTo(target, linkOptions);
    T ISourceBlock<T>.ConsumeMessage(DataflowMessageHeader messageHeader, ITargetBlock<T> target, out bool messageConsumed) => ((ISourceBlock<T>)_innerBuffer).ConsumeMessage(messageHeader, target, out messageConsumed);
    bool ISourceBlock<T>.ReserveMessage(DataflowMessageHeader messageHeader, ITargetBlock<T> target) => ((ISourceBlock<T>)_innerBuffer).ReserveMessage(messageHeader, target);
    void ISourceBlock<T>.ReleaseReservation(DataflowMessageHeader messageHeader, ITargetBlock<T> target) => ((ISourceBlock<T>)_innerBuffer).ReleaseReservation(messageHeader, target);
}

使用示例 #

// 初始化容量为200的滑动缓冲区
var slidingBuffer = new SlidingBufferBlock<Data>(200);

// 生产者写入数据
while (IsDataFlowActive)
{
    await slidingBuffer.SendAsync(rawData);
}

// 消费端随时读取数据，缓冲区始终保留最新200条
// 单次读取1条
if (slidingBuffer.TryReceive(out var latestData))
{
    // 处理单条数据
}

// 一次性读取所有当前缓冲的最新数据
if (slidingBuffer.TryReceiveAll(out var allLatestData))
{
    // 批量处理数据
}

方案2：多消费者广播场景 #

适用于每个独立消费者都需要拿到全量最新N条数据的场景
如果需要类似BroadcastBlock的广播能力，只需要在上述滑动缓冲区的下游为每个消费者挂载独立的BufferBlock，配置PropagateCompletion = true的链接选项即可。

注意事项 #

• 实现中的锁逻辑是为了避免多生产者并发写入时的竞态问题，如果你的场景只有单生产者可以移除锁进一步提升性能
• 如果需要支持数据克隆避免多消费者修改同一份实例引发冲突，可以在OfferMessage中添加和BroadcastBlock一致的克隆委托逻辑

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