在TPL数据流中保证交付的BroadcastBlock [英] BroadcastBlock with guaranteed delivery in TPL Dataflow

问题描述

我有一种数据流，可以用几种不同的方式处理...所以我想将收到的每条消息的副本发送到多个目标，以便这些目标可以并行执行...但是，我需要在我的数据块上设置 BoundedCapacity ，因为数据流的传输速度比我的目标可以处理它们的速度快，并且有大量数据.没有 BoundedCapacity ，我很快就会用光内存.

I have a stream of data that I process in several different ways... so I would like to send a copy of each message I get to multiple targets so that these targets may execute in parallel... however, I need to set BoundedCapacity on my blocks because the data is streamed in way faster than my targets can handle them and there is a ton of data. Without BoundedCapacity I would quickly run out of memory.

但是问题是 BroadcastBlock 如果目标无法处理它，则会丢弃消息(由于 BoundedCapacity ).

However the problem is BroadcastBlock will drop messages if a target cannot handle it (due to the BoundedCapacity).

我需要的是一个 BroadcastBlock ，它不会删除消息，但实际上会拒绝其他输入，直到它可以将消息传递到每个目标然后为更多目标做好准备为止.

What I need is a BroadcastBlock that will not drop messages, but will essentially refuse additional input until it can deliver messages to each target and then is ready for more.

是否存在类似的内容，或者有人编写了以这种方式运行的自定义块吗?

Is there something like this, or has anybody written a custom block that behaves in this manner?

推荐答案

使用 ActionBlock 和 SendAsync()构建所需的内容非常简单像:

It is fairly simple to build what you're asking using ActionBlock and SendAsync(), something like:

public static ITargetBlock<T> CreateGuaranteedBroadcastBlock<T>(
    IEnumerable<ITargetBlock<T>> targets)
{
    var targetsList = targets.ToList();

    return new ActionBlock<T>(
        async item =>
        {
            foreach (var target in targetsList)
            {
                await target.SendAsync(item);
            }
        }, new ExecutionDataflowBlockOptions { BoundedCapacity = 1 });
}

这是最基本的版本，但是扩展它以支持可变的目标列表，传播完成或克隆功能应该很容易.

This is the most basic version, but extending it to support mutable list of targets, propagating completion or cloning function should be easy.

这篇关于在TPL数据流中保证交付的BroadcastBlock的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！