为什么我在跳过的队列中收到消息

Question

我在 fork/join 配置中有一个 saga 设置。

在 saga 上定义的事件

• FileMetadataMsg
• FileReadyMsg
• SomeOtherMsg

当文件进入单独的侦听器时，进程开始。

• 发布SagaStart(correlationId)
• 发布FileSavedToMsg(correlationId, fileLoc)
• 发布FileMetadataMsg(correlationId, metadata)
• 发布FileReadyMsg(correlationId, fileLoc)

对文件做一些工作的下游端点

Consumer<FileSavedToMsg>

• 发布SomeOtherMsg(GotTheFileMsg.correlationId, data)

我在 saga_skipped 队列中获得了 FileSavedToMsg。我只能假设这是由于 FileSavedToMsg 上有一个correlationId，因为 saga 本身没有在其状态机中使用 FileSavedToMsg 并且没有Event<FileSavedToMsg>。

如果这就是为什么...我应该在 CorrelationId 以外的字段中传递correlationId，所以传奇看不到它吗？我在某个地方需要它，所以我可以用它标记 SomeOtherMsg。

下面是定义 saga 端点的方式

return Bus.Factory.CreateUsingRabbitMq(cfg =>
{
    var host = cfg.Host(new Uri("rabbitmq://localhost/"), h =>
            {
                h.Username("guest");
                h.Password("guest");
            });              

            cfg.ReceiveEndpoint(host, "study_saga", epCfg =>
            {
                epCfg.StateMachineSaga(machine, repository);
            });
});

以下是工作人员端点的定义方式

return Bus.Factory.CreateUsingRabbitMq(x =>
{
    var host = x.Host(new Uri("rabbitmq://localhost/"), h =>
        {
            h.Username("guest");
            h.Password("guest");
        });

        x.ReceiveEndpoint(host, "study_3d_volume_worker", c =>
        {
            c.PrefetchCount = 1;
            c.Instance(_studyCreatedMsgConsumer);
        });
 });

它们在同一台机器上运行，但在单独的控制台/Topshelf 应用程序中。

Answer 1

如果您在队列上获取的消息未被该接收端点上的消费者消费，则可能是您之前正在消费该消息类型并将其从消费者（或 saga，在您的情况下）中删除，或者您正在出于其他目的使用队列，并且它使用了该消息类型。

无论哪种方式，如果您进入 RabbitMQ 管理控制台并查找队列，则可以展开 Bindings chevron，单击以转到同名交换（这是标准的 MassTransit 约定），然后展开绑定以查看哪些消息类型（命名为 .NET 类型名称的交换）绑定到该交换。

如果您看到一个没有被端点消耗的，那就是罪魁祸首。您可以使用 UI 解绑它，之后发布的消息将不再发送到队列。