07-EFT.WP.Core.Threads v1.0 | 第3章 消息通道与背压

目录 ／ 文档-技术白皮书 ／ 07-EFT.WP.Core.Threads v1.0

第3章 消息通道与背压

I. 范围与目标

• 统一定义通道 chan、消息 msg、容量 cap、队列长度 q_len、背压 bp 与确认 ack 的语义与度量。
• 给出稳态条件、丢弃策略、批处理与 ACK 协议、限流协同与闭环调节流程 Mx-2。
• 产出公设 P73-*、最小方程 S73-*、接口绑定到 I70-3（见本卷实现段落）。

II. 通道模型与消息格式

1. 通道基本形制
   • chan = {name, mode, cap, policy, metrics}，其中 mode ∈ {"spsc","mpsc","spmc","mpmc"}。
   • 队列状态：q_len(t)、cap、age_head(t)、blocked_put/blocked_get。
2. 消息最小字段
   • msg = {eid, ts, tau_mono, idemp_key?, sem, ttl?, deadline?, payload_ref, trace_ctx?}。
   • 投递语义 sem ∈ {"at_most_once","at_least_once","exactly_once*"}（* 通过去重近似实现）。

III. 稳态与队列度量（S73 系列）

1. 核心定义与恒等
   • 利用率：rho = lambda / mu；稳态必要条件：rho < 1。
   • Little 定律：L = lambda * W，对应队列：L_q = lambda * W_q。
2. 近似公式（M/M/1 场景）
   • S73-1：W_q approx rho / ( mu * (1 - rho) )。
   • S73-2：W approx 1 / ( mu - lambda )。
3. 批处理修正
   批大小 b >= 1 时有有效服务率：mu_eff approx mu * b；等效队长：q_len_eff = ceil( q_len / b )。

IV. 公设与约束（P73-*)

• P73-1（容量声明）：任一 chan 必声明 cap；cap > 0 且为常量或分段常量。
• P73-2（稳定性）：若采用稳态近似，需记录 lambda 与 mu 的估计来源，并满足 rho < 1 approx stability。
• P73-3（背压闭环）：bp 必为 q_len/cap 与 W_q 的单调非减函数；阈值与动作成对声明。
• P73-4（溢出可见）：当触发丢弃或降级，必须发出 eid 级审计事件并计入 TS.drop.count。
• P73-5（ACK 时界）：ACK 模式必须声明 ack_timeout 与最大重投 retries，并满足 W_retry <= timeout * (retries + 1) + J_total。
• P73-6（幂等保障）：sem!="at_most_once" 时，消费者处理函数需满足 f(x; idemp_key) = f(x; idemp_key) 且声明 Delta_t_dedup。

V. 背压信号与函数定义

1. 背压强度
   • 定义：bp = min( 1 , max( 0 , a * ( q_len / cap ) + b * ( W_q / W_q_target ) ) )，a,b ∈ [0,1]。
   • 水位：wm_low < wm_high，当 q_len/cap >= wm_high 触发限流或丢弃；低于 wm_low 放宽。
2. 控制动作
   • 输入端：shaping(rps)、batch_size↑、priority demotion。
   • 输出端：prefetch↓、consumer K_thr↑、worker 扩容。
3. 反馈频率
   更新周期 T_bp 建议满足 T_bp << W_q，避免欠/超调。

VI. 溢出与丢弃策略

1. 策略族与适用性
   • tail_drop：丢弃新到消息；对有序通道常用。
   • head_drop：丢弃最旧消息；用于时效性场景（最小化过期负担）。
   • random_drop：近似 RED；平滑拥塞。
   • shed_by_key(id)：按 idemp_key 维度限流或丢弃高噪声源。
2. 选择准则
   • 若优化 P99(latency)：优先 head_drop 或 random_drop。
   • 若优化准确性：优先 tail_drop 或进入阻塞（需评估上游 rho）。

VII. 批处理与 ACK 语义

• 生产批
  chan_put(ch, batch{msg_1..msg_b})；达到 b_target 或 timeout_batch 则出队。
• 消费批
  chan_get(ch, max_batch) 返回 [msg...]；回写 ack(batch_id) 或逐条 ack(eid)。
• 失败与重投
  ack_timeout 到期：p_retry 触发重投；sem="at_least_once" 需结合去重窗口 Delta_t_dedup。
• 顺序与幂等
  对 order=strict 的通道，ACK 必须按序或携带 offset，并保证 hb 下的有序可见。

VIII. 投递语义与去重

1. 语义对照
   • at_most_once：不重试，p_drop 可非零；最低延迟。
   • at_least_once：允许重复，结合 idemp_key 去重。
   • exactly_once*：通过 idemp_key + 幂等存储近似。
2. 去重窗口
   dedup_table 存 (idemp_key, ts_last)；规则：if tau_mono - ts_last <= Delta_t_dedup then drop duplicate。

IX. 端到端时序与年龄控制

• 年龄与过期
  定义消息年龄：age(msg,t) = t - ts；过期条件：age(msg,t) > ttl。
• 延迟预算拆分
  budget_total = W_q_in + w(proc) + W_q_out；目标 P99(budget_total) <= SLO。
• 到达时引用
  如使用传播到达时 T_arr 校准跨域时序，采用两口径：
  T_arr = ( 1 / c_ref ) * ( ∫ n_eff d ell ) 与 T_arr = ( ∫ ( n_eff / c_ref ) d ell )；报告 delta_form 差异。

X. 限流协同与稳定条件

• 限流器
  令牌桶：rate_limiter(name, rps, burst)，消费路径使用 limit_acquire。
• 稳定性复核
  若 bp≈1 持续 > T_hold，则下调 rps := max( rps_min , rps * k_down )；当 bp<wm_low 连续 > T_relax，上调 rps := min( rps_max , rps * k_up )（AIMD）。
• 多级链路
  链式通道的瓶颈 mu* 决定全局 rho*：rho* = max_k ( lambda_k / mu_k )。

XI. 容量规划与基线

• 基线估计
  cap >= ceil( lambda_peak * T_absorb )，其中 T_absorb 为需吸收的抖动窗口。
• 目标设置
  wm_low ≈ 0.5，wm_high ≈ 0.8；初始 a=b=0.5；批大小 b_target 依据 c(e) 与 w(v) 比值调谐。

XII. 接口绑定与不变量（I70-3 对齐）

1. 绑定摘要
   • chan_open(name:str, cap:int, mode:str="mpmc") -> ChanRef（需校验 cap>0）。
   • chan_put(ch, msg|batch, key?, timeout?) -> bool（超时计入 TS.timeout.count）。
   • chan_get(ch, max_batch:int=1, timeout?) -> list（空超时计入 TS.timeout.count）。
   • set_backpressure(ch, strategy:dict) -> None（策略含 wm_low, wm_high, actions[]）。
2. 不变量
   q_len <= cap；bp ∈ [0,1]；ACK 串行或幂等；trace_link(span,eid) 完整。

XIII. 可观测性与 SLI

1. 指标映射 TS.*
   • TS.chan.cap、TS.chan.q_len、TS.backpressure.level、TS.drop.count、TS.retry.count、TS.timeout.count。
   • 延迟：TS.sli.queue_time_ms、TS.sli.latency_ms、TS.sli.p99_ms；吞吐：TS.sli.qps；错误：TS.sli.err_rate。
2. 合规检验
   校核 check_dim( L - lambda * W ) = 0；窗口为 SLA_window。

XIV. 契约与测试矩阵

1. 必测用例
   • rho < 1 稳态通过；wm_high 触发限流；head_drop/tail_drop/random_drop 行为正确。
   • at_least_once 去重有效；ack_timeout 重投不破坏顺序约束；Delta_t_dedup 生效。
   • 老化与过期：ttl 到期自动丢弃并审计。
2. 断言结构（示例）
   assert_thread_contract(G, tests=[ {"type":"queue_stability","chan":"ingress","rho_lt":1.0}, {"type":"slo","p99_ms_le":SLO}, {"type":"drop_policy","mode":"head_drop"} ])。

XV. 运行流程 Mx-2（背压闭环）

• chan_open 并设置 cap、wm_low/wm_high、strategy。
• 周期性采样 q_len, W_q, bp，计算 bp = f(q_len, cap, W_q)。
• 若 bp >= wm_high：执行 actions_high = {limit rps, enlarge batch, prefetch↓}。
• 若 bp <= wm_low：执行 actions_low = {relax rps, shrink batch, prefetch↑}。
• 监控 TS.drop.count/err_rate/P99，必要时触发 shed_by_key 或扩容 K_thr。
• assert_thread_contract 校核并落库 Trace 与 TS.*。

XVI. 出厂条件与交付件

• 通过 P73-*、S73-* 与 Mx-2 全流程；TS.* 指标完整；溢出与 ACK 审计可追溯。
• 交付通道规范、容量与阈值表、丢弃与重试策略卡、去重配置与回归基线。