MassTransit中如何在IPublishObserver内发布消息？

我尝试在MassTransit的IPublishObserver中发布消息，目的是在队列/流中生成追踪数据。虽然可以使用数据库表存储，但我更倾向于采用队列/流的方式。

以下是我的实现代码：

using MassTransit.Kvint.Tracking.Events;

public class PublishObserverForKvintTracking : IPublishObserver
{
    public async Task PostPublish<T>(PublishContext<T> context) where T : class
    {       
        // TrackedEvent 
        var trackedEvent = new TrackedEvent();
       
        await context.Publish(trackedEvent);     
    }

    public async Task PrePublish<T>(PublishContext<T> context) where T : class
    {
        await Task.CompletedTask;
    }

    public async Task PublishFault<T>(PublishContext<T> context, Exception exception) where T : class
    {
        await Task.CompletedTask;
    }
}

但我发现PublishContext接口并未暴露Publish方法，我理解其中的原因，但想知道是否仍有办法实现消息发布？

解决方案 #

要在IPublishObserver中发布消息，你需要通过依赖注入获取IPublishEndpoint（或IBus）实例，而非直接使用当前的PublishContext：

1. 修改观察者类，注入IPublishEndpoint #

using MassTransit;
using MassTransit.Kvint.Tracking.Events;

public class PublishObserverForKvintTracking : IPublishObserver
{
    private readonly IPublishEndpoint _publishEndpoint;

    // 通过构造函数注入发布端点
    public PublishObserverForKvintTracking(IPublishEndpoint publishEndpoint)
    {
        _publishEndpoint = publishEndpoint;
    }

    public async Task PostPublish<T>(PublishContext<T> context) where T : class
    {       
        var trackedEvent = new TrackedEvent();
        // 使用注入的IPublishEndpoint发布追踪事件，复用原上下文的取消令牌
        await _publishEndpoint.Publish(trackedEvent, context.CancellationToken);     
    }

    public async Task PrePublish<T>(PublishContext<T> context) where T : class
    {
        await Task.CompletedTask;
    }

    public async Task PublishFault<T>(PublishContext<T> context, Exception exception) where T : class
    {
        await Task.CompletedTask;
    }
}

2. 注册观察者到MassTransit配置 #

在总线配置阶段，将观察者注册到容器，确保依赖能被正确注入：

services.AddMassTransit(x =>
{
    // 注册发布观察者
    x.AddPublishObserver<PublishObserverForKvintTracking>();

    // 其他总线配置（以RabbitMQ为例）
    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host("rabbitmq://localhost");
        // 其余队列/交换器配置...
    });
});

关键说明 #

• PublishContext未开放Publish方法是为了避免潜在的发布循环（比如追踪消息触发观察者再次发布），如果需要循环发布，需自行添加过滤逻辑（比如判断消息类型是否为追踪事件）。
• IPublishEndpoint是MassTransit推荐的消息发布方式，它会复用总线的连接与配置，保证消息发布的可靠性。

内容的提问来源于stack exchange，提问作者Ola Hällvall