如何将 TPL 数据流 TranformBlock 或 ActionBlock 放在单独的文件中？

Question

我想将 TPL 数据流用于我的 .NET Core 应用程序并关注 the example from the docs.

我不想将所有逻辑都放在一个文件中，而是将每个 TransformBlock 和 ActionBlock（我还不需要其他的）分开到它们自己的文件中。 TransformBlock 将整数转换为字符串的小例子

class IntToStringTransformer : TransformBlock<int, string>
{
    public IntToStringTransformer() : base(number => number.ToString()) { }
}

还有一个小的 ActionBlock 示例将字符串写入控制台

class StringWriter : ActionBlock<string>
{
    public StringWriter() : base(Console.WriteLine) { }
}

不幸的是，这不起作用，因为块类是密封的。有没有办法可以将这些块组织到它们自己的文件中？

Answer 1

数据流步骤/块/goroutines 本质上是功能性的，最好组织成工厂函数的模块，而不是单独的类。 TPL DataFlow 管道与 F# 或任何其他语言中的函数调用管道非常相似。实际上，可以将其视为 PowerShell 管道，只是它更易于编写。

无需创建类或实现接口即可将新功能添加到该管道，您只需添加它并将输出重定向到下一个功能。

TPL 数据流块已经提供了构造管道的原语，并且只需要一个转换函数。这就是它们被密封的原因，以防止误用。

组织数据流的自然方式也类似于 F# - 创建具有执行每项工作的 函数 的库，将它们放入相关函数的模块中。这些函数是无状态的，因此它们可以像扩展方法一样轻松进入静态库。

例如，可能有一个模块用于执行批量插入或读取数据的数据库相关功能，另一个用于处理各种文件格式的导出，用于调用外部 Web 服务的单独类，另一个用于解析特定消息格式。

一个真实的例子

在过去的 7 年中，我一直在为一家在线旅行社 (OTA) 处理多个复杂的管道。其中一个调用几个 GDS（OTA 和航空公司之间的中介）来检索交易信息 - 机票问题、退款、取消等。下一步检索机票记录，详细的机票信息。最后，记录被插入到数据库中。

GDS 太大而无暇顾及标准，因此它们的“SOAP”Web 服务甚至不符合 SOAP，更不用说遵循 WS-* 标准了。所以每个 GDS 都需要一个单独的类库来调用服务并解析输出。那里还没有数据流，项目已经够复杂了

将数据写入数据库几乎总是相同的，因此有一个单独的项目，其方法采用例如IEnumerable<T> 并使用SqlBulkCopy 将其写入数据库。

但加载新数据是不够的，经常会出错，所以我需要能够加载已经存储的票证信息。

组织

为了保持理智：

• 每个管道都有自己的文件：
  • 用于加载新数据的每日管道，
  • 用于加载所有存储数据的 Reload 管道
  • “重新运行”管道以使用现有数据并再次询问任何丢失的数据。
• 静态类用于保存工作函数和单独根据配置生成 Dataflow 块的工厂方法。例如，CreateLogger(path,level) 创建一个记录特定消息的ActionBlock<Message>。
• 常见的dataflow 扩展方法 - 因为DataFlow 块遵循相同的基本模式，通过组合Func<TIn,TOut> 和记录器块很容易创建记录块。或者创建一个LinkTo 重载，将不良记录重定向到记录器或数据库。这些很常见，它们可以成为扩展方法。

如果它们在同一个文件中，则很难在不影响另一个管道的情况下编辑一个管道。此外，管道除了核心任务之外还有很多其他内容，例如：

• 日志记录
• 处理不良记录和部分结果（无法为 10 个错误停止 100K 导入）
• 错误处理（与处理不良记录不同）
• 监控 - 这个怪物在过去的 15 分钟里在做什么？ DOP=10 是否提高了性能？

不要创建父管道类。

有些步骤是常见的，所以一开始，我创建了一个父类，其中的常见步骤被重载，或者只是在子类中替换。 非常糟糕的想法。每个管道都相似但不完全相同，继承意味着修改一个步骤或一个连接可能会破坏一切。大约 1 年后，事情变得难以忍受，所以我将父类分成不同的类。

Answer 2

正如@Panagiotis 解释的那样，我认为您必须稍微抛开 OOP 思维。 您使用 DataFlow 所拥有的是构建块，您可以配置这些构建块来执行您需要的内容。我将尝试创建一个小例子来说明我的意思：

// Interface and impl. are in separate files. Actually, they could 
// even be in a different project ...
public interface IMyComplicatedTransform
{
     Task<string> TransformFunction(int input);
}

public class MyComplicatedTransform : IMyComplicatedTransform
{
     public Task<string> IMyComplicatedTransform.TransformFunction(int input)
     {
         // Some complex logic
     }
}

class DataFlowUsingClass{

     private readonly IMyComplicatedTransform myTransformer;
     private readonly TransformBlock<int , string> myTransform;
     // ... some more blocks ...

     public DataFlowUsingClass()
     {
          myTransformer = new MyComplicatedTransform(); // maybe use ctor injection?
          CreatePipeline();
     }

     private void CreatePipeline()
     {
          // create blocks
          myTransform = new TransformBlock<int, string>(myTransformer.TransformFunction);
          // ... init some more blocks

          // TODO link blocks
     }
}

我认为这是最接近您想要做的事情。

您最终得到的是一组可以独立测试的接口和实现。客户端基本上归结为“gluecode”。

编辑：正如@Panagiotis 正确指出的那样，界面甚至超级流畅。你可以不用。