使用 NServiceBus 的多线程 CQRS

Question

我目前正在研究使用 NServiceBus 来解决以下问题。我只是想确保我不会因此而陷入困境。

我有一个基于 CQRS 架构的解决方案。本质上，我通过 NServiceBus 将一系列命令传递到端点，我执行一些处理以更改聚合根的状态，然后触发一系列事件以通知系统的其余部分更改。

我遇到的问题是，这一切都在一个线程上运行良好，我无需担心在更改状态时锁定任何给定的聚合根。

我们已经到了一个线程无法切断它的地步，我需要开始研究使用多个工作线程/进程来处理消息。

基于我可以同时处理多个聚合根的事实，有一个自然的任务分解，但是我不能同时为同一个聚合根处理多个消息，因为这会导致高争用由于接收消息的速度。

我试图避免需要锁定特定聚合根 ID 而是将聚合根分配给队列/线程/进程的情况，这将确保同步处理来自同一聚合根的所有消息。

我正在使用带有 NServiceBus 的 Pub/Sub 模型来发布事件。据我所知，问题是这些事件需要从同一个端点发布，即使该消息的处理可能会委托给另一个端点。即我需要欺骗系统认为消息是从“MyDomainQueue”发布的，而实际上该消息是在“MyDomainQueue-Worker1”上处理的。

我的计划是创建一个自定义分发器，它允许主端点将消息处理委托给工作端点。分发器将以循环方式将队列分配给任何给定的聚合根。主端点将向这些工作人员发送命令，进行处理，工作人员将回复一个事件列表。然后主端点会将事件发布回总线。

我正在寻找有关此方法的一些反馈。感觉就像我正在努力让它工作，我只是想检查是否有其他人处理过类似的问题。也许 NServiceBus 是适合这项工作的错误工具，或者我的方法可能是错误的。欢迎提供任何和所有反馈。

谢谢，

PS - 我也担心上述解决方案所需的配置量。

Answer 1

关于从不同端点发布事件，我认为您有点过于沉迷于从一个服务发布事件的概念。这仅在逻辑意义上是正确的——一个事件应该由一个逻辑服务完全拥有，但是一个逻辑服务可以由许多端点组成。

事件并没有真正从队列中发布。当您想到发布事件的队列时，您真正在谈论的是您为该事件发送订阅请求的输入队列。

因此，您可以让多个端点都发布同一个事件，只要该事件的订阅请求都发往同一个地方。

这通常是这种情况，例如，在批量与优先级​​场景中，您有两个端点处理相同的命令（然后发布相同的事件），除了一个是具有长 SLA 的批量，另一个具有一个更短的 SLA - 也许它是一个大客户，或者命令来自等待响应的实际人类用户。 QueueA 和 PriorityQueueA 都处理相同的命令并发布相同的事件，但 QueueA 处理订阅，因此两个进程都从 QueueA“发布”。

也就是说，您是否尝试过让多个线程访问聚合根？即使存在一些争用，您也可能会发现，只要进行少量重试，它可能并没有您想象的那么有争议。我在使用 NServiceBus 的生产过程中有一些相当有争议的过程，虽然我偶尔在日志中看到争用的证据是一个例外，但在重试 5 次后，我从来没有在错误队列中取得任何进展。

-- 最近，添加了二级重试功能，这进一步降低了消息最终进入错误队列的机会。

如果有这么多争用，另一种策略可能是维护当前正在操作的聚合根的内存列表，然后如果有消息进入应该“锁定”，只需调用 Bus.HandleCurrentMessageLater() 坚持该消息回到队列的末尾。