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

Question

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

在 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 约定)，然后展开交易所的绑定(bind)查看哪些消息类型(命名为 .NET 类型名称的交换)绑定(bind)到该交换。

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