Sparkdo.Messaging.Reliability
3.0.0-preview.1
dotnet add package Sparkdo.Messaging.Reliability --version 3.0.0-preview.1
NuGet\Install-Package Sparkdo.Messaging.Reliability -Version 3.0.0-preview.1
<PackageReference Include="Sparkdo.Messaging.Reliability" Version="3.0.0-preview.1" />
<PackageVersion Include="Sparkdo.Messaging.Reliability" Version="3.0.0-preview.1" />
<PackageReference Include="Sparkdo.Messaging.Reliability" />
paket add Sparkdo.Messaging.Reliability --version 3.0.0-preview.1
#r "nuget: Sparkdo.Messaging.Reliability, 3.0.0-preview.1"
#:package Sparkdo.Messaging.Reliability@3.0.0-preview.1
#addin nuget:?package=Sparkdo.Messaging.Reliability&version=3.0.0-preview.1&prerelease
#tool nuget:?package=Sparkdo.Messaging.Reliability&version=3.0.0-preview.1&prerelease
Sparkdo.Messaging.Reliability
可靠性设施的默认实现:指数退避重试、进程内 Inbox / Outbox / Saga 存储。
端口(IRetryPolicy / IMessageInbox / IMessageOutbox / ISagaStore / IDeadLetterSink / IWorkUnit)
全在 Sparkdo.Messaging 内核里;本包只提供实现,宿主可以整体替换。
| 项 | 值 |
|---|---|
| 包 | Sparkdo.Messaging.Reliability |
| 交付框架 | net11.0 |
| 外部 NuGet 依赖 | 无 |
| 入口 | AddSparkdoMessagingReliability(Action<RetryPolicyOptions>?) |
| AOT / 裁剪 | 兼容 |
接线
builder.Services
.AddSparkdoMessagingReliability(options =>
{
options.MaxAttempts = 5; // 默认 5(含首次尝试)
options.BaseDelay = TimeSpan.FromSeconds(1); // 默认 1s:第一次重试的基础延迟
options.MaxDelay = TimeSpan.FromMinutes(1); // 默认 1min:指数增长到这里封顶
options.JitterRatio = 0.25; // 默认 0.25:±25% 抖动,0 表示不抖
})
.AddSparkdoMessaging(serializers, bus => bus
.AddTransport(new RabbitMQTransport(options), TransportDefault.Default)
.Use(InboxMiddleware.Create(sp => sp.GetRequiredService<IMessageInbox>(), tenant: "orders", handlerId: "sparkdo.billing.order-placed-handler")))
.AddSparkdoMessagingHosting();
一次登记全部装好,且一律 TryAdd——先登记的赢,宿主要换实现就在它之前注册,或者干脆不叫这个方法:
| 容器里出现 | 生命周期 | 来源 |
|---|---|---|
IRetryPolicy = ExponentialRetryPolicy |
Singleton(实例工厂读 RetryPolicyOptions) |
本包 |
IMessageInbox = InMemoryMessageInbox |
Singleton | 本包 |
IMessageOutbox = InMemoryMessageOutbox |
Singleton(具体类型也单独登记一份,CreateWorkUnit() 要从它取) |
本包 |
ISagaStore = InMemorySagaStore |
Singleton | 本包 |
WorkUnitAccessor |
Scoped | 本包 |
IDeadLetterSink |
—— | 本包不装,留给宿主或适配包 |
IWorkUnit |
—— | 本包不装:内存档的单元由 InMemoryMessageOutbox.CreateWorkUnit() 现取 |
ISagaStore 这一条要盯住:不显式登记 <ISagaStore, 真实存储>,Saga 的状态就落在内存里,
进程一重启就没了,而不会报错。换 EF 档就是 AddMessagingEntityFrameworkCore<AppDbContext>()
(它登记的也是 TryAdd,所以调用顺序决定谁赢)。
重试:默认那条曲线
var delay = BaseDelay * 2^(Attempt - 1) // 位移封在 30,防溢出
封顶到 MaxDelay
× (1 + 随机[-1,1) × JitterRatio) // JitterRatio ≤ 0 时不抖
| 事 | 语义 |
|---|---|
| 谁决定 | 接收端点在处理器抛异常之后叫 IRetryPolicy.NextAsync(new RetryState(Attempt, Failure, Elapsed), ct) |
Attempt 从哪来 |
(Envelope.DeliveryCount ?? 0) + 1——跨进程的那一份由传输递增(RabbitMQ 用代理的 x-delivery-count,Postgres 用表里的计数,InMemory 自己数),不是进程内计数器 |
Attempt >= MaxAttempts |
判 DeadLetter,不再排延迟 |
Retry + Delay |
端点就地 Task.Delay(delay, ct) 之后再结算成 Retry(回队)。响应取消是停机的一部分 |
Discard |
也结算成 DeliveryOutcome.DeadLetter,差别只在不叫 IDeadLetterSink——结论只有两个出口,丢弃是"不去死信那条线之外再处置" |
| 策略自己抛 | 兜底按 Retry 处置并留痕:一条坏策略不该把端点打死 |
| 没注册策略 | 内核用默认 ExponentialRetryPolicy(上面那四个默认值),不需要本包也能跑 |
| 死信出口失手 | IDeadLetterSink 抛异常只记警告,不改这条投递的结论 |
Inbox:幂等中间件怎么摆
端口在核心(IMessageInbox),实现(InMemoryMessageInbox)在这里,登记在管道上:
bus.Use(InboxMiddleware.Create(
services => services.GetRequiredService<IMessageInbox>(),
tenant: "orders",
handlerId: "sparkdo.billing.order-placed-handler"));
| 参数 | 用法 |
|---|---|
factory |
按本条投递的作用域取端口,不要捕获构建期那一个实例(EF 档的实现是 scoped,捕获=所有投递共用一个 DbContext) |
tenant |
进幂等键。多租户共用一条队列时它是唯一的分隔;单租户给个稳定常量就行 |
handlerId |
进幂等键。每个消费方一个,不能留空(空串当场抛);换处理器名等于换幂等命名空间 |
键是 (Tenant, MessageId, HandlerId) 三元组——"同一条消息在同一个处理器上只跑一次",
不是"同一条消息只跑一次"。三家订阅者各自都要跑,各自记一格。
| 判定 | 会发生什么 |
|---|---|
Proceed |
进处理器;返回后 CompleteAsync;抛异常则 AbandonAsync 后原异常照冒(吞掉它就等于把失败咽下去)。放弃失败只记日志,不顶掉原异常 |
AlreadyCompleted |
不进处理器,上层照常结算成 Ack——副作用不重跑 |
AlreadyInFlight |
抛 InboxInFlightException 走失败分支,让重试策略决定重投节奏 |
顺序要紧:Use(workUnit) 与 Use(inbox) 谁先登记谁在外层。想让"提交成功"等于"幂等标记落库",
就把工作单元放外层(先开单元,再判幂等,一起提交)。
Outbox:内存档只到"单进程能用"
var outbox = provider.GetRequiredService<InMemoryMessageOutbox>(); // 具体类型也登记了,为了取单元
var unit = outbox.CreateWorkUnit();
await outbox.EnqueueAsync(unit, envelope, payload, ct); // 攒在单元里
await unit.CommitAsync(ct); // 一次性放出去
| 事 | 语义 |
|---|---|
| 认领是真的认领 | ClaimPendingAsync 把取走的一批从待发挪进 _claimed,第二个调用者看不见。两台中继同跑也不会各发一遍 |
| 发不出去 | ReleaseAsync 放回待发,下一轮立刻能重来;MarkSentAsync 才是终态 |
| 载荷复制 | EnqueueAsync 当场拷一份字节——载荷租约在提交之前就归还了,发件箱必须持有自己的 |
| 单元配对 | 传进来的 IWorkUnit 不是本发件箱造的就当场拒绝("工作单元必须由本发件箱创建") |
| 没有中继 | 本包不驱动 ClaimPendingAsync:没有任何后台循环。要么自己起一个(await foreach (var m in outbox.ClaimPendingAsync(ct)) → IRawPublisher.PublishRawAsync → MarkSentAsync),要么换 EF 档(它的 AddMessagingEntityFrameworkCore 附带一个 hosted service 中继) |
| 崩溃 | 内存里的未发消息随进程一起没。这是内存档的边界,不是缺陷的兜底 |
"要不要自己写循环"这一条在 EF 档不存在(那边中继是自带的)。所以内存档的合理用处是
用例与单进程旁路,生产路径请直接上 EF 档或自己实现 IMessageOutbox。
Saga 存储:内存档
services.AddSingleton<ISagaStore>(new InMemorySagaStore()); // 时钟可注入
services.AddSingleton<ISagaStore>(new InMemorySagaStore(fakeTimeProvider)); // 用例把"租约过期"推到眼前
| 事 | 语义 |
|---|---|
| 租约 | TryBeginAsync(sagaId, sagaType, owner, leaseDuration, ct) 拿到才算持有;ReleaseAsync 放手 |
| 写回 | SaveAsync(lease, state, ct) 返回 false = 版本被别人抢了(CAS)。内核把它翻成 Contended → SagaContentionException → 按可重试结算 |
| 完成标记 | 进程内档不落终态标记:本包没有清理服务,存了也没人读。带 completed 的四参 SaveAsync 用默认实现退回三参,所以内存档的 Saga 永远不会被自动清 |
ReadAsync(sagaId, ct) |
给用例断言用的读口(返回 SagaRecord?(State, Version))。生产路径只走 ISagaStore 的三个动作 |
| 时钟 | 构造参数 clock 是 TimeProvider?,缺省用系统的。测试里让租约"过期"不必真等 |
| 跨进程 | 不支持:状态与租约都在这个进程的堆里。跨重启请用 Sparkdo.Messaging.EntityFrameworkCore 的 EfCoreSagaStore |
什么时候必须离开本包
| 需求 | 去处 |
|---|---|
| 幂等/发件箱/Saga 跨进程或跨重启 | Sparkdo.Messaging.EntityFrameworkCore(同事务,provider 无关) |
| 死信要落库/落盘/报警 | 自己实现 IDeadLetterSink(本包不装),或读 {scheme}://dead/{消息身份}/{订阅者} 那条线 |
| 中继循环有人管 | EF 档(自带 hosted service),或自己起一个 |
| 按批重试/按异常类型分流 | 自己实现 IRetryPolicy(state.Failure 就是那条异常,RetryDecision 只有结论与延迟两格) |
| Product | Versions Compatible and additional computed target framework versions. |
|---|---|
| .NET | net11.0 is compatible. |
-
net11.0
- Microsoft.Extensions.ObjectPool (>= 10.0.8)
- Sparkdo.Messaging (>= 3.0.0-preview.1)
NuGet packages
This package is not used by any NuGet packages.
GitHub repositories
This package is not used by any popular GitHub repositories.
| Version | Downloads | Last Updated |
|---|---|---|
| 3.0.0-preview.1 | 59 | 10/30/2026 |