一起使用 BlockingCollection<T> 和 TPL 数据流时的死锁答案

【问题标题】：Deadlocks when using BlockingCollection<T> and TPL dataflow together一起使用 BlockingCollection<T> 和 TPL 数据流时的死锁
【发布时间】：2014-05-27 18:32:48
【问题描述】：

我编写了一个示例测试来复制该问题。这不是我的实际代码，我试图写一个小复制。如果将边界容量增加到迭代次数，有效地使其没有边界，则不会死锁，如果将最大并行度设置为像 1 这样的小数字，则不会死锁。

再一次，我知道下面的代码不是很好，但我实际找到的代码要大得多且难以理解。基本上有一个与远程资源的连接的阻塞对象池，并且流中的几个块使用了该连接。

关于如何解决这个问题的任何想法？乍一看，这似乎是数据流的问题。当我停下来查看线程时，我看到许多线程在 Add 时被阻塞，0 个线程在 take 时被阻塞。 addBlocks 出站队列中有几个项目尚未传播到 takeblock，因此它被卡住或死锁。

    var blockingCollection = new BlockingCollection<int>(10000);

    var takeBlock = new ActionBlock<int>((i) =>
    {
        int j = blockingCollection.Take();

    }, new ExecutionDataflowBlockOptions()
           {
              MaxDegreeOfParallelism = 20,
              SingleProducerConstrained = true
           });

    var addBlock = new TransformBlock<int, int>((i) => 
    {
        blockingCollection.Add(i);
        return i;

    }, new ExecutionDataflowBlockOptions()
           {
              MaxDegreeOfParallelism = 20
           });

    addBlock.LinkTo(takeBlock, new DataflowLinkOptions()
          {
             PropagateCompletion = true
          });

    for (int i = 0; i < 100000; i++)
    {
        addBlock.Post(i);
    }

    addBlock.Complete();
    await addBlock.Completion;
    await takeBlock.Completion;

【问题讨论】：

你能不能稍微格式化一下，很难破译
我更新了代码的格式，好些了吗？
感谢您编写可执行的重现！

标签： c# multithreading task-parallel-library tpl-dataflow blockingcollection

【解决方案1】：

TPL 数据流不适合与阻塞很多的代码一起使用，我认为这个问题源于此。

我无法弄清楚到底发生了什么，但我认为解决方案是使用非阻塞集合。方便的是，Dataflow 以BufferBlock 的形式为您提供了一个。这样，您的代码将如下所示：

var bufferBlock = new BufferBlock<int>(
    new DataflowBlockOptions { BoundedCapacity = 10000 });

var takeBlock = new ActionBlock<int>(
    async i =>
    {
        int j = await bufferBlock.ReceiveAsync();
    }, new ExecutionDataflowBlockOptions
    {
        MaxDegreeOfParallelism = 20,
        SingleProducerConstrained = true
    });

var addBlock = new TransformBlock<int, int>(
    async i =>
    {
        await bufferBlock.SendAsync(i);
        return i;
    }, new ExecutionDataflowBlockOptions
    {
        MaxDegreeOfParallelism = 20
    });

虽然我发现您的代码的整个设计很可疑。如果您想与块的正常结果一起发送一些附加数据，请将该块的输出类型更改为包含该附加数据的类型。

【讨论】：