Sparkdo.Messaging 3.0.0-preview.1

Sparkdo.Messaging

内核：注册面（IMessagingBuilder）、编译一次就不再变的冻结图、派发与接收管道、 六条扩展接缝、Saga、请求回投、端点运行时。Roslyn 生成器作为分析器资产随本包发出—— PackageReference 装上就生效，不需要任何挂接。

1. 五分钟跑通

三件事：定义消息与处理器、接线、起消费。

// ① 消息：POCO 就够，不需要基类或属性
public sealed record OrderPlaced(string OrderId);

// ② 处理器：实现一个接口
public sealed class OrderPlacedHandler(ILogger<OrderPlacedHandler> log) : IEventHandler<OrderPlaced>
{
    public ValueTask HandleAsync(OrderPlaced message, CancellationToken cancellationToken)
    {
        log.LogInformation("收到 {OrderId}", message.OrderId);
        return ValueTask.CompletedTask;
    }
}

// ③ 接线（Web 宿主；控制台把 Host 那段换成同样的 provider 操作）
var builder = WebApplication.CreateBuilder(args);

builder.Services
    .AddSingleton<OrderPlacedHandler>()                        // 处理器必须在容器里
    .AddSparkdoMessaging(
        new MessageSerializerRegistry().Add(new JsonMessageSerializer(json)),
        bus => bus.AddTransport(new InMemoryTransport(), TransportDefault.Default))
    .AddSparkdoMessagingHosting();                             // 起消费、停机收尾（见 Hosting 包）

var app = builder.Build();
await app.StartAsync();

var bus = app.Services.GetRequiredService<IMessageBus>();
await bus.PublishAsync(new OrderPlaced("A-1"), PublishOptions.Default, ct);

没有 AddHandler。 生成器在编译期发现 OrderPlacedHandler，把 AddHandler<OrderPlacedHandler, OrderPlaced>("程序集名") 写进 [ModuleInitializer]， AddSparkdoMessaging 建构建器时回放这批登记。要显式登记也完全可以（下一节），两条路能混用。

不想引宿主包时，自己起循环（这段就是 Sparkdo.Messaging.Hosting 里那段）：

var graph = await provider.BuildMessagingAsync(ct);        // 建图：打开端口、编译管道

using var consumers = new CancellationTokenSource();
var running = graph.ReceiveEndpoints
    .Select(endpoint => Task.Run(() => endpoint.RunAsync(consumers.Token).AsTask()))
    .ToArray();

var bus = graph.CreateBus(provider);
await bus.PublishAsync(new OrderPlaced("A-1"), PublishOptions.Default, ct);

await consumers.CancelAsync();
await Task.WhenAll(running);

2. 接线面：登记什么、什么时候能取

AddSparkdoMessaging(serializers, configure?) 只登记一个"怎么造构建器"的工厂—— 不在登记时造构建器，更不建图：造它要读容器里的命名策略，建图要开端口且是异步的。

图由 provider.BuildMessagingAsync(ct) 建，只能挂一次：

var graph = await provider.BuildMessagingAsync(ct);   // 内部：builder.BuildAsync(provider, ct) + Attach

模块回放的范围（必须知道）

MessagingModuleRegistry.Modules 是进程级的：凡是编译时带了这个分析器的程序集，它的处理器 都会进这批登记。所以：

3. IMessagingBuilder 逐方法

配置期对象，单次可变：BuildAsync 之后不得再改，且只能编译一次（失败的那次也算一次）。

处理器两种生命周期，选错的代价

两种形态的差别只有这一处，所以必须在登记时选好——把它做成"运行时看着办"， 就等于允许"0 分配"与"每消息新实例"互相撒谎。 IHandlerActivator 换的是从哪里取，不改什么时候取：那个决定写在登记方法里。

还有四件事与形态判定有关：

// 处理器要拿"这条投递"的信封/工作单元 → 必须走 scoped 形态
public sealed class SettleOrderHandler(ReceiveContextAccessor contexts, IMessageOutbox outbox)
    : IEventHandler<Settled>
{
    public async ValueTask HandleAsync(Settled message, CancellationToken ct)
    {
        var context = contexts.Current ?? throw new InvalidOperationException("这条投递不在接收路径上。");
        var unit = WorkUnitAccessor.Of(context) ?? throw new InvalidOperationException("没有当前工作单元。");

        await outbox.EnqueueAsync(unit, envelope, payload, ct);   // 与业务写同一笔提交
    }
}

4. 冻结图上能读什么

public sealed class MessageBusGraph
{
    IMessageBus CreateBus(IServiceProvider services);
    string Describe();
    IReadOnlyList<ITransportPort> Transports { get; }
    IReadOnlyList<EndpointPlan> Endpoints { get; }
    IReadOnlyList<EndpointReceivePlan> ReceivePlans { get; }
    IReadOnlyList<ReceiveEndpoint> ReceiveEndpoints { get; }
    IMessageScheduler? Scheduler { get; }
    IReadOnlyList<TopologyEntity> DescribeTopology();
    RoutePlan GetRoute(Type messageType);
    RoutePlan GetRoute(MessageIdentity identity);
    EndpointPlan GetEndpoint(EndpointAddress address);
    EndpointReceivePlan GetReceivePlan(Type consumerType, MessageIdentity subscription);
}

Describe() 的形状（每段的排序口径：传输按方案名序数、拓扑保留各传输自报顺序、 routes 按消息身份、receive 按订阅地址）：

sparkdo.messaging
  transport rabbitmq  capabilities Durable
  topology
    topic sparkdo.messaging
    queue urn:message:sparkdo.orders:order-placed.sparkdo.orders.order-placed-handler
    binding urn:message:sparkdo.orders:order-placed.sparkdo.orders.order-placed-handler  urn:message:sparkdo.orders:order-placed
  routes
    OrderPlaced urn:message:sparkdo.orders:order-placed application/json rabbitmq://topic/urn:message:sparkdo.orders:order-placed identity-chain
  receive
    OrderPlacedHandler urn:message:sparkdo.orders:order-placed rabbitmq://in/urn:message:sparkdo.orders:order-placed/sparkdo.orders.order-placed-handler  middlewares inbox, work-unit

不说积压：那是传输运行期的状态，不是图的知识；要问传输——传输实现可选的 ITransportBacklogProvider 自报积压（InMemory / RabbitMQ / Postgres 三个传输都已实现），Hosting 的健康检查按它逐个探。

5. 中间件：怎么写一条

两侧各一条管道，各一个配置类型，形状都是 (context, next) => 返回新的委托：

// 出站：能看见 Envelope / Payload（还没交出去），改头、改名、审计都在这
var audit = new DispatchMiddlewareConfiguration(
    "audit",
    static (_, next) => async context =>
    {
        Console.WriteLine($"dispatch {context.Envelope?.MessageType}");
        await next(context);
    });

builder.Use(audit);                       // 追加
builder.Use(audit, after: "tenant");      // 定位：before / after 只能给一个

// 入站：能看见这条投递，决定要不要往下走、要不要自己结算
var recordId = new ReceiveMiddlewareConfiguration(
    "record-id",
    (_, next) => context =>
    {
        delivered.TrySetResult(context.Envelope.MessageId);
        return next(context);
    });

builder.Use(recordId);

特性集合：中间件与处理器之间的临时总线

// 上游中间件挂
context.Features.Set(new TenantFeature(header["tenant"]));

// 下游中间件或处理器取（同一个集合，不是快照）
var tenant = context.Features.Get<TenantFeature>();

IFeatureCollection 默认实现是懒建的 FeatureCollection；集合在上下文归还池之前清空， 所以上一条投递挂的东西不会留到下一条。要换集合形状就实现这个接口并替换（接缝表里那条"特性"）。

约定（IConvention）：把重复的登记写成一次

internal sealed class TenantConvention : IConvention
{
    public string Name => "tenant";
    public void Apply(IMessagingBuilder builder) => builder.Use(TenantMiddleware.Create());
}

services.AddSingleton<IConventionRegistry>(new ConventionRegistry(new TenantConvention()));

约定在 BuildAsync 里、编译图之前应用，顺序＝注册顺序。数组里有 null 当场 ArgumentException（"空约定在建图期会直接炸"）。不注册 IConventionRegistry 就不跑任何约定。

6. 派发与接收的运行语义

出站

入站

ReceiveEndpoint.RunAsync(ct) 是唯一入口，构造是内部的——端点由运行时按图装配，宿主不该自己 new （那会绕过图上的校验与端点缓存）。

// 并发上限：每端点取一次。注册成 Transient 才各端点一份，单例＝全端点共用同一个预算
services.AddSingleton<IConsumerLimiter>(new ConcurrencyConsumerLimiter(2));

死信：去处，不是删除

结算成 DeadLetter 的投递先落到那条线的死信地址，IDeadLetterSink 是额外处置：

public interface IDeadLetterSink
{
    ValueTask DeadLetterAsync(IDeliveryLease lease, Exception? failure, CancellationToken ct);
}

7. Saga

三件，各管一件事：SagaDefinition<TState>（初始状态与转移）、SagaExecution（编排）、 ISagaStore（持久化 SPI，必须带状态版本戳与租约，否则接口不成立）。内核只认字节： 状态以 byte[] 进出，序列化是读写器的事，与发件箱同一条口径。

builder.AddSaga(new SagaDefinition<OrderState>("order-saga", OrderState.Empty)
    .On<OrderOpened>(openedIdentity, static e => e.OrderId, static (s, e) => s with { Id = e.OrderId })
    .On<OrderPaid>(paidIdentity,     static e => e.OrderId, static (s, e) => s with { Paid = e.Amount })
    .WhenCompleted(static s => s.Paid >= s.Total));   // 可选：不声明就永不自动清

services.AddSingleton<ISagaStore>(new InMemorySagaStore());   // 或 AddMessagingEntityFrameworkCore<AppDbContext>()

8. 请求与回投

public sealed record GetOrderStatus(string OrderId) : IEventRequest<OrderStatus>;

public sealed class GetOrderStatusHandler(...) : IEventRequestHandler<GetOrderStatus, OrderStatus>
{
    public async ValueTask<OrderStatus> HandleAsync(GetOrderStatus request, CancellationToken ct) => ...;
}

builder.AddRequestHandler<GetOrderStatusHandler, GetOrderStatus, OrderStatus>();

var status = await bus.RequestAsync<GetOrderStatus, OrderStatus>(
    new GetOrderStatus("A-1"), new SendOptions { RequestTimeout = TimeSpan.FromSeconds(5) }, ct);

失败回投：别让你干等到超时

处理器终局失败（策略判死信或丢弃）且结算真的落地之后，往 Envelope.Response 回投一条故障： 载荷为空，原因写在保留头 sparkdo.fault / sparkdo.fault.message 里，请求方当场拿到 MessagingFaultException（FaultedMessageId、FailureType；原因句子在 Message 里）。

9. 扩展接缝

每条缝都是「契约 + 默认实现 + 替换路径」三者齐备，否则不保留。默认在核心，替换方式是从容器取： 宿主注册什么就用什么，取不到就用默认。

AddRequestHandler 之类都返回 IMessagingBuilder，可链式；SparkdoMessaging.CreateBuilder(names, serializers) 是绕开 DI 的入口（测试与非 MS DI 宿主用）。

10. 观测

只用 System.Diagnostics：一个 ActivitySource + 一个 Meter，都在 MessagingObservability。 核心只发事件，采集是消费方的事——不引 OpenTelemetry 包，也不假定任何导出器。

名字都是常量（MessagingObservability.*、MessagingTags.*），别在采集侧写裸字符串。

// 没有监听者时这些名字一次都不产生：创建 Activity 前先问 HasListeners()，记指标前先问仪器 Enabled
builder.Services.AddOpenTelemetry().Tracing(t => t.AddSource("Sparkdo.Messaging"))
                                          .Metrics(m => m.AddMeter("Sparkdo.Messaging"));

链路跨进程接得上：派发时把当前 span 的 W3C 标识（traceparent，带 tracestate 就一并）写进 信封那格保留头；接收侧从字节里读出来当 span 之父——投递跑在接收循环自己的线程上， 本地 Activity.Current 里没有派发那条 span。两处都在 HasListeners() 之后：没人监听就一个字节都不多写。

11. 失败与排查（逐字文案）

配置期的错全部在建图时一次报全，文案都带"修法"。下表是本域的全部失败面，出现频率从高到低：

运行期还会撞到两条与传输有关的：

12. 声明而非约束（公开面里今天不成立的几项）

逐条实测过才写在这里。"有 API 但走不通"与"没有 API"是两种缺口，分开列：

TransportCapability.Batch、IBatchEventHandler、AddBatchHandler 已在 2026-10-16 摘掉：它的"批" 是一条数组消息的载荷，与传输批量收发无关，声明它也没有任何契约问责。真要原生批量，得先有 "成批取投 + 一批一次结算"的端口，连同一道能力档用例一起回来。

13. 出证与门禁（仓库侧）

设计口径在 src/messaging/docs/messaging-next-design.md；性能读数在 src/messaging/docs/perf-baselines.md； 容量与故障读数的跑法在 src/messaging/docs/baselines/。

不做什么

没有 DI 之外的容器集成、没有属性/约定扫描（除了生成器那一条编译期发现）、没有消息版本协商、 没有跨服务路由表、没有内建调度器（走适配包）、没有 OpenTelemetry 依赖、没有"自动拓扑清理"—— 队列与表都是部署物，消费者退出不该拆别人的信箱。