Mass Transit：确保消息类型不同时的消息处理顺序

Question

我是 Mass Transit 的新手，我想了解它是否对我的方案有帮助。 我正在构建一个使用 CQRS 事件源架构实现的示例应用程序，并且我需要一个服务总线，以便将命令堆栈创建的事件分派到查询堆栈非规范化器。

假设我们的域中有一个聚合，我们称之为 Photo，以及两个不同的域事件：PhotoUploaded 和 PhotoArchived。

在这种情况下，我们有两种不同的消息类型，默认的 Mass Transit 行为是创建两个不同的 RabbitMq 交换：一个用于 PhotoUploaded 消息类型，另一个用于 PhotoArchived 消息类型。

让我们假设有一个名为 PhotoDenormalizer 的反规范化器：该服务将成为两种消息类型的消费者，因为无论何时上传或存档照片，都必须更新照片读取模型。

给定默认的 Mass Transit 拓扑，将有两个不同的交换，因此无法保证不同类型的事件之间的消息处理顺序：我们唯一能保证的是所有相同类型的事件都会按顺序处理，但我们不能保证不同类型事件之间的处理顺序（请注意，鉴于我的示例的事件语义，处理顺序很重要）。

我该如何处理这种情况？捷运适合我的需要吗？我是否完全错过了域事件调度的重点？

Answer 1

免责声明：这不是对您问题的回答，而是一个预防性信息，为什么您不应该做您计划做的事情。

虽然 RMQ 等消息代理和 MassTransit 等消息中间件库非常适合集成，但我强烈建议不要使用消息代理进行事件溯源。我可以参考我的旧答案Event-sourcing: when (and not) should I use Message Queue? 解释其背后的原因。

您自己发现的原因之一 - 事件顺序永远无法保证。

另一个明显的原因是，从通过消息代理发布的事件构建读取模型有效地消除了重放的可能性，并构建了需要从一开始就开始处理事件的新读取模型，但他们得到的只是正在发布的事件现在。

聚合形成事务边界，因此每个命令都需要保证它在一个事务中完成。虽然 MT 支持 transaction middleware，但它只保证您获得支持它们的依赖项的事务，但不能保证消费者主体中的 context.Publish(@event)，因为 RMQ 不支持事务。您很有可能提交更改而不是在读取端获取事件。因此，事件存储的经验法则是，您应该能够从存储订阅更改流，而不是从代码中发布事件，除非这些事件是集成事件而不是域事件。

对于事件溯源，至关重要的是每个读取模型都在其投射的事件流中保留自己的检查点。消息代理不会给你这种权力，因为“检查点”实际上是你的队列，一旦消息从队列中消失——它就永远消失了，再也不会回来了。

关于实际问题：

您可以使用message topology configuration 为不同的消息设置相同的实体名称，然后将它们发布到同一个交易所，但这属于克里斯在该页面上所写的“滥用”类别。我没有尝试过，但你绝对可以尝试。消息 CLR 类型是元数据的一部分，因此不应该存在反序列化问题。

但同样，将消息放在同一个交换器中不会为您提供任何排序​​保证，除了所有消息都将放在一个队列中以供消费服务使用。

您必须至少根据您的聚合 ID 设置分区过滤器，以防止并行处理同一聚合的多条消息。顺便说一句，这对于集成也很有用。我们就是这样做的：

void AddHandler<T>(Func<ConsumeContext<T>, string> partition) where T : class
    => ep.Handler<T>(
        c => appService.Handle(c, aggregateStore), 
        hc => hc.UsePartitioner(8, partition));

AddHandler<InternalCommands.V1.Whatever>(c => c.Message.StreamGuid);