MassTransit 中多个 subscriber 订阅同一 event 如何隔离？

2026-10-24支付老李

结论

通常没有必要在 Payment microservice 中分别创建 CollectFuelPayment 和 CollectGroceriesPayment 两套 consumer。

如果支付流程相同，保留一个通用的 CollectPayment consumer 即可。具体怎么处理，取决于 PaymentCollected 所表达的消息类型：

如果它表示“支付已经完成”这一业务事实，就应该作为 event 发布。所有订阅者都会收到消息，各订阅者再根据业务标识和 Saga 关联关系判断是否需要处理。
如果它只是某次 CollectPaymentCommand 的定向处理结果，就不适合广播。此时应使用 MassTransit 的 request/response，只把结果返回给请求方。
只有当 Fuel 和 Groceries 的支付语义、处理流程或数据契约确实不同时，才有必要拆分 command 和 event。不要仅仅为了避免订阅者收到消息而拆分 consumer。

首先确保每个服务使用独立队列

Groceries 和 Fuel 必须配置不同的 receive endpoint：

services.AddMassTransit(x =>
{
    x.AddSagaStateMachine<GroceriesStateMachine, GroceriesState>()
        .EntityFrameworkRepository(...);

    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.ReceiveEndpoint("groceries-payment-events", e =>
        {
            e.ConfigureSaga<GroceriesState>(context);
        });
    });
});

Fuel Service 使用另一个队列：

services.AddMassTransit(x =>
{
    x.AddSagaStateMachine<FuelStateMachine, FuelState>()
        .EntityFrameworkRepository(...);

    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.ReceiveEndpoint("fuel-payment-events", e =>
        {
            e.ConfigureSaga<FuelState>(context);
        });
    });
});

event 发布后，两个独立队列会各自收到一份消息。这是发布/订阅模型的正常行为。

不要让两个服务共同消费同一个队列，比如都使用 payment-collected。这样会形成 competing consumers，一条消息通常只会由其中一个服务取走，无法保证 Groceries 和 Fuel 都能收到。

方案一：通过 CorrelationId 关联各自的 Saga

如果 PaymentCollected 是真正的领域事件，可以在 command 和 event 中携带同一个关联标识：

public record CollectPaymentCommand(
    Guid PaymentId,
    Guid CorrelationId,
    decimal Amount,
    string Currency);

public record PaymentCollectedEvent(
    Guid PaymentId,
    Guid CorrelationId,
    DateTime CollectedAt);

Payment consumer 可以保持通用：

public sealed class CollectPaymentConsumer
    : IConsumer<CollectPaymentCommand>
{
    public async Task Consume(
        ConsumeContext<CollectPaymentCommand> context)
    {
        var command = context.Message;

        // 执行支付。实际项目中还应考虑幂等、事务和失败处理。
        await CollectPayment(command);

        await context.Publish(new PaymentCollectedEvent(
            command.PaymentId,
            command.CorrelationId,
            DateTime.UtcNow));
    }

    private static Task CollectPayment(CollectPaymentCommand command)
    {
        return Task.CompletedTask;
    }
}

Groceries Saga 只关联由自己创建的 Saga 实例：

Event(() => PaymentCollected, e =>
{
    e.CorrelateById(context =>
        context.Message.CorrelationId);
});

Fuel Saga 也使用自己的 CorrelationId 查找 Saga：

Event(() => PaymentCollected, e =>
{
    e.CorrelateById(context =>
        context.Message.CorrelationId);
});

两个 endpoint 虽然都会收到 event，但只有拥有对应 Saga 实例的服务才会更新状态。

这里有几点需要注意：

CorrelationId 应由业务流程的发起方生成，并在 command 和 event 之间原样传递。
不要用 PaymentId、订单号或其他可能在不同业务域中重复的值作为 Saga 主键，除非能够保证它全局唯一。
需要明确配置找不到 Saga 实例时的处理方式。MassTransit 版本和状态机配置不同，可用 API 也可能不同，应根据当前版本决定是丢弃、记录还是将消息送入错误队列。
如果大量无关消息都会投递到两个服务，再由 Saga 丢弃，业务处理虽然没有问题，但会增加 broker、consumer 和日志的处理成本。

方案二：在事件中加入来源信息

还可以在支付请求中加入业务来源或所有者：

public enum PaymentContext
{
    Groceries,
    Fuel
}

public record CollectPaymentCommand(
    Guid PaymentId,
    Guid CorrelationId,
    PaymentContext Context,
    decimal Amount,
    string Currency);

public record PaymentCollectedEvent(
    Guid PaymentId,
    Guid CorrelationId,
    PaymentContext Context,
    DateTime CollectedAt);

Groceries 只处理属于自己的事件：

public async Task Consume(
    ConsumeContext<PaymentCollectedEvent> context)
{
    if (context.Message.Context != PaymentContext.Groceries)
        return;

    await UpdateGroceriesRequest(context.Message.CorrelationId);
}

这种做法比较简单，但也有明确的限制：消息依然会投递到两个队列，只是在 consumer 内部进行过滤。它可以避免误处理，却不能减少无关消息的传输。

Context 也不能代替 Saga 关联。更稳妥的方式是同时检查业务来源和 CorrelationId。

如果业务来源会不断增加，不建议让 Payment 服务维护一个越来越大的枚举。可以改用稳定的业务域标识，但不要把服务名、队列名等部署细节直接写入核心领域契约。

方案三：使用 request/response 定向返回结果

如果只有发送 CollectPaymentCommand 的服务需要处理结果，那么 PaymentCollectedEvent 更接近 response，而不是需要广播的 event。

调用方可以使用 IRequestClient<T>：

public record CollectPayment(
    Guid PaymentId,
    Guid CorrelationId,
    decimal Amount,
    string Currency);

public record PaymentCollected(
    Guid PaymentId,
    Guid CorrelationId,
    DateTime CollectedAt);

public record PaymentRejected(
    Guid PaymentId,
    Guid CorrelationId,
    string Reason);

Payment consumer 直接返回处理结果：

public sealed class CollectPaymentConsumer
    : IConsumer<CollectPayment>
{
    public async Task Consume(ConsumeContext<CollectPayment> context)
    {
        var command = context.Message;

        var succeeded = await TryCollectPayment(command);

        if (!succeeded)
        {
            await context.RespondAsync(new PaymentRejected(
                command.PaymentId,
                command.CorrelationId,
                "Payment was rejected"));

            return;
        }

        await context.RespondAsync(new PaymentCollected(
            command.PaymentId,
            command.CorrelationId,
            DateTime.UtcNow));
    }

    private static Task<bool> TryCollectPayment(CollectPayment command)
    {
        return Task.FromResult(true);
    }
}

调用方发送请求，并等待指定类型的响应：

var response = await requestClient.GetResponse<
    PaymentCollected,
    PaymentRejected>(
    new CollectPayment(
        paymentId,
        correlationId,
        amount,
        currency));

if (response.Is(out Response<PaymentCollected>? collected))
{
    // 更新当前业务流程状态
}
else if (response.Is(out Response<PaymentRejected>? rejected))
{
    // 处理支付失败
}

MassTransit 会根据请求中的响应地址和请求标识，把结果路由回对应的请求方。Groceries 发起请求后，结果不会广播给 Fuel，反过来也一样。

如果 Saga 的运行时间很长，并且需要在进程重启后继续执行，就不适合简单依赖同步等待式调用。可以使用 MassTransit 状态机提供的 request 模式，或者设计持久化的回复队列。具体配置取决于 MassTransit 版本，但基本原则不变：请求携带可关联的请求标识和返回地址，Payment 再定向响应。

方案四：定向发送结果

还可以在 command 中携带一个受控的回复目标。Payment 完成处理后使用 Send，而不是 Publish。

不过，只要可以使用 RespondAsync，通常就应优先使用它。MassTransit 已经处理了响应地址和请求标识。自行在业务消息中保存任意 URI，会带来一些额外问题：

业务契约会与 broker 地址耦合。
队列重命名会影响消息契约。
必须校验目标地址，不能无条件向外部提供的任意地址发送消息。
重试、超时和请求关联都需要自行设计。

如果处理结果只属于某个请求方，request/response 通常比手工传递队列地址更清楚。

方案五：使用 broker 级过滤或路由

部分 transport 支持在 broker 侧过滤消息，例如：

RabbitMQ 可以使用 exchange、binding 和 routing key。
Azure Service Bus 可以使用 subscription rule/filter。
其他 transport 可能提供不同的路由方式。

可以使用 groceries、fuel 等 routing key，只把消息绑定到对应的 endpoint。这样可以减少无关消息的实际投递，但会引入 transport-specific 配置，消息拓扑也会承担一部分业务路由职责。

MassTransit 的具体配置 API 会因 transport 和版本而异，因此应参考当前 MassTransit 版本及对应 transport 的文档。不要把某一种 broker 的配置示例直接套用到另一种 broker。

即使已经使用 broker 过滤，消息中仍应保留 CorrelationId，consumer 也不能只依靠 routing key 来保证业务处理正确。

什么时候应该拆分 command 和 event

在以下情况下，拆分消息契约是合理的：

public record CollectGroceriesPayment(...);
public record GroceriesPaymentCollected(...);

public record CollectFuelPayment(...);
public record FuelPaymentCollected(...);

Groceries 和 Fuel 使用不同的支付规则。
两者的必填数据、权限或验证规则不同。
两者有不同的失败状态和补偿流程。
两类支付需要独立演进或进行版本管理。
GroceriesPaymentCollected 和 FuelPaymentCollected 本身表示不同的领域事实。

即使消息契约不同，也不一定要复制两套支付实现。两个 consumer 可以调用同一个应用服务：

public sealed class CollectGroceriesPaymentConsumer
    : IConsumer<CollectGroceriesPayment>
{
    private readonly IPaymentCollector _collector;

    public CollectGroceriesPaymentConsumer(IPaymentCollector collector)
    {
        _collector = collector;
    }

    public async Task Consume(
        ConsumeContext<CollectGroceriesPayment> context)
    {
        await _collector.Collect(...);

        await context.Publish(
            new GroceriesPaymentCollected(...));
    }
}

Fuel consumer 也可以复用 IPaymentCollector。这样既能让消息契约保持各自的业务语义，也不会重复实现支付逻辑。