TPL DataFlow 让我头疼

Question

我有一个 Windows 服务（使用 Topshelf），它托管一个 WCF 服务（使用 SimpleInjectorServiceHost 创建）并启动线程（使用 Task.Run）。

这个想法是：

• WCF服务接收命令
• 将这些命令发布到BufferBlock<T>。服务需要将此BufferBlock<T> 注入其构造函数
• 线程应该是TransformBlock<T>'s，它将执行所需的操作并返回结果
• 此结果需要通过WCF服务方法读回并返回给调用者

我刚开始使用 TPL 数据流，所以我对动作块的理解可能完全错误

我在 Windows 服务的 OnStart 方法中旋转线程

for(var i = 0, i < Environment.ProcessorCount, ++i)
{
    Task.Run(() =>
    {
        using (var scope = Bootstrapper.BeginScope())
        {
            var threadHost = Bootstrapper.GetInstance<IThreadHost>();
            threadHost.DoWork(_cancellationTokenSource.Token);
        }
    }, _cancellationTokenSource.Token);
};

Bootstrapper.BeginScope 只是抽象出 SimpleInjector 的 _container.BeginLifetimeScope() 方法

IThreadHost/ThreadHost 是一个简单的接口/类，每个命令类型都有方法。 DoWork 应该在 BufferBlock 上等待，一旦可用就处理每个项目。

WCF 服务启动使用

if (_serviceHost != null)
{
    _serviceHost.Close();
}
_serviceHost = Bootstrapper.CreateServiceHost(typeof(NodeService));
_serviceHost.Open();

Bootstrapper.CreateServiceHost 只是抽象出new SimpleInjectorServiceHost(_container, serviceType)。

问：BufferBlock<T> 是否应该是在SimpleInjector 注册的单例？

问：如何将每个threadHost的DoWork设置为TransformBlock<T>并链接到BufferBlock<T>？

问：如何在我的WCF 服务方法中获得TransformBlock<T> 的输出？ WCF 服务方法将命令写入BufferBlock<T> 然后需要以某种方式等待输出？

Answer 1

我认为你有点过于复杂了。您真的不需要为您的TransformBlock 工作人员使用线程 - TPL DataFlow 中的每个块已经有一个相应的任务。如果你愿意，你可以optimize it's usage for your needs with options。

您在解决方案中得到的是一个通用的Producer/Consumer pattern，使用 TPL DataFlow 可以更轻松地完成此操作。所以你的第一个问题的答案：

问：BufferBlock 是否应该是使用 SimpleInjector 注册的单例？

是的，BufferBlock 和 TransformBlock 都应该是单例，但这取决于第三个问题的答案

正如我已经说过的，你真的不需要TPL 和TPL DataFlow 的线程，所以第二个问题也很容易回答：

Q：如何将每个threadHost的DoWork设置为TransformBlock和link 到 BufferBlock？

您不应在解决方案中使用任何线程，并将TransformBlock 直接链接到BufferBlock

至于第三个问题，这里是有趣的部分，因为答案是：

问：如何在我的 WCF 服务方法中获取 TransformBlock 的输出？ WCF 服务方法将命令写入 BufferBlock 然后需要 以某种方式等待输出？

视情况而定。

使用 TPL DataFlow 处理数据的主要思想是它是异步的，您无法确定退出 TransformBlock 时所拥有的结果正是您的请求 - 可能还有其他数据正在转换与其他请求一起，简单地使用OutputAvailableAsync() 轮询输出对于您的解决方案来说可能是非常低效的方法。

在这种情况下，您应该创建某种字典来处理结果，以便您可以轻松获取信息，无论您的请求是否完成。在这种实现情况下，您可以返回请求的 id，并引入另一种服务方法来通过该 id 获取结果。

另一种方法是将数据流封装到一个辅助类中，该类将包含它自己的BufferBlock 和TransformBlock 副本，在这种情况下，您可以使用轮询创建async 方法，如链接示例中所示。请注意，在这种实施情况下，您实际上根本不需要 TPL 数据流。