.NET 使用 WolverineFX 实现消息处理…

在 .NET 项目中用 WolverineFX/Marten 编写 CQRS、消息 Handler、HTTP 端点、可靠消息、批处理、重试与多租户时，按以下做法落地。

适用场景

解决思路

1. 核心原则落地：用纯函数和最小样板写 Handler，优先方法注入，让业务逻辑不依赖基础设施；按垂直切片组织代码，把基础设施与业务逻辑隔离。
2. 用级联消息替代深层注入 IMessageBus：Handler 通过返回值发出后续消息；用事务性 Outbox 保证消息投递；用运行时代码生成或预生成代码优化性能。
3. 安装核心 NuGet 包：添加 <PackageReference Include="Wolverine" /> 和 <PackageReference Include="WolverineFx.Http" />。
4. 按需安装持久化与传输 NuGet 包：添加 <PackageReference Include="WolverineFx.Marten" />；传输添加 <PackageReference Include="WolverineFx.RabbitMQ" />、<PackageReference Include="WolverineFx.AzureServiceBus" />、<PackageReference Include="WolverineFx.Kafka" />、<PackageReference Include="WolverineFx.AmazonSQS" />。
5. 在 ASP.NET Core Program.cs 中配置：builder.Host.UseWolverine(opts => { opts.Policies.AutoApplyTransactions(); opts.Policies.UseDurableLocalQueues(); }); builder.Services.AddWolverineHttp(); var app = builder.Build(); app.MapWolverineEndpoints(); return await app.RunJasperFxCommands(args);
6. 编写简单消息 Handler：定义 public record DebitAccount(long AccountId, decimal Amount); 在静态类中写 public static void Handle(DebitAccount command, IAccountRepository repository) { repository.Debit(command.AccountId, command.Amount); }。
7. 编写带级联消息的 Handler：定义 CreateOrder 和 OrderCreated 等 record，在 Handler 中通过 IDocumentSession 保存订单，并返回 (new OrderCreated(command.OrderId), new ShipOrder(command.OrderId)) 这样的消息元组，而不是注入 IMessageBus。
8. 发送多条级联消息：让 Handler 返回 OutgoingMessages，new OutgoingMessages { new OrderProcessed(command.OrderId), new SendEmail(command.CustomerEmail, "Order processed"), new UpdateInventory(command.Items) }; 然后调用 messages.Delay(new CleanupOrder(command.OrderId), 5.Minutes()); 并 return messages。
9. 创建 HTTP GET 端点：用 [WolverineGet("/users/{id}")] 标注 public static Task<User?> GetUser(int id, IQuerySession session) => session.LoadAsync<User>(id);。
10. 创建 HTTP POST 并发布消息：用 [WolverinePost("/orders")] 标注 async Task<IResult> CreateOrder(CreateOrderRequest request, IDocumentSession session, IMessageBus bus)，创建并 session.Store(order)，await bus.PublishAsync(new OrderCreated(order.Id))，return Results.Created($"/orders/{order.Id}", order)。
11. 创建复合 Handler 分离 Load/Validate/Handle：在 UpdateOrderEndpoint 中写 static async Task<(Order?, IResult)> LoadAsync(UpdateOrder command, IDocumentSession session)，找到返回 (order, new WolverineContinue())，否则返回 (order, Results.NotFound())；再用 [WolverinePut("/orders")] 标注 Handle(UpdateOrder command, Order order, IDocumentSession session) 更新并 session.Store(order)。
12. 实现事件溯源聚合 Handler：定义聚合 Order 及其 Apply(ItemReady) 和 Apply(IEvent<OrderShipped>) 等方法；用 [AggregateHandler] 标注 static IEnumerable<object> Handle(MarkItemReady command, Order order)，根据条件 yield return new ItemReady(command.ItemName) 和 yield return new OrderReady()。
13. 读取只读聚合：用 [WolverineGet("/orders/{id}")] 标注 public static Order GetOrder([ReadAggregate] Order order) => order;。
14. 写聚合时带验证：在 Handle 参数使用 [WriteAggregate(Required = true, OnMissing = OnMissing.ProblemDetailsWith404)] Order order，然后修改聚合并 yield return new ItemReady(command.ItemName)。
15. 返回更新后的聚合：用 [AggregateHandler] 标注 static (UpdatedAggregate, Events) Handle(MarkItemReady command, Order order)，创建 var events = new Events(); events.Add(new ItemReady(command.ItemName)); return (new UpdatedAggregate(), events);。
16. 集成 Marten 与事务性 Outbox：builder.Services.AddMarten(opts => { opts.Connection(connectionString); }).IntegrateWithWolverine(); builder.Host.UseWolverine(opts => { opts.Policies.AutoApplyTransactions(); });。
17. 在 Controller 中使用 Outbox：注入 [FromServices] IDocumentSession session 和 [FromServices] IMartenOutbox outbox，调用 outbox.Enroll(session); session.Store(order); await outbox.PublishAsync(new OrderCreated(command.OrderId)); await session.SaveChangesAsync();。
18. 配置 RabbitMQ 传输：opts.UseRabbitMq("host=localhost").AutoProvision().AutoPurgeOnStartup(); opts.PublishAllMessages().ToRabbitExchange("wolverine.events", exchange => { exchange.ExchangeType = ExchangeType.Topic; });。
19. 配置 Azure Service Bus 传输：opts.UseAzureServiceBus(asbConnectionString).AutoProvision().ConfigureQueue(q => q.MaxDeliveryCount = 5);。
20. 配置 Amazon SQS 传输：opts.UseAmazonSqs(sqsConfig).AutoProvision();。
21. 配置批量消息处理：opts.BatchMessagesOf<SubTaskCompleted>(batching => { batching.BatchSize = 500; batching.TriggerTime = 1.Seconds(); });。
22. 编写批量 Handler：定义 public static void Handle(Item[] items, IRepository repository)，在方法内 foreach (var item in items) { repository.Process(item); }。
23. 自定义批处理策略：定义 SubTaskCompleted 和 SubTaskBatch；实现 IMessageBatcher，在 Group 中按 x.Message!.As<SubTaskCompleted>().TaskId 分组，提取 SubTaskId 数组，yield return new Envelope(new SubTaskBatch(group.Key, subTaskIds), group); 并设置 public Type BatchMessageType => typeof(SubTaskBatch);。
24. 配置错误重试策略：opts.Policies.OnException<SqlException>().RetryWithCooldown(50.Milliseconds(), 100.Milliseconds(), 250.Milliseconds()); opts.Policies.OnException<TimeoutException>().RetryTimes(3); opts.Policies.OnException<InvalidOperationException>().Requeue(); opts.Policies.OnAnyException().MoveToErrorQueue();。
25. 配置断路器：opts.ListenToRabbitQueue("incoming").CircuitBreaker(cb => { cb.PauseTime = 1.Minutes(); cb.FailurePercentageThreshold = 50; cb.MinimumThreshold = 10; });。
26. 发送延迟消息：在 Handler 中 yield return new ProcessPayment(command.OrderId); yield return new ShipOrder(command.OrderId).DelayedFor(30.Minutes()); 或使用 OutgoingMessages 的 messages.Delay(new Reminder(orderId), TimeSpan.FromHours(24));。
27. 发送定时消息：yield return new GenerateReport().ScheduledAt(DateTime.Today.AddDays(1)); 或使用 messages.Schedule(new MonthlyReport(), DateTime.Today.AddMonths(1));。
28. 实现请求/回复：调用 await bus.InvokeAsync<OrderStatus>(new GetOrder(orderId))；编写 Handler public static OrderStatus Handle(GetOrder query, IQuerySession session) 并返回 OrderStatus。
29. 添加自定义中间件：定义 LoggingMiddleware，在 Before(Envelope envelope, ILogger logger) 和 After(Envelope envelope, ILogger logger) 中记录日志，并用 opts.Handlers.AddMiddleware<LoggingMiddleware>(); 注册；事务中间件通过 opts.Policies.AutoApplyTransactions(); 启用。
30. 预生成代码：运行 dotnet run -- codegen write，生成代码到 ./Internal/Generated/WolverineHandlers/；为 AOT/Trimming 配置 opts.CodeGeneration.TypeLoadMode = TypeLoadMode.Auto;。
31. 配置多租户：opts.Policies.ConjoinedTenancy(x => { x.TenantIdStyle = TenantIdStyle.PerRequest; }); 发布到指定租户时用 await bus.PublishAsync(new OrderCreated(orderId), new DeliveryOptions { TenantId = "tenant-1" });。
32. 使用 Marten 副作用：让 Handler 返回 IMartenOp，用 return MartenOps.Store(todo); 存储，或用 return MartenOps.Delete<Todo>(command.Id); 删除。
33. 配置附属存储：定义 public interface IPlayerStore : IDocumentStore;，用 opts.Services.AddMartenStore<IPlayerStore>(m => { m.Connection(connectionString); m.DatabaseSchemaName = "players"; }).IntegrateWithWolverine();，在 Handler 上标注 [MartenStore(typeof(IPlayerStore))] 并返回 MartenOps.Store(new Player { Id = message.Id });。
34. 使用 CLI 命令：运行 dotnet run -- help 列出命令，dotnet run -- describe 查看应用描述，dotnet run -- resources 管理资源，dotnet run -- storage 管理消息存储，dotnet run -- codegen write 预生成代码。
35. 遵循最佳实践：优先纯函数，使用级联消息，保持短调用栈，预生成代码，使用复合 Handler，配置错误处理，使用事务性 Outbox，在合适时批处理。
36. 避免反模式：不要在调用栈深处注入 IMessageBus，不要过度使用构造函数注入，不要忽略事务性 Outbox，不要不预生成代码，不要在一个 Handler 中混太多关注点，不要不配置错误处理。
37. 配合相关技能：结合 efcore-patterns、csharp-coding-standards、http-client-resilience、background-services、aspire-configuration 使用。

适用边界 · 注意事项

• 项目未引入 WolverineFX/Marten 或不能添加这些 NuGet 与传输依赖时，不要按本经验硬套。
• 只需要简单 CRUD，且不需要消息、事件溯源、CQRS 或事务性 Outbox 时，不必引入 WolverineFX 的复杂度。
• 没有 RabbitMQ、Azure Service Bus、Amazon SQS、Kafka 等传输基础设施，且不需要本地队列或 TCP 时，不要配置这些传输。