相关疑难解决方法(0)

我对通过Post()或SendAsync()发送项目之间的区别感到困惑.我的理解是,在所有情况下,一旦项目到达数据块的输入缓冲区,控制权将返回到调用上下文,对吗？那为什么我需要SendAsync？如果我的假设不正确,那么我想,相反,如果使用数据块的整个想法是建立并发和异步环境,为什么有人会使用Post().

我当然理解技术上的差异,Post()返回一个bool,而SendAsync返回一个等待bool的任务.但是它有什么影响呢？何时返回bool(我理解是否确认该项是否放在数据块的队列中)会被延迟？我理解async/await并发框架的一般概念,但在这里并没有多大意义,因为除了bool之外,对传入项所做的任何操作的结果都不会返回给调用者,而是放在一个"out-queue"并转发到链接数据块或丢弃.

发送项目时两种方法之间是否存在性能差异？

我在我的应用程序中使用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 …

我的情况是我有一个来自外部源的BufferBlock<Stream>接收Stream,比方说文件系统或一些FTP服务器.这些文件Stream将传递到另一个块并进行处理.

唯一的问题是其中一些文件是压缩的,我想Block在中间添加一个可以在必要时解压缩文件,并Stream为每个条目创建多个输出.

但是我不想使用TransformBlockMany,因为这意味着我必须完全接收ZIP Stream并立即创建输出Stream数组.

我希望这Block能够接收ZIP Stream,开始解压缩,并Push在条目就绪时进入下一个流,因此Process Block可以在第一个文件解压缩后立即开始处理,而不是等到所有内容都解压缩.

我该怎么做呢？

我试图在 TPL 数据流中控制数据流。我有一个非常快的生产者和一个非常慢的消费者。（我的实际代码更复杂，但无论如何，这是一个非常好的模型，它重现了问题。）

当我运行它时，代码开始消耗内存，就像它已经过时一样——并且生产者上的输出队列尽可能快地填满。我真正希望看到的是生产者停止运行一段时间，直到消费者有机会要求它。从我对文档的阅读来看，这就是应该发生的事情：也就是说，我认为生产者会等到消费者有空间。

事实并非如此，很明显。如何修复它以便队列不会发疯？

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

namespace MemoryLeakTestCase
{
    class Program
    {

        static void Main(string[] args)
        {
            var CreateData = new TransformManyBlock<int, string>(ignore =>
            {
                return Enumerable.Range(0, 1000 * 1000 * 1000).Select((s,i) => "Hello, World " + i);
            });

            var ParseFile = new TransformManyBlock<string, string>(fileContent =>
            {
                Thread.Sleep(1000);
                return Enumerable.Range(0, 100).Select((sst, iii) => "Hello, " + iii);
            }, new ExecutionDataflowBlockOptions() { BoundedCapacity = 1000 }
            );

            var EndOfTheLine = new ActionBlock<object>(f …