如何在 DDD 中创建管道流程？

Question

我在使用 DDD 定义管道过程时遇到了一些困难。我的情况如下：

有 3 个有界上下文：

• 患者数据管理器：负责检索和管理整个患者的数据，如mrna、mirna、dna甲基化等。

• 注释数据管理：即对基因、mirna、蛋白质等进行注释。

• 分析器：这是一个特定的分析器，它使用一些患者的数据并提取一些有用的信息。此外，用户可以创建管道并创建 puglin，定义一些额外的步骤以使用患者数据进行处理。每一步都是独一无二的，只做一项工作，如下所示：

  管道1：任务1->任务2->任务3->任务4 管道2：任务1->任务3->任务5->任务6

每个任务都可以产生一些数据，因此任务必须管理数据，读写数据。

我想减少任务和任务与患者数据管理器上下文之间的耦合。

例如：task1 必须在患者数据管理器中获取数据，因为数据量很大。 task1 获取一些数据，对其进行处理，然后获取更多数据并再次处理。此过程会导致 task1 保存一些数据。

然后，管道中的下一个任务是 task3。 Task3 从task1 中获取全部数据。 taks3 处理数据，产生另一个数据。任务 3 保存它们。管道已完成。

注意管道是在运行之前配置的，并且每个任务都需要一些数据才能工作并生成一些数据。

我想过使用命令模式在患者数据管理器上下文中获取数据。关于管道，它有工厂来定义和配置管道，每个任务都有一个用于处理数据的命令和一个用于保存数据的命令。

我知道因为命令紧密耦合。我想过使用消息来使用编排模式，但是由于管道和患者数据管理器中的数据量以及每个任务结果的数据量，这确实很难实现。

这样做可以吗？还有更好的解决方案吗？

代码在python中。

非常感谢！

Answer 1

DDD 不会以任何方式与管道发生冲突。这个问题有点假设，但我会尽力回答。从技术上讲，您可以只拥有一个执行特定领域任务的通用任务接口。例如，请参阅下面的样板文件。

class DomainTask1:
    def __init__(self):
        pass

    def run(self):
        # do something

class DomainTask2:
    def __init__(self):
        pass

    def run(self):
        # do something else

def run_pipeline(pipeline):
    for task in pipeline:
        task.run()

pipeline = [
    DomainTask1(),
    DomainTask2()
]

run_pipeline(pipeline)