对队列进行分区以进行并行处理答案

【问题标题】：Partition a Queue for Parallel Processing对队列进行分区以进行并行处理
【发布时间】：2017-10-11 15:04:58
【问题描述】：

我正在编写一个从数千个客户端获取数据的服务器。我想并行处理这些数据，但是每个客户端的数据必须按顺序依次处理。

例如，如果我有客户端 A、B 和 C，我将同时处理来自所有 3 个客户端的数据，但每次只处理来自每个客户端的单个项目。

我的第一次尝试是创建一种分区队列，每个分区都有一个Task 消费：ConcurrentDictionary<string, BlockingCollection<T>>，其中string 是客户端标识符。有了这个，我创建了一个Task 来从每个客户端消费，并将它们保存在字典ConcurrentDictionary<string, Task> 中。任务很简单，使用GetConsumingEnumerable：

return Task.Run(() =>
{
    foreach (var item in list.GetConsumingEnumerable())
    {
        this.action(item);
    }
});

这很有效，但最多只有大约 75 个客户端 - 此后，每秒项目数的性能迅速下降，可能是由于任务之间的争用。

有没有更好的方法来做到这一点？也许使用较小的固定数量的消费线程，可以以某种方式轮换它们从哪个分区消费？

我在 .NET Framework 本身中没有发现任何有用的东西，但我觉得有可能以某种方式利用 TPL？

编辑我不只是询问如何处理具有多个生产者和单个消费者的单个队列。另外，我只对基于 TPL 的解决方案感兴趣。我只是认为它可能会有所帮助。

【问题讨论】：

@GregoryNeal 不，这比我的问题简单得多；事实上，我在我的问题中举了一个单一消费者的例子！能否请您删除此标记，以免我的问题被未阅读该问题的其他人忽略或标记为欺骗？
@Cocowalla - 如果您指出这些是连接的客户端还是断开的客户端（厚客户端与瘦客户端），这可能会有所帮助。每个人的解决方案可能会采用截然不同的方法。
@NightOwl888 实际上，客户端正在向消息代理发送消息，而服务器正在使用它们。因此，在服务器上运行的消息代理客户端正在将传入消息添加到我要并行处理的队列中。
假设队列中的下一个项目是给客户端 A 的，但是客户端 A 已经有另一个项目正在被一个消费者处理。其他消费者应该等待该项目完成还是允许他们跳过该项目并获取下一个项目（对于另一个客户）？
@Evk 他们可以跳过它并获取另一个；要求是我们一次只为任何给定的客户端处理一条消息 - 但我们希望同时处理多条消息（针对不同的客户端）

标签： c# .net queue task-parallel-library

【解决方案1】：

我想我是因为考虑为每个客户端设置一个单独的分区而偏离了方向 - 我想出的是根据客户端 ID 的哈希码（这是一个 GUID）划分成固定数量的队列)。

这样做的好处是让我可以控制并发级别，而且实现起来也很简单。

所以我的队列定义如下： Dictionary<int, BlockingCollection<T>> queues;

数据添加如下：

public void Enqueue(T val)
{
    var bucket = val.Id.GetHashCode() % this.maxConcurrent;
    this.queues[bucket].Add(val);
}

【讨论】：

我知道您已经解决了自己的问题，但您可能对这篇文档文章感兴趣 docs.microsoft.com/en-us/dotnet/standard/parallel-programming/…
@ScottChamberlain 是的，我在研究解决方案时确实遇到了 Partitioners，但我无法弄清楚如何真正自定义如何对数据进行分区。似乎没有很多示例 - 我真的很想看到一个使用分区器以与我的解决方案类似的方式对 BlockingCollection 进行分区的解决方案？
要以您希望的方式创建自定义分区器，您只需编写一个派生自Partitioner<T> 的类，在您的GetPartitions(Int32) 方法中抛出new NotSupportedException()，让SupportsDynamicPartitions 返回是的，然后实现GetDynamicPartitions()，这样每次调用它都会返回一个特定客户端数据的IEnumerable<T>，并且如果原始可枚举尚未被处理，则永远不会给出同一客户端可枚举的新副本.