.net - 如何平衡和并行处理多个顺序数据流

Question

我有一个必须尽快处理的数据流。单个流包含来自多达 200 个源的数据。并非所有来源都产生相同数量的数据，并且速率可能会有所不同。

作为最初的尝试，我决定创建 10 个(基于服务器规范，双四核)长时间运行的任务。每个任务将从 BlockCollection 中读取。在开始之前，我创建了一个映射，以便在入站流中接收到数据时，我知道要将源数据添加到哪个 BlockingCollection。

我认为，问题在于我不知道哪个来源会产生最多的数据，事实上这可能会随着时间的推移而改变，我看到一些集合非常空，而另一些则接收更多更新。

如果我有 8 个可用的硬件线程并且我创建了大约 10 个队列并且任务没有绑定(bind)到一个线程(再次不确定 TaskCreationOptions.LongRunning 是否属实)，那么即使一个队列不忙，另一个忙队列不能使用备用线程，因为理论上我最终可能会乱序处理一段数据。

我是否会更好地为每个源创建一个任务和阻塞集合，然后 TPL 可以充分利用可用线程，因为数据是最隔离的？

我的另一个选择是以某种方式锻炼过去的统计数据和各种外部/人类信息，如何最好地在一组有限的 BlockingCollections/Tasks 中传播源，然后随着时间的推移调整映射。

我希望我已经很好地解释了我的情况。

我正在使用一个封装 BlockingCollection and Task 的类

我可以将其可视化为 40 多个交错流，如果同时处理拆分(只要每个流保持其自己的序列)，但流比可用的硬件线程多得多。

编辑 - 尝试澄清我的查询

尝试澄清我在寻找什么。我目前正在有效地将源分成子组，并为每个组分配它自己的队列。我的问题真的是:要创建多少组？如果我有 200 个源，我是否应该创建 200 个组(然后是 200 个任务和阻塞集合)，然后让 TPL 像疯子一样到处乱跑，在每个任务获得 CPU 时间时分配线程。还是我最好为每个底层硬件线程分配 1 个组？

Answer 1

我个人会在这里利用 TPL 数据流并定义一个 ActionBlock<T>代表您的工作并链接 BufferBlock<T> “在”它的前面，以防止各种生产者过度饱和。然后你所做的就是发布到BufferBlock<T>来自您的各种来源(生产者)，并确保您已相应地对您的 block 选项(BoundedCapacity、MaxDegreeOfParallelism、MaxMessagesPerTask 等)进行了负载测试/配置，并让 TPL Dataflow 发挥其魔力。将所有繁重的工作从您手中拿走。