当使用BlockingCollection<T>和TPL数据流结合时出现死锁

Question

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

再次说明，我知道下面的代码不太好，但我实际发现的代码要大得多，难以理解。基本上，有一个阻塞对象池连接到一个远程资源和流中的几个块使用该连接。

有什么解决方法吗？乍一看似乎是数据流的问题。当我打断点查看线程时，我看到许多线程被阻塞在添加上，而0个线程被阻塞在取走上。在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;

- Aaron Stainback

---

你能稍微格式化一下吗？这样更容易阅读。 - thumbmunkeys

我更新了代码的格式，有改善吗？ - Aaron Stainback

感谢您编写一个可执行的复现！ - usr

Answer 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
    });

虽然我发现您的代码设计有些可疑。如果您想要在正常结果之外发送一些附加数据，则需要将该块的输出类型更改为包含该附加数据的类型。

- svick