我使用 TPL 数据流 API 编写了一个小型管道，它从多个线程接收数据并对其执行处理。

设置 1

当我将其配置为使用 MaxDegreeOfParallelism = Environment.ProcessorCount 时(在我的例子中是 8)对于每个 block ，我注意到它填满了多个线程中的缓冲区并且处理第二个 block 直到所有线程都接收到 +- 1700 个元素才开始。你可以在行动中看到这个 here .

设置 2

当我设置 MaxDegreeOfParallelism = 1然后我注意到所有元素都在单个线程上接收，并且在接收到 +- 40 个元素后已经开始处理发送。 Data here .

设置 3

当我设置 MaxDegreeOfParallelism = 1并且我在发送每个输入之前引入了 1000 毫秒的延迟，我注意到元素在收到后立即发送，并且每个收到的元素都放在一个单独的线程中。资料here .

到目前为止设置。我的问题如下:

1. 当我比较设置 1 和 2 时，我注意到与并行相比，串行完成时处理元素的启动速度要快得多(即使考虑到并行的线程数是原来的 8 倍)。是什么导致了这种差异？

2. 由于这将在 ASP.NET 环境中运行，我不想生成不必要的线程，因为它们都来自单个线程池。如设置 3 所示，即使只有少量数据，它仍会分布在多个线程上。这也令人惊讶，因为从设置 1 开始，我假设数据按顺序分布在线程上(注意前 50 个元素如何全部进入线程 16)。我能否确保它仅按需创建新线程？

3. 还有一个概念叫做 BufferBlock<T> .如果TransformBlock<T>已经对输入进行了排队，将我的管道 ( ReceiveElement ) 中的第一步交换为 BufferBlock 的实际区别是什么？ ?

class Program
{
    static void Main(string[] args)
    {
        var dataflowProcessor = new DataflowProcessor<string>();
        var amountOfTasks = 5;
        var tasks = new Task[amountOfTasks];

        for (var i = 0; i < amountOfTasks; i++)
        {
            tasks[i] = SpawnThread(dataflowProcessor, $"Task {i + 1}");
        }

        foreach (var task in tasks)
        {
            task.Start();
        }

        Task.WaitAll(tasks);
        Console.WriteLine("Finished feeding threads"); // Needs to use async main
        Console.Read();
    }

    private static Task SpawnThread(DataflowProcessor<string> dataflowProcessor, string taskName)
    {
        return new Task(async () =>
        {
            await FeedData(dataflowProcessor, taskName);
        });
    }

    private static async Task FeedData(DataflowProcessor<string> dataflowProcessor, string threadName)
    {
        foreach (var i in Enumerable.Range(0, short.MaxValue))
        {
            await Task.Delay(1000); // Only used for the delayedSerialProcessing test
            dataflowProcessor.Process($"Thread name: {threadName}\t Thread ID:{Thread.CurrentThread.ManagedThreadId}\t Value:{i}");
        }
    }
}


public class DataflowProcessor<T>
{
    private static readonly ExecutionDataflowBlockOptions ExecutionOptions = new ExecutionDataflowBlockOptions
    {
        MaxDegreeOfParallelism = Environment.ProcessorCount
    };

    private static readonly TransformBlock<T, T> ReceiveElement = new TransformBlock<T, T>(element =>
    {
        Console.WriteLine($"Processing received element in thread {Thread.CurrentThread.ManagedThreadId}");
        return element;
    }, ExecutionOptions);

    private static readonly ActionBlock<T> SendElement = new ActionBlock<T>(element =>
    {
        Console.WriteLine($"Processing sent element in thread {Thread.CurrentThread.ManagedThreadId}");
        Console.WriteLine(element);
    }, ExecutionOptions);

    static DataflowProcessor()
    {
        ReceiveElement.LinkTo(SendElement);

        ReceiveElement.Completion.ContinueWith(x =>
        {
            if (x.IsFaulted)
            {
                ((IDataflowBlock) ReceiveElement).Fault(x.Exception);
            }
            else
            {
                ReceiveElement.Complete();
            }
        });
    }


    public void Process(T newElement)
    {      
        ReceiveElement.Post(newElement);
    }
}

最佳答案

在将解决方案部署到 ASP.NET 环境之前，我建议您更改体系结构:IIS 可以在请求处理后暂停 ASP.NET 中的线程供自己使用，这样您的任务就可以未完成。更好的方法是创建一个单独的 Windows 服务守护进程来处理您的数据流。

现在回到 TPL 数据流。

我喜欢 TPL Dataflow 库，但它的文档真是一团糟。
我找到的唯一有用的文件是 Introduction to TPL Dataflow .

其中有一些线索可能会有所帮助，尤其是关于配置设置的线索(我建议您研究使用您自己的 TheadPool 实现您自己的 TaskScheduler实现和 MaxMessagesPerTask 选项)如果需要:

The built-in dataflow blocks are configurable, with a wealth of control provided over how and where blocks perform their work. Here are some key knobs available to the developer, all of which are exposed through the DataflowBlockOptions class and its derived types (ExecutionDataflowBlockOptions and GroupingDataflowBlockOptions), instances of which may be provided to blocks at construction time.

• TaskScheduler 自定义，如@i3arnon 所述:

  By default, dataflow blocks schedule work to TaskScheduler.Default, which targets the internal workings of the .NET ThreadPool.

• 最大并行度

  It defaults to 1, meaning only one thing may happen in a block at a time. If set to a value higher than 1, that number of messages may be processed concurrently by the block. If set to DataflowBlockOptions.Unbounded (-1), any number of messages may be processed concurrently, with the maximum automatically managed by the underlying scheduler targeted by the dataflow block. Note that MaxDegreeOfParallelism is a maximum, not a requirement.

• MaxMessagesPerTask

  TPL Dataflow is focused on both efficiency and control. Where there are necessary trade-offs between the two, the system strives to provide a quality default but also enable the developer to customize behavior according to a particular situation. One such example is the trade-off between performance and fairness. By default, dataflow blocks try to minimize the number of task objects that are necessary to process all of their data. This provides for very efficient execution; as long as a block has data available to be processed, that block’s tasks will remain to process the available data, only retiring when no more data is available (until data is available again, at which point more tasks will be spun up). However, this can lead to problems of fairness. If the system is currently saturated processing data from a given set of blocks, and then data arrives at other blocks, those latter blocks will either need to wait for the first blocks to finish processing before they’re able to begin, or alternatively risk oversubscribing the system. This may or may not be the correct behavior for a given situation. To address this, the MaxMessagesPerTask option exists. It defaults to DataflowBlockOptions.Unbounded (-1), meaning that there is no maximum. However, if set to a positive number, that number will represent the maximum number of messages a given block may use a single task to process. Once that limit is reached, the block must retire the task and replace it with a replica to continue processing. These replicas are treated fairly with regards to all other tasks scheduled to the scheduler, allowing blocks to achieve a modicum of fairness between them. In the extreme, if MaxMessagesPerTask is set to 1, a single task will be used per message, achieving ultimate fairness at the potential expense of more tasks than may otherwise have been necessary.

• MaxNumberOfGroups

  The grouping blocks are capable of tracking how many groups they’ve produced, and automatically complete themselves (declining further offered messages) after that number of groups has been generated. By default, the number of groups is DataflowBlockOptions.Unbounded (-1), but it may be explicitly set to a value greater than one.

• 取消 token

  This token is monitored during the dataflow block’s lifetime. If a cancellation request arrives prior to the block’s completion, the block will cease operation as politely and quickly as possible.

• 贪心

  By default, target blocks are greedy and want all data offered to them.

• 有限容量

  This is the limit on the number of items the block may be storing and have in flight at any one time.

关于c# - 如何确保数据流 block 仅按需创建线程？，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/37833712/