MassTransit SendFilter无法阻止消息发送，如何解决？

我在多租户应用中使用MassTransit与RabbitMQ，通过IFilter实现自定义Outbox模式，只希望把消息保存到数据库，却发现消息还是会被发送到RabbitMQ。我原本以为只要不调用await next.Send(context)就能中断管道，但现在不管调不调用，消息都会被发送。以下是复现问题的简化代码，请问该如何阻止消息发送？

复现代码 #

using MassTransit;
namespace Test;
public class Program
{
    public static async Task Main(string[] args)
    {
        var builder = WebApplication.CreateBuilder(args);
        IServiceCollection services = builder.Services;

        services.AddMassTransit(x =>
        {
            x.UsingRabbitMq((context, cfg) =>
            {
                // 拦截发送操作
                cfg.UseSendFilter(typeof(OutboxPreSendFilter<>), context);

                cfg.Host("localhost", 5672, "/", h =>
                {
                    h.Username("test");
                    h.Password("test");
                });
            });
        });

        var app = builder.Build();

        using var scope = app.Services.CreateScope();
        var sendEndpointProvider = scope.ServiceProvider.GetRequiredService<ISendEndpointProvider>();
        // 尝试发送消息到RabbitMQ
        await SendAndReceiveFromQueue(sendEndpointProvider);

        app.Run();
    }

    public static async Task SendAndReceiveFromQueue(ISendEndpointProvider sendEndpointProvider)
    {
        var messageToSent = Guid.NewGuid().ToString();
        var sendEndpoint = await sendEndpointProvider.GetSendEndpoint(new Uri($"queue:{nameof(TestMsg)}"));
        // 发送消息
        await sendEndpoint.Send(new TestMsg() { Id = 1, Name = messageToSent, Description = $"{DateTime.Now:f}" });
    }
}

public class OutboxPreSendFilter<T> : IFilter<SendContext<T>> where T : class
{
    public async Task Send(SendContext<T> context, IPipe<SendContext<T>> next)
    {
        // 这里执行保存消息到数据库的逻辑
 
        // 不想发送消息，所以注释了下面这行
        // await next.Send(context);
    }

    public void Probe(ProbeContext context)
    {
        context.CreateFilterScope("outboxPreSendFilter");
    }
}

public class TestMsg
{
    public int Id { get; set; }
    public string? Name { get; set; }
    public string? Description { get; set; }
}

解决方案 #

1. 修正过滤器注册方式 #

把RabbitMQ传输级的过滤器改为全局发送过滤器，确保所有Send操作都被拦截：

services.AddMassTransit(x =>
{
    // 添加全局发送过滤器，覆盖所有发送场景
    x.AddSendFilter(typeof(OutboxPreSendFilter<>), context);

    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host("localhost", 5672, "/", h =>
        {
            h.Username("test");
            h.Password("test");
        });
    });
});

2. 在过滤器中显式取消发送上下文 #

仅不调用next.Send(context)可能不足以阻止发送（部分MassTransit版本存在逻辑差异），需要显式取消上下文，确保后续发送逻辑终止：

public class OutboxPreSendFilter<T> : IFilter<SendContext<T>> where T : class
{
    public async Task Send(SendContext<T> context, IPipe<SendContext<T>> next)
    {
        // 执行保存消息到数据库的逻辑
        // 例如：await _dbContext.OutboxMessages.AddAsync(...);
        // await _dbContext.SaveChangesAsync();

        // 显式取消发送上下文，阻止后续发送操作
        context.Cancel();

        // 不要调用next.Send(context)
    }

    public void Probe(ProbeContext context)
    {
        context.CreateFilterScope("outboxPreSendFilter");
    }
}

原因说明 #

• 原注册方式cfg.UseSendFilter仅针对RabbitMQ传输层的发送操作，无法覆盖SendEndpoint.Send这类全局发送场景，导致过滤器未生效。
• 部分MassTransit版本中，即使不调用next.Send(context)，底层默认逻辑仍可能执行发送，显式调用context.Cancel()会让后续发送逻辑检查到上下文已取消，从而终止发送流程。

内容的提问来源于stack exchange，提问作者Antonio Padrin