rabbitmq - 公共(public)交通 : ensure message processing order when there are different message types

Question

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

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

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

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

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

我该如何处理这种情况？ 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为不同的消息设置相同的实体名称，然后将它们发布到同一个交换，但这属于“滥用”类别，就像 Chris 在该页面上写的那样。我没试过，但你绝对可以尝试。消息 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);