相关疑难解决方法(0)

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

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

我需要的是一个不会丢弃消息的BroadcastBlock,但实质上会拒绝其他输入,直到它可以向每个目标传递消息然后准备好更多.

有这样的东西,还是有人编写了一个以这种方式运行的自定义块？

我在我的应用程序中使用TPL Dataflow实现了生产者/消费者模式.我有大数据流网格,其中有大约40个块.网格中有两个主要功能部分:生产者部分和消费者部分.生产者应该继续为消费者提供大量工作,而消费者有时会缓慢地处理传入的工作.当消费者忙于一些指定数量的工作项时,我想暂停生产者.否则,该应用程序会占用大量内存/ CPU,并且行为不可持续.

我制作了演示应用程序以演示此问题:

using System;
using System.Linq;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;

namespace DataflowTest
{
    class Program
    {
        static void Main(string[] args)
        {
            var options = new ExecutionDataflowBlockOptions
            {
                MaxDegreeOfParallelism = 4,
                EnsureOrdered = false
            };

            var boundedOptions = new ExecutionDataflowBlockOptions
            {
                MaxDegreeOfParallelism = 4,
                EnsureOrdered = false,
                BoundedCapacity = 5
            };

            var bufferBlock = new BufferBlock<int>(boundedOptions);
            var producerBlock = new TransformBlock<int, int>(x => x + 1, options);
            var broadcastBlock = new BroadcastBlock<int>(x => x, options);

            var consumerBlock = new …