如何在MassTransit中仅为特定消息类型启用Outbox功能？

不需要使用MultiBus，有几种轻量方案可以满足你的需求：

1. 显式指定单条消息使用Outbox #

如果你的全局Outbox是启用状态，可以在发布UploadImage时显式强制启用Outbox，而ImageDeleted则保持默认（不触发Outbox）。反之如果全局未启用Outbox，也可以仅在需要的消息发布时手动开启：

// 发布UploadImage时启用Outbox
await _publishEndpoint.Publish<UploadImage>(message, context =>
{
    context.UseOutbox();
});

// 发布ImageDeleted时直接发送，不使用Outbox
await _publishEndpoint.Publish<ImageDeleted>(message);

2. 基于消息类型的条件中间件 #

自定义一个发布过滤器，根据消息类型自动判断是否启用Outbox，无需在业务代码中重复编写配置：

public class ConditionalOutboxFilter<T> : IFilter<PublishContext<T>> where T : class
{
    public async Task Send(PublishContext<T> context, IPipe<PublishContext<T>> next)
    {
        // 仅对UploadImage类型消息启用Outbox
        if (typeof(T) == typeof(UploadImage))
            context.UseOutbox();
        
        await next.Send(context);
    }

    public void Probe(ProbeContext context)
    {
        context.CreateFilterScope("conditional-outbox");
    }
}

然后在MassTransit配置中注册该过滤器：

services.AddMassTransit(cfg =>
{
    cfg.UsingRabbitMq((ctx, rbc) =>
    {
        // 为指定消息类型添加过滤器
        rbc.Publish<UploadImage>(p => p.AddPipeSpecification(new ConditionalOutboxFilter<UploadImage>()));
        rbc.Publish<ImageDeleted>(p => {}); // 不添加Outbox相关配置
        
        // 其他总线配置
        rbc.ConfigureEndpoints(ctx);
    });
});

3. 注入不同配置的发布端点 #

通过依赖注入注册两个IPublishEndpoint实例，一个绑定Outbox逻辑，另一个直接发送消息：

// 注册带Outbox的发布端点
services.AddScoped<IPublishEndpoint>(sp =>
{
    var bus = sp.GetRequiredService<IBus>();
    var pipe = new PublishPipeConfigurator().UseOutbox().Build();
    return new PublishEndpoint(bus, pipe);
});

// 注册不带Outbox的命名发布端点
services.AddKeyedScoped<IPublishEndpoint>("NoOutbox", (sp, _) =>
{
    var bus = sp.GetRequiredService<IBus>();
    var pipe = new PublishPipeConfigurator().Build();
    return new PublishEndpoint(bus, pipe);
});

业务代码中按需注入使用：

public class ImageService
{
    private readonly IPublishEndpoint _outboxPublisher;
    private readonly IPublishEndpoint _directPublisher;

    public ImageService(IPublishEndpoint outboxPublisher, [FromKeyedServices("NoOutbox")] IPublishEndpoint directPublisher)
    {
        _outboxPublisher = outboxPublisher;
        _directPublisher = directPublisher;
    }

    public async Task HandleUpload(UploadImage message)
    {
        // 执行数据库变更操作
        await _outboxPublisher.Publish(message); // 随事务提交到Outbox
    }

    public async Task HandleDelete(ImageDeleted message)
    {
        await _directPublisher.Publish(message); // 直接发送到消息代理
    }
}

以上方案都可以避免引入MultiBus的复杂度，根据你的代码结构和业务习惯选择即可。

内容的提问来源于stack exchange，提问作者Q.Aimad