PubSub 收到 Lagged

主线位置：跨任务主线的「每个订阅者都要收到」变体。PubSubChannel 的队列、每条消息的剩余读者计数和 Lagged 的算法见 Signal、Watch 与 PubSub。

flowchart TB
  p["publish_immediate"] --> full{"队列满了"}
  full -->|否| push["放到队尾"]
  full -->|是| pop["pop_front 丢掉最旧"]
  pop --> push
  sub["慢订阅者读"] --> cmp{"要读的编号还在吗"}
  cmp -->|在| msg["Message(v)"]
  cmp -->|已被挤掉| lag["Lagged(n)"]

实验 crate exp-embassy/pitb-std，bin p16_pubsub（release）；M2 Max、macOS 27、rustc 1.99.0。

现象

一个采样任务每 10 ms 用 ImmediatePublisher 发一个值（就像 ADC 中断里那样，不能等），一个记日志的订阅者每处理一条要 35 ms，容量 4：

static SAMPLES: PubSubChannel<CriticalSectionRawMutex, u32, 4, 2, 1> = PubSubChannel::new();

#[embassy_executor::task]
async fn producer_immediate() {
  let p = SAMPLES.immediate_publisher();
  for v in 0..20 {
    p.publish_immediate(v); // e.g. from an ADC interrupt: never waits
    Timer::after_millis(10).await;
  }
}

#[embassy_executor::task]
async fn slow_logger() {
  let mut sub = SAMPLES.subscriber().unwrap();
  // ...
  match sub.next_message().await {
    WaitResult::Message(v) => { /* 记下 v */ Timer::after_millis(35).await; }
    WaitResult::Lagged(n) => { /* 记下丢了 n 条 */ }
  }
}

  slow logger saw: 0 1 2 Lagged(2) 5 Lagged(2) 8 Lagged(2) 11 Lagged(2) 14 Lagged(1) 16 17 18 19
  11 messages, 9 lost, done at 417 ms

20 条里只收到 11 条。如果代码只写了 if let WaitResult::Message(v) = sub.next_message().await，或者用的是 next_message_pure()，这 9 条就无声无息地没了。

出处

publish_immediate 从不等待，队列满了就先丢最旧的：

    fn publish_immediate(&mut self, message: T) {
        // Make space in the queue if required
        if self.queue.is_full() {
            self.queue.pop_front();
        }

        // This will succeed because we made sure there is space
        self.try_publish(message).ok().unwrap();
    }

它不看队首那条还有几个订阅者没读。每个订阅者自己记着下一条要读的编号；get_message 发现这个编号比队首还小，就返回 WaitResult::Lagged（差几条），并把订阅者的编号跳到队首。上面的输出：生产者每 10 ms 一条，订阅者每 35 ms 才回来读一次，这段时间里来了 3 到 4 条，容量 4 的队列装不下，队首的被挤掉，于是隔一条就是一个 Lagged(2)。最后 4 条（16 到 19）发完时生产者停了，队列不再被挤，订阅者慢慢读完。

next_message_pure 就是 Lagged(_) => continue，只适合「丢了也无所谓」的场合。订阅者当 Stream 用时也一样，遇到 Lagged 自动跳过。

修法

先决定丢消息能不能接受，再选：

能丢，但要知道丢了多少：像上面那样匹配 WaitResult::Lagged(n)，计数、报警或请求重发。这是 publish_immediate 的正确用法，中断里发布只能这样。
不能丢，发布方可以等：用 publisher() 拿一个 Publisher，publish(v).await 在队列满时等最慢的订阅者读完队首（p16_pubsub fix）：
```
let p = SAMPLES.publisher().unwrap();
for v in 0..20 {
  p.publish(v).await; // waits for the slowest subscriber
  Timer::after_millis(10).await;
}
```
```
  waiting publisher done after 591 ms
  slow logger saw: 0 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19
  20 messages, 0 lost, done at 774 ms
```
一条都没丢，代价是生产者被拖慢：20 条本来约 200 ms 发完，现在要 591 ms，速度由最慢的订阅者决定。Publisher 占 PUBS 的名额，publish 是 async 的，不能在中断里用。
只是偶尔一阵突发：把 CAP 加大到能装下一次突发。PubSubChannel 的内存是 CAP 个 (T, usize)，在 static 里，编译期定死。

怎么确认没中招

grep -rn 'next_message_pure\|try_next_message_pure\|WaitResult::Message' src/：每一处都问一句「这里丢消息可以吗」。
用 publish_immediate 的通道，订阅者里应该有 WaitResult::Lagged(n) 分支，并且做了点什么（至少计数）。
测试时给订阅者人为加一段延时，看计数是否增长。

小结

publish_immediate 满了就 pop_front，不管有没有人没读；慢订阅者下次读到 Lagged(n)。
next_message_pure 和 Stream 接口把 Lagged 吞掉，丢失不可见。
不能丢就用 Publisher::publish().await（发布者等最慢的订阅者），或加大 CAP；能丢就处理 Lagged(n)。

PubSub 收到 Lagged

现象

出处

修法

怎么确认没中招

小结

💬 关联讨论

登录