messaging

消息传递

核心原则

1. Wolverine 是推荐的默认选择——采用 MIT 许可证，将中介器和消息传递整合在一个库中，内置 outbox、saga 支持和基于约定的处理器。MassTransit 是一种替代方案，但从 v9 起需要商业许可证。
2. 使用 outbox 模式确保可靠性——始终使用事务性 outbox，确保只有在数据库事务成功后才发布消息。
3. 简单流程使用编舞模式，复杂流程使用 saga——如果工作流包含 2-3 个步骤，请使用事件编舞。如果涉及补偿操作或复杂状态，请使用 saga。
4. 消息就是契约——将消息类型放在共享的契约项目中。将其保持为使用基本类型的简单记录。

模式

Wolverine 设置

// Program.cs
builder.Host.UseWolverine(opts =>
{
    // Auto-discover handlers from this assembly
    opts.Discovery.IncludeAssembly(typeof(Program).Assembly);

    // RabbitMQ transport
    opts.UseRabbitMq(rabbit =>
    {
        rabbit.HostName = "localhost";
        // Or from configuration:
        // rabbit.HostName = builder.Configuration["RabbitMq:Host"]!;
    })
    .AutoProvision()   // Create queues/exchanges automatically
    .AutoPurgeOnStartup(); // Dev only — clear queues on startup

    // Enable transactional outbox with EF Core
    opts.Services.AddDbContextWithWolverineIntegration<AppDbContext>(x =>
        x.UseNpgsql(builder.Configuration.GetConnectionString("Default")));

    opts.Policies.AutoApplyTransactions(); // Wrap handlers in DB transactions
});

原因：UseWolverine() 会在同一处注册处理器发现、传输和 outbox。AutoProvision() 可在开发期间免去手动配置消息代理。

发布事件

Wolverine 支持两种发布方式：级联消息（返回值）和显式发布。

// Message contract (in shared Contracts project)
public record OrderCreated(Guid OrderId, string CustomerId, decimal Total, DateTimeOffset CreatedAt);

// Style 1: Cascading messages — return the event from the handler
// Wolverine automatically publishes returned messages after the handler completes.
public static class CreateOrder
{
    public record Command(string CustomerId, List<OrderItem> Items);
    public record Response(Guid OrderId, decimal Total);

    public static async Task<(Response, OrderCreated)> HandleAsync(
        Command command, AppDbContext db, TimeProvider clock, CancellationToken ct)
    {
        var order = Order.Create(command.CustomerId, command.Items, clock.GetUtcNow());
        db.Orders.Add(order);
        await db.SaveChangesAsync(ct);

        var response = new Response(order.Id, order.Total);
        var @event = new OrderCreated(order.Id, order.CustomerId, order.Total, order.CreatedAt);

        return (response, @event); // Both are published automatically
    }
}

// Style 2: Explicit publishing via IMessageBus
public static class CreateOrder
{
    public record Command(string CustomerId, List<OrderItem> Items);
    public record Response(Guid OrderId, decimal Total);

    public static async Task<Response> HandleAsync(
        Command command, AppDbContext db, IMessageBus bus, TimeProvider clock, CancellationToken ct)
    {
        var order = Order.Create(command.CustomerId, command.Items, clock.GetUtcNow());
        db.Orders.Add(order);
        await db.SaveChangesAsync(ct);

        await bus.PublishAsync(new OrderCreated(
            order.Id, order.CustomerId, order.Total, order.CreatedAt));

        return new Response(order.Id, order.Total);
    }
}

原因：级联消息（元组返回值）更简单且易于测试——处理器是一个纯函数。若发布条件不固定或需要发布多个事件，请使用显式的 IMessageBus。

消费事件

Wolverine 使用基于约定的处理器——无需接口，也无需基类。只需编写一个以消息类型作为第一个参数的 Handle 方法。

// Notifications module — handles OrderCreated from Orders module
public static class OrderCreatedHandler
{
    public static async Task HandleAsync(
        OrderCreated message, NotificationsDbContext db, ILogger logger, CancellationToken ct)
    {
        logger.LogInformation("Processing OrderCreated: {OrderId}", message.OrderId);

        var notification = new OrderNotification(message.OrderId, message.CustomerId);
        db.Notifications.Add(notification);
        await db.SaveChangesAsync(ct);
    }
}

原因：基于约定的处理器不需要任何样板代码。Wolverine 会根据签名发现处理器：任何以消息类型作为第一个参数、名为 Handle/HandleAsync/Consume/ConsumeAsync 的公共方法都可以。

事务性 Outbox

确保只有在数据库事务成功时才发布消息。

// 1. Register DbContext with Wolverine integration
builder.Host.UseWolverine(opts =>
{
    opts.Services.AddDbContextWithWolverineIntegration<AppDbContext>(x =>
        x.UseNpgsql(builder.Configuration.GetConnectionString("Default")));

    opts.Policies.AutoApplyTransactions();
});

// 2. DbContext — add Wolverine outbox tables
public class AppDbContext(DbContextOptions<AppDbContext> options) : DbContext(options)
{
    public DbSet<Order> Orders => Set<Order>();

    protected override void OnModelCreating(ModelBuilder modelBuilder)
    {
        // Wolverine inbox/outbox tables — required for transactional messaging
        modelBuilder.AddIncomingWolverineMessageTable();
        modelBuilder.AddOutgoingWolverineMessageTable();
    }
}

原因：AddDbContextWithWolverineIntegration + AutoApplyTransactions 会将每个处理器包装在一个包含 outbox 写入操作的事务中。只有在事务提交后才会发送消息——不会出现双写问题。

Saga（有状态编排）

Wolverine saga 使用 Saga<T> 基类及其 Start 和 Handle 方法。级联消息推动 saga 向前执行。

public record OrderSagaState(Guid Id)
{
    public string? CustomerId { get; set; }
    public bool PaymentReceived { get; set; }
}

public class OrderSaga : Saga<OrderSagaState>
{
    public Guid Id { get; set; }

    // Start the saga when an OrderCreated event arrives
    public static (OrderSagaState, ProcessPayment) Start(OrderCreated message)
    {
        var state = new OrderSagaState(message.OrderId)
        {
            CustomerId = message.CustomerId
        };

        var command = new ProcessPayment(message.OrderId, message.Total);
        return (state, command); // State is persisted, command is sent
    }

    // Handle payment result
    public CompleteOrder Handle(PaymentCompleted message)
    {
        PaymentReceived = true;
        MarkCompleted(); // Ends the saga
        return new CompleteOrder(Id);
    }

    // Compensating action on failure
    public CancelOrder Handle(PaymentFailed message)
    {
        MarkCompleted();
        return new CancelOrder(Id);
    }
}

原因：Wolverine saga 使用简单的 C# 方法，而非状态机 DSL。每个处理器都会返回级联消息以推动工作流执行。MarkCompleted() 会清理 saga 状态。

替代方案：MassTransit

MassTransit 是一种成熟的替代方案，但从 v9 起需要商业许可证。主要 API：

// Setup
builder.Services.AddMassTransit(x =>
{
    x.SetKebabCaseEndpointNameFormatter();
    x.AddConsumers(typeof(Program).Assembly);
    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host(builder.Configuration.GetConnectionString("RabbitMq"));
        cfg.ConfigureEndpoints(context);
    });
});

// Publishing
await publishEndpoint.Publish(new OrderCreated(...), ct);

// Consuming — requires IConsumer<T> interface
public class OrderCreatedConsumer(AppDbContext db) : IConsumer<OrderCreated>
{
    public async Task Consume(ConsumeContext<OrderCreated> context)
    {
        var message = context.Message;
        // Handle event...
    }
}

// Outbox
x.AddEntityFrameworkOutbox<AppDbContext>(o =>
{
    o.UsePostgres();
    o.UseBusOutbox();
});

// Saga — uses MassTransitStateMachine<TState>
public class OrderSaga : MassTransitStateMachine<OrderSagaState> { /* ... */ }

许可证说明：MassTransit v9+ 在生产环境中使用需要商业许可证。Wolverine（MIT）是新项目推荐的默认选择。

反模式

不要在没有 Outbox 的情况下发布事件

// BAD — if SaveChanges succeeds but Publish fails, data is inconsistent
await db.SaveChangesAsync(ct);
await bus.PublishAsync(new OrderCreated(...));

// GOOD — use transactional outbox (messages are in the same transaction)
// Configure AddDbContextWithWolverineIntegration() + AutoApplyTransactions()
// Wolverine handles this automatically

不要在消息契约中放入复杂逻辑

// BAD — behavior in a message
public record OrderCreated(Guid OrderId)
{
    public decimal CalculateShipping() => /* logic */; // DON'T
}

// GOOD — messages are pure data
public record OrderCreated(Guid OrderId, string CustomerId, decimal Total, DateTimeOffset CreatedAt);

不要对重要事件使用即发即弃模式

// BAD — no guarantee of delivery
_ = Task.Run(() => bus.PublishAsync(new OrderCreated(...)));

// GOOD — await the publish (with outbox, this is transactional)
await bus.PublishAsync(new OrderCreated(...));

决策指南

场景	建议
模块间通信（新项目）	使用事件的 Wolverine（MIT，免费）
模块间通信（现有 MassTransit 项目）	MassTransit（从 v9 起需要商业许可证）
可靠地发布事件	事务性 outbox（Wolverine 和 MassTransit 均支持）
简单的 2-3 步工作流	事件编舞
带补偿操作的复杂工作流	Wolverine saga 或 MassTransit saga
本地开发消息代理	RabbitMQ（通过 Docker 或 Aspire）
生产环境云消息代理	Azure Service Bus 或 RabbitMQ
希望使用一个库同时实现中介器和消息传递	Wolverine（同时替代 Mediator 和 MassTransit）