mcpField
C
csharp-wolverinefx

csharp-wolverinefx

github.com/wshaddix开发86570 次安装35 周前更新
代码仓库

关于 csharp-wolverinefx

面向使用 Wolverine 构建 .NET 应用的开发者，提供消息处理、HTTP 端点、事件溯源和传输配置的实践指导。内容还覆盖事务发件箱、批处理、错误处理与代码生成。

主要功能

  1. 01消息与命令处理器
  2. 02WolverineFx.HTTP 端点
  3. 03Marten 事件溯源集成
  4. 04事务发件箱与消息传输
  5. 05重试策略和批处理
  6. 06代码预生成

适用场景

  1. 01构建消息处理工作流
  2. 02实现 CQRS 与事件溯源
  3. 03配置 RabbitMQ 或云消息传输
  4. 04设置可靠投递和失败重试

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/wshaddix/dotnet-skills --skill csharp-wolverinefx

安装后在 Claude Code 中输入/csharp-wolverinefx确认已加载。

详细介绍

.NET 的 WolverineFX

何时使用此技能

在以下情况下使用此技能：

  • 使用 Wolverine 构建消息处理程序或命令处理程序
  • 使用 WolverineFx.HTTP 创建 HTTP 端点（Minimal API/MVC 的替代方案）
  • 使用 Marten 和 Wolverine 实现事件溯源
  • 设置事务发件箱模式，实现可靠消息传递
  • 配置消息传输（RabbitMQ、Azure Service Bus、Amazon SQS、TCP）
  • 使用事件溯源实现 CQRS
  • 批量处理消息
  • 使用级联消息构建可测试的纯函数处理程序
  • 配置错误处理和重试策略
  • 预生成代码以优化冷启动

相关技能

  • efcore-patterns — 用于数据访问的 Entity Framework Core 模式
  • csharp-coding-standards — 现代 C# 模式（记录、模式匹配）
  • http-client-resilience — Polly 弹性模式（互为补充）
  • background-services — 托管服务和后台作业模式
  • aspire-configuration — .NET Aspire 编排

核心原则

  1. 低仪式感代码 — 纯函数、方法注入、最少样板代码
  2. 级联消息 — 从处理程序返回消息，而不是注入 IMessageBus
  3. 事务发件箱 — 通过数据库事务保证消息送达
  4. 代码生成 — 运行时生成或预生成代码，以实现最佳性能
  5. 垂直切片架构 — 按功能而非技术层组织代码
  6. 业务逻辑使用纯函数 — 将基础设施与业务逻辑隔离

必需的 NuGet 包

核心消息传递
<PackageReference Include="Wolverine" />
<PackageReference Include="WolverineFx.Http" />
持久化集成
<PackageReference Include="WolverineFx.Marten" />
传输
<PackageReference Include="WolverineFx.RabbitMQ" />
<PackageReference Include="WolverineFx.AzureServiceBus" />
<PackageReference Include="WolverineFx.Kafka" />
<PackageReference Include="WolverineFx.AmazonSQS" />

基本设置

Program.cs（ASP.NET Core）
using JasperFx;
using Wolverine;
using Wolverine.Http;

var builder = WebApplication.CreateBuilder(args);

builder.Host.UseWolverine(opts =>
{
    opts.Policies.AutoApplyTransactions();
    opts.Policies.UseDurableLocalQueues();
});

builder.Services.AddWolverineHttp();

var app = builder.Build();

app.MapWolverineEndpoints();

return await app.RunJasperFxCommands(args);

消息处理程序

简单消息处理程序
public record DebitAccount(long AccountId, decimal Amount);

public static class DebitAccountHandler
{
    public static void Handle(DebitAccount command, IAccountRepository repository)
    {
        repository.Debit(command.AccountId, command.Amount);
    }
}
带级联消息的处理程序
public record CreateOrder(Guid OrderId, string[] Items);
public record OrderCreated(Guid OrderId);

public static class CreateOrderHandler
{
    public static (OrderCreated, ShipOrder) Handle(
        CreateOrder command,
        IDocumentSession session)
    {
        var order = new Order { Id = command.OrderId, Items = command.Items };
        session.Store(order);
        
        return (
            new OrderCreated(command.OrderId),
            new ShipOrder(command.OrderId)
        );
    }
}
使用 OutgoingMessages 发送多条消息
public static OutgoingMessages Handle(ProcessOrder command)
{
    var messages = 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;
}

HTTP 端点（WolverineFx.HTTP）

基本 GET 端点
[WolverineGet("/users/{id}")]
public static Task<User?> GetUser(int id, IQuerySession session)
    => session.LoadAsync<User>(id);
使用消息发布的 POST
[WolverinePost("/orders")]
public static async Task<IResult> CreateOrder(
    CreateOrderRequest request,
    IDocumentSession session,
    IMessageBus bus)
{
    var order = new Order { Id = Guid.NewGuid(), Items = request.Items };
    session.Store(order);
    
    await bus.PublishAsync(new OrderCreated(order.Id));
    
    return Results.Created($"/orders/{order.Id}", order);
}
复合处理程序（加载/验证/处理）
public static class UpdateOrderEndpoint
{
    public static async Task<(Order?, IResult)> LoadAsync(
        UpdateOrder command,
        IDocumentSession session)
    {
        var order = await session.LoadAsync<Order>(command.OrderId);
        return order != null
            ? (order, new WolverineContinue())
            : (order, Results.NotFound());
    }

    [WolverinePut("/orders")]
    public static void Handle(UpdateOrder command, Order order, IDocumentSession session)
    {
        order.Items = command.Items;
        session.Store(order);
    }
}

使用 Marten 实现事件溯源

聚合处理程序工作流
public class Order
{
    public Guid Id { get; set; }
    public int Version { get; set; }
    public Dictionary<string, Item> Items { get; set; } = new();
    public DateTimeOffset? Shipped { get; private set; }

    public void Apply(ItemReady ready) => Items[ready.Name].Ready = true;
    public void Apply(IEvent<OrderShipped> shipped) => Shipped = shipped.Timestamp;

    public bool IsReadyToShip() => Shipped == null && Items.Values.All(x => x.Ready);
}

public record MarkItemReady(Guid OrderId, string ItemName, int Version);

[AggregateHandler]
public static IEnumerable<object> Handle(MarkItemReady command, Order order)
{
    if (order.Items.TryGetValue(command.ItemName, out var item))
    {
        item.Ready = true;
        yield return new ItemReady(command.ItemName);
    }
    
    if (order.IsReadyToShip())
    {
        yield return new OrderReady();
    }
}
读取聚合（只读）
[WolverineGet("/orders/{id}")]
public static Order GetOrder([ReadAggregate] Order order) => order;
验证后写入聚合
public static IEnumerable<object> Handle(
    MarkItemReady command,
    [WriteAggregate(Required = true, OnMissing = OnMissing.ProblemDetailsWith404)] Order order)
{
    order.Items[command.ItemName].Ready = true;
    yield return new ItemReady(command.ItemName);
}
返回更新后的聚合
[AggregateHandler]
public static (UpdatedAggregate, Events) Handle(
    MarkItemReady command,
    Order order)
{
    var events = new Events();
    events.Add(new ItemReady(command.ItemName));
    return (new UpdatedAggregate(), events);
}

事务发件箱

Marten 集成
builder.Services.AddMarten(opts =>
{
    opts.Connection(connectionString);
}).IntegrateWithWolverine();

builder.Host.UseWolverine(opts =>
{
    opts.Policies.AutoApplyTransactions();
});
在控制器中使用发件箱
[HttpPost("/orders")]
public async Task Post(
    [FromBody] CreateOrder command,
    [FromServices] IDocumentSession session,
    [FromServices] IMartenOutbox outbox)
{
    outbox.Enroll(session);
    
    var order = new Order { Id = command.OrderId };
    session.Store(order);
    
    await outbox.PublishAsync(new OrderCreated(command.OrderId));
    
    await session.SaveChangesAsync();
}

传输配置

RabbitMQ
builder.Host.UseWolverine(opts =>
{
    opts.UseRabbitMq("host=localhost")
        .AutoProvision()
        .AutoPurgeOnStartup();
    
    opts.PublishAllMessages()
        .ToRabbitExchange("wolverine.events", exchange =>
        {
            exchange.ExchangeType = ExchangeType.Topic;
        });
});
Azure Service Bus
builder.Host.UseWolverine(opts =>
{
    opts.UseAzureServiceBus(asbConnectionString)
        .AutoProvision()
        .ConfigureQueue(q => q.MaxDeliveryCount = 5);
});
Amazon SQS
builder.Host.UseWolverine(opts =>
{
    opts.UseAmazonSqs(sqsConfig)
        .AutoProvision();
});

批量消息处理

配置批处理
builder.Host.UseWolverine(opts =>
{
    opts.BatchMessagesOf<SubTaskCompleted>(batching =>
    {
        batching.BatchSize = 500;
        batching.TriggerTime = 1.Seconds();
    });
});
批处理程序
public static class ItemBatchHandler
{
    public static void Handle(Item[] items, IRepository repository)
    {
        foreach (var item in items)
        {
            repository.Process(item);
        }
    }
}
自定义批处理策略
public record SubTaskCompleted(string TaskId, string SubTaskId);
public record SubTaskBatch(string TaskId, string[] SubTaskIds);

public class SubTaskBatcher : IMessageBatcher
{
    public IEnumerable<Envelope> Group(IReadOnlyList<Envelope> envelopes)
    {
        var groups = envelopes
            .GroupBy(x => x.Message!.As<SubTaskCompleted>().TaskId);
        
        foreach (var group in groups)
        {
            var subTaskIds = group
                .Select(x => x.Message)
                .OfType<SubTaskCompleted>()
                .Select(x => x.SubTaskId)
                .ToArray();
            
            yield return new Envelope(
                new SubTaskBatch(group.Key, subTaskIds),
                group);
        }
    }

    public Type BatchMessageType => typeof(SubTaskBatch);
}

错误处理

重试策略
builder.Host.UseWolverine(opts =>
{
    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();
});
断路器
opts.ListenToRabbitQueue("incoming")
    .CircuitBreaker(cb =>
    {
        cb.PauseTime = 1.Minutes();
        cb.FailurePercentageThreshold = 50;
        cb.MinimumThreshold = 10;
    });

定时消息

延迟消息
public static IEnumerable<object> Handle(OrderCreated command)
{
    yield return new ProcessPayment(command.OrderId);
    yield return new ShipOrder(command.OrderId)
        .DelayedFor(30.Minutes());
}
在指定时间定时发送
yield return new GenerateReport()
    .ScheduledAt(DateTime.Today.AddDays(1));
使用 OutgoingMessages
var messages = new OutgoingMessages();
messages.Delay(new Reminder(orderId), TimeSpan.FromHours(24));
messages.Schedule(new MonthlyReport(), DateTime.Today.AddMonths(1));

请求/响应模式

发送请求并等待响应
public async Task<OrderStatus> GetOrderStatus(IMessageBus bus, Guid orderId)
{
    return await bus.InvokeAsync<OrderStatus>(new GetOrder(orderId));
}
返回响应的处理程序
public record GetOrder(Guid OrderId);
public record OrderStatus(Guid OrderId, string Status);

public static class GetOrderHandler
{
    public static OrderStatus Handle(GetOrder query, IQuerySession session)
    {
        var order = session.Load<Order>(query.OrderId);
        return new OrderStatus(query.OrderId, order?.Status ?? "NotFound");
    }
}

中间件

自定义中间件
public class LoggingMiddleware
{
    public void Before(Envelope envelope, ILogger logger)
    {
        logger.LogInformation("Processing {MessageType}", envelope.MessageType);
    }

    public void After(Envelope envelope, ILogger logger)
    {
        logger.LogInformation("Completed {MessageType}", envelope.MessageType);
    }
}

builder.Host.UseWolverine(opts =>
{
    opts.Handlers.AddMiddleware<LoggingMiddleware>();
});
事务中间件
builder.Host.UseWolverine(opts =>
{
    opts.Policies.AutoApplyTransactions();
});

代码生成

预生成类型
dotnet run -- codegen write

生成的代码位于 ./Internal/Generated/WolverineHandlers/

配置 AOT/修剪
builder.Host.UseWolverine(opts =>
{
    opts.CodeGeneration.TypeLoadMode = TypeLoadMode.Auto;
});

多租户

联合式租户
builder.Host.UseWolverine(opts =>
{
    opts.Policies.ConjoinedTenancy(x =>
    {
        x.TenantIdStyle = TenantIdStyle.PerRequest;
    });
});
发布到指定租户
await bus.PublishAsync(new OrderCreated(orderId), 
    new DeliveryOptions { TenantId = "tenant-1" });

Marten 副作用

public static IMartenOp Handle(CreateTodo command)
{
    var todo = new Todo { Name = command.Name };
    return MartenOps.Store(todo);
}

public static IMartenOp Handle(DeleteTodo command)
{
    return MartenOps.Delete<Todo>(command.Id);
}

辅助存储（模块化单体）

public interface IPlayerStore : IDocumentStore;

builder.Host.UseWolverine(opts =>
{
    opts.Services.AddMartenStore<IPlayerStore>(m =>
    {
        m.Connection(connectionString);
        m.DatabaseSchemaName = "players";
    })
    .IntegrateWithWolverine();
});

[MartenStore(typeof(IPlayerStore))]
public static class PlayerMessageHandler
{
    public static IMartenOp Handle(PlayerMessage message)
    {
        return MartenOps.Store(new Player { Id = message.Id });
    }
}

命令行工具

可用命令
dotnet run -- help           # List all commands
dotnet run -- describe       # Application description
dotnet run -- resources      # Resource management
dotnet run -- storage        # Message storage admin
dotnet run -- codegen write  # Pre-generate code

最佳实践

  1. 优先使用纯函数 - 业务逻辑应当无需模拟对象即可测试
  2. 使用级联消息 - 返回消息，而不是注入 IMessageBus
  3. 保持调用栈简短 - 避免深层服务层级
  4. 预生成代码 - 优化生产环境中的冷启动
  5. 使用复合处理程序 - 分离加载、验证和处理逻辑
  6. 配置错误处理 - 让 Wolverine 处理重试和错误
  7. 使用事务发件箱 - 确保消息送达
  8. 适时批处理 - 提高高吞吐量消息的处理能力

应避免的反模式

  1. 在调用栈深处注入 IMessageBus - 使工作流难以理解
  2. 过度使用构造函数注入 - 优先使用方法注入
  3. 忽略事务发件箱 - 失败时可能丢失消息
  4. 不预生成代码 - 导致生产环境冷启动缓慢
  5. 在一个处理程序中混杂过多职责 - 保持处理程序专注
  6. 未配置错误处理 - 消息可能意外进入错误队列

评论

支持 Markdown