添加到集合完成时从 BlockingCollection.TakeFromAny 优雅退出答案

【问题标题】：Graceful exit from BlockingCollection.TakeFromAny when adding to collections completed添加到集合完成时从 BlockingCollection.TakeFromAny 优雅退出
【发布时间】：2020-01-09 16:54:05
【问题描述】：

我使用BlockingCollection.TakeFromAny 在循环中等待两个BlockingCollections 中的可用项目。项目被添加到其他主题的集合中。在某个时候，我完成了向集合的添加，并且在下一次调用 TakeFromAny 时，我的代码被卡住了并且没有从中返回。

我尝试将CompleteAdding 用于这两个集合或将TakeFromAny 与CancellationToken 参数一起使用，但在这两种情况下都会出现异常：

与CompleteAdding 有一个ArgumentException 与消息

关于添加的所有集合都标记为完整。参数名称：集合

CancellationToken 显然有一个OperationCanceledException。

是否可以以某种方式设置集合，使我的代码无异常地从 TakeFromAny 中取出，并且返回值表明基础集合中不会有任何新项目？

TryTakeFromAny 不适合我的需要，因为它总是在集合当前为空但项目仍会稍后添加时返回。我想阻止，直到下一个项目可用或没有要添加的内容。

【问题讨论】：

标签： c# parallel-processing task-parallel-library blockingcollection

【解决方案1】：

TryTakeFromAny 不适合我的需要，因为它总是在集合当前为空但项目仍会稍后添加时返回。我想阻止，直到下一个项目可用或没有要添加的内容。

TryTakeFromAny() 方法具有允许您指定超时值的重载。直到有要返回的值或已超过超时期限，该方法才会返回。

对于这些重载，如果您指定 -1 毫秒的超时，该方法将无限期等待。如果任一集合调用了其CompleteAdding() 方法，它不会抛出异常，并且如果传递给它的所有集合都调用了它们的CompleteAdding() 方法，它将返回-1 作为集合索引值。

一个简单的例子来说明这个工作：

static void Main(string[] args)
{
    BlockingCollection<int> c1 = new BlockingCollection<int>(), c2 = new BlockingCollection<int>();

    Task task = Produce(c1, c2);

    while (BlockingCollection<int>.TryTakeFromAny(new[] { c1, c2 }, out int value, -1) >= 0)
    {
        Console.WriteLine($"value: {value}");
    }

    Console.WriteLine($"task.IsCompleted: {task.IsCompleted}");
    task.Wait();
}

private static async Task Produce(BlockingCollection<int> c1, BlockingCollection<int> c2)
{
    await Task.Delay(TimeSpan.FromSeconds(1));
    c1.Add(1);
    await Task.Delay(TimeSpan.FromSeconds(1));
    c1.CompleteAdding();
    await Task.Delay(TimeSpan.FromSeconds(1));
    c2.Add(2);
    await Task.Delay(TimeSpan.FromSeconds(1));
    c2.CompleteAdding();
}

【讨论】：