流数据的 CQRS

Posted 2023-02-25

【中文标题】流数据的 CQRS

【英文标题】：CQRS for streamed data

【发布时间】：2020-06-20 05:05:22

【问题描述】：

我正在使用 CQRS 隔离，非常适用于从一个节点到远程节点的事务性命令或请求-响应。

我有一个用例，其中将向远程节点发出命令，这将产生“流”数据（很像远程命令运行，服务器在执行过程中以文本形式向我们提供更新）：

// this is sent from requesting node to remote node to initiate the stream
public class LongRunningCommand: ICommand 

    Guid Session  get; set;  // the session ID to use
    string CommandLine get; set;  // the command the remote note will run

然后，这些数据会在一段时间内以多个数据包的形式从远程节点发送到请求节点：

// this is sent from remote node to requestor in multiple updates over time
public class UpdateProgress: ICommand

    Guid Session  get; set;  // possibility to multiplex sessions
    int Sequence  get; set;  // de-dupe/resequencing out of order packets (lower QOS)
    byte[] Payload  get; set;  // the data to be passed to the application

这不是一个真正的命令，也不是一个请求-回复（因为有多个回复） - 这是一个长时间运行的会话，但我不确定这如何适合 CQRS。

订购这个的最佳方式是什么？我的请求节点是否可以具有如下所示的命令处理程序（其中UpdateProgress 是正在处理的“命令”）：

public class UpdateProgressCommandHandler : ICommandHandler<UpdateProgress>

   public async Task HandleAsync(UpdateProgress message)
   
      // resequence in handler or chained infrastructure - omitted for brevity
      var window = GetWindowForSession(message.Session);
      var updateFromServer = System.Text.Encoding.UTF8.GetString(message.Payload);
      await window.WriteLine(updateFromServer);
   

上述工作（我认为相当不错），但术语似乎有点时髦（命令名称 UpdateProgress 更像是一个事件而不是命令）。

或者我最好还是放弃命令/查询的概念，并使用完整的事件总线，如果我这样做了，我将如何处理初始请求，因为这不是一个事件，它更多的是一个命令（这将在处理事件（不是命令或查询）的事件总线上语义上没有意义。

或者我只是被命名约定所困扰？自从第一次这样做以来，请欣赏上述用例的最佳实践视图。

【参考方案1】：

我不确定我是否理解正确，需要与远程节点通信的Command 似乎不是您的域的一部分。这并不是说它不是Command，您仍然可以将其定义为Command，但不在您的Domain IMO 中。您可以在此处查看集成事件。

在不完全了解您的领域的情况下，您可以通过以下方式定义您的流程：

执行命令修改您的域（类似于Status: Pending）

将集成事件从您的 CommandHandler 引发到单独的工作器/服务总线中

单独的工作人员完成该过程，然后引发另一个集成事件

您的工作人员订阅该事件并更新您域的相关部分（例如Status: Completed）。

【讨论】：

是的，这很有效，感谢您抽出宝贵时间回复。它本身不是域功能，更多的是运行命令的应用程序层（因此没有真正的域逻辑可言）。有趣地提到了 cqrs 通道和单独的服务总线通道。理想情况下，我希望为命令和流式响应提供一个通用的通信渠道。

老实说，CQRS 分区目前运行良好，因此有一个命令说 ProcessUpdatesCommand 会不会是错误的？

如果您必须在您的域中更新您的Progress，您需要有一个Command 才能这样做。如果它在您的Domain 之外，那么它不是Domain 关注的问题，它超出了 DDD 或 CQRS。那时你怎么做并不重要。