如何使用 TPL 在 C# 中编组对特定线程的调用答案

【问题标题】：How to marshal calls to specific threads in C# using TPL如何使用 TPL 在 C# 中编组对特定线程的调用
【发布时间】：2012-08-14 02:25:23
【问题描述】：

我有一种情况，我有一个 TCPClient 的轮询线程（这是离散 TCP 设备的最佳计划吗？），它聚合消息并偶尔通过触发事件来响应这些消息。如果线程被长时间阻塞，事件生产者真的不太在意，但是消费者的设计是这样的，我宁愿让它调用我为处理状态而拥有的单个工作线程上的处理器机器。

那么问题是这样的。我应该如何使用任务库最好地管理这些线程的创建、配置（线程名称、背景等）生命周期以及调用的编组？我对使用 Thread 类型显式执行此操作有些熟悉，但我的公司可能更愿意通过使用 Task 来做我们可以做的事情。

编辑：我相信我在这里需要的是基于消费者类型的 SynchronizationContext ，以确保任务是与该上下文相关的单个线程上的计划。

【问题讨论】：

标签： c# multithreading task task-parallel-library

【解决方案1】：

那么问题是这样的。我应该如何使用 Task 库最好地管理这些线程的创建、配置（线程名称、背景等）生命周期以及调用封送处理？

这听起来像是BlockingCollection<T> 的完美用例。此类专为生产者/消费者场景而设计，允许您将任何线程添加到集合中（其作用类似于线程安全队列），以及一个（或多个）线程或任务调用 blockingCollection.GetConsumingEnumerable() 来“使用”项目。

【讨论】：

这似乎是一个合理的想法，但我认为需要进行非常讨厌的重组，我可以更谨慎地使用更轻量级的东西吗？
@Firoso 也许——但如果不了解更多关于您的架构的信息，就很难知道。 ConcurrentQueue<T> 是 BC 在后台使用的 - 您可以维护自己的队列以在那里处理。除非线程具有同步上下文，否则无法直接“编组”到线程...
公元前？我在这里错过了什么吗？无论如何，我认为我需要让同步上下文发挥作用。
@Firoso BlockingCollection<T> 可用于让多个生产者线程在不使用同步上下文的情况下将项目添加到“消费”，因为它是 100% 线程安全的。
我会做一些挖掘工作并尝试实现这一点，或者只是编辑我的问题以更完整地说明我的要求，谢谢！

【解决方案2】：

您可以考虑使用TPL DataFlow，您可以在其中设置一个ActionBlock<T>，将消息从您的 TCP 线程推送到其中，然后 TPL DataFlow 将通过扩展您的硬件可以处理的操作处理来处理其余部分.您还可以通过将ActionBlock<T> 配置为MaxDegreeOfParallelism 来准确控制操作的处理量。

由于处理有时跟不上传入数据的流动，您可能需要考虑在ActionBlock<T> 前面“链接”一个BufferBlock<T>，以确保TCP 处理线程不会走得太远领先于您实际可以处理的内容。这与使用有限容量的BlockingCollection<T> 具有相同的效果。

最后，请注意，我链接到 .NET 4.5 文档是因为它最简单，但 TPL DataFlow 可用于 .NET 4.0 via a separate download。不幸的是，他们从未用它制作过 NuGet 包。

【讨论】：

我实际上经常使用这些，所以您认为我可以通过将 MaxDegreeofParallelism 设置为 1 来简单地强制处理？
你可以。也许我需要更多地了解您的调度算法。您描述的模式似乎与典型的生产者/消费者模式相反，因为您实际上关心您被分派到哪些消费者线程。像 ActionBlock 和 BlockingCollection 这样的类更多的是关于将工作分散到尽可能多的工人身上，而生产者仍然不知道有多少人/谁做这项工作。因此，也许如果您详细解释为什么需要分派到特定线程，我们可以提供比这些更好的解决方案。
这与它自身进入等待状态的协议有关，当该线程被挂起时，我希望它只处理事件。
好吧，如果您使用诸如 TPL 数据流方法之类的方法，则根本不会使用任何线程，除非有数据推送到块中。由于它们是线程池线程，因此它们当时可用于处理进程中的任何其他请求。无论是其他块还是与您的特定工作负载完全无关的东西。