Rust 队列满了怎么办：这是产品决策，不是技术细节

凌晨三点的 OOM，和一条没人做过的决策

内存曲线爬了六个小时，随着一次下游 API 抖动陡直向上，最后停在 OOM。heap dump 里是几百万个 Sample，整整齐齐排在一个 mpsc::unbounded_channel 里——那是它们唯一能去的地方。出事的代码长这样：

let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
tokio::spawn(async move {
    while let Some(s) = rx.recv().await {
        flush_to_downstream(s).await; // 下游一抖动，这里就变慢
    }
});
// 上游：每秒五万个采样点，send 永远成功，因为队列没有底
tx.send(sample).unwrap();

unbounded_channel 用起来太顺手：send 不返回 future、不阻塞，只要 rx 还活着就永远 Ok(())。于是压力从调用点消失了，全部转进内存，等着在最贵的时刻结算。另一类事故对称地发生：有人把 unbounded 换成 mpsc::channel(1024)，然后 tx.send(job).await，下游一慢，所有上游 task 依次停在 send 上，健康检查也排进同一条队——进程活着，但整体停摆。两种翻车的根因是同一个：没有人回答过"队列满了之后应该发生什么"。

四种策略，四种产品语义

“队列满了怎么办"不是实现细节，是产品决策。用户提交的表单和一条埋点采样，在队列满时的正确行为完全不同。可选的策略只有四种：

阻塞（反压）：等，让上游减速——任务系统
拒绝（try_send）：立刻失败——交互请求
丢弃：丢新值或丢旧值——指标采样
降级：换一个更大更慢的地方排队——日志审计

拒绝：对交互请求诚实一点

在线接口的延迟预算以毫秒计。系统过载时把请求闷在队列里拖二十秒再失败，不如第一次就拒绝：

use tokio::sync::mpsc::error::TrySendError;

enum Admission { Accepted, Rejected, ShuttingDown }
fn admit(tx: &tokio::sync::mpsc::Sender<Request>, req: Request) -> Admission {
    match tx.try_send(req) {
        Ok(()) => Admission::Accepted,
        Err(TrySendError::Full(_req)) => Admission::Rejected,      // 满 = 过载：返回 429/503
        Err(TrySendError::Closed(_req)) => Admission::ShuttingDown, // 关机生命周期，不是过载
    }
}

Full 和 Closed 必须分开处理：前者是容量问题，后者是生命周期问题。都记成 “send failed”，事后排查时分不清是过载还是 bug。

丢弃旧值：指标管道只要趋势

指标采样的语义是"最新状态”，丢一百条旧的和丢一条新的，价值完全不同。所以正确动作是覆盖，而不是排队：

use std::collections::VecDeque;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Mutex;
use tokio::sync::Notify;

/// 覆盖式有界队列：满时丢最旧的值，只保留最近 cap 条（消费端是普通的 pop + Notify 等待）
struct OverwritingQueue<T> {
    q: Mutex<VecDeque<T>>,
    cap: usize,
    ready: Notify,
    dropped: AtomicU64,
}

impl<T> OverwritingQueue<T> {
    fn new(cap: usize) -> Self {
        Self { q: Mutex::new(VecDeque::with_capacity(cap)), cap, ready: Notify::new(),
               dropped: AtomicU64::new(0) } }

    fn push(&self, value: T) {
        let mut q = self.q.lock().unwrap();
        if q.len() == self.cap {
            q.pop_front(); // 满了：牺牲最旧的采样，保住最新的
            self.dropped.fetch_add(1, Ordering::Relaxed);
        }
        q.push_back(value);
        drop(q);
        self.ready.notify_one();
    }
}

dropped 这个计数不是装饰。采样覆盖率掉了三成，报表曲线照样平滑——只有丢弃计数能告诉你真相。同样的"只要最新值"语义，后文的 watch 是官方答案。

降级：热通道满了，去冷通道排队

日志、审计这类数据，慢一点可以，丢不行。两层通道各司其职：

use tokio::sync::mpsc::{self, error::TrySendError};
enum Ingest { Ok, Spilled, Dropped }

fn ingest(ev: Event, hot: &mpsc::Sender<Event>, cold: &mpsc::Sender<Event>) -> Ingest {
    let ev = match hot.try_send(ev) {
        Ok(()) => return Ingest::Ok,
        Err(TrySendError::Full(ev)) => ev, // 热通道满：降级而不是拒绝
        Err(TrySendError::Closed(ev)) => ev,
    };
    // cold 由专职写盘 task 消费；容量是"突发预算"，不是无限保险
    match cold.try_send(ev) {
        Ok(()) => Ingest::Spilled,
        Err(_) => { record_final_drop(); Ingest::Dropped } // 两层都满才谈"丢"
    }
}

降级的本质是给数据分级：热通道服务实时语义，冷通道服务最终到达语义。冷通道也必须有界，否则只是把 OOM 往下推了一层。

阻塞可以，但要带表

任务系统确实应该排队——反压是对的。但"等"不等于"无限等"：

use tokio::time::timeout;
enum SubmitError { Closed, Stalled }

async fn submit(tx: &tokio::sync::mpsc::Sender<Job>, job: Job) -> Result<(), SubmitError> {
    // 愿意排队（反压），但不接受永久卡死
    match timeout(std::time::Duration::from_secs(5), tx.send(job)).await {
        Ok(Ok(())) => Ok(()),
        Ok(Err(_)) => Err(SubmitError::Closed),     // rx 全部 drop，见下文
        Err(_elapsed) => Err(SubmitError::Stalled), // 5 秒还满：消费端已经病了
    }
}

一个容易被忽略的细节：timeout 超时后，被 move 进 send future 的 job 直接析构，你拿不回来。如果超时后还需要这个值做补偿，要么传 Arc<Job>，要么改成 try_send 加 sleep 的循环。

选型：五种通道，五种失败模式

策略定了，通道选错一样翻车。Tokio 常用的五种 channel，各自回答一种"队列满"：

mpsc::channel(n)：满了 send().await 等待——阻塞策略的载体
mpsc::unbounded_channel：永远不满，也永远不反压。只有当生产速率存在硬上限（上游已被限流、任务总数有限）时才成立
watch：通道里永远只有最新值，“满"这个状态不存在——配置、行情快照
broadcast：扇出，各消费者独立游标，慢的会 Lagged
oneshot：单值一次性，没有"满”，只有"没等到"

use tokio::sync::watch;

// watch：覆盖式语义是内建的
let (tx, rx) = watch::channel(Config::default());
// 推送端：直接覆盖，两次 changed() 之间的中间值会被合并掉
tx.send_modify(|c| c.debug = true);
// 订阅端：拿到的永远是当前值，不存在积压
let mut sub = rx.clone();
while sub.changed().await.is_ok() {
    let cfg = sub.borrow().clone();
    reload(cfg);
}

use tokio::sync::broadcast::{self, error::RecvError};

let (tx, _) = broadcast::channel::<Quote>(1024);
let mut slow = tx.subscribe();
tokio::spawn(async move {
    loop {
        match slow.recv().await {
            Ok(q) => persist(q).await,
            Err(RecvError::Lagged(missed)) => {
                // 消费太慢，错过了 missed 条；游标已被拉到最旧的可用消息。
                // 行情场景的合理动作：对账一次快照，而不是装作没发生
                resync_from_snapshot().await;
            }
            Err(RecvError::Closed) => break, // 所有 sender 都没了
        }
    }
});

Lagged 不是异常，是 broadcast 的正常语义：环形缓冲只保留最近 capacity 条，谁跟不上谁丢。错的是无视它。

关闭语义是一等公民

上面所有代码反复出现 Closed 和 SendError，值得说清楚：rx 全部 drop 后，send 才返回 Err。这是 channel 的正常退出路径，不是故障：

let (tx, mut rx) = tokio::sync::mpsc::channel::<Job>(64);

tokio::spawn(async move {
    while let Some(job) = rx.recv().await {
        process(job).await;
    } // rx 在 task 结束时 drop
});
// 生产端：send 失败 = 消费者已退出，正确动作是跟着退出
if tx.send(job).await.is_err() {
    cleanup().await;
    return;
}
// 优雅关停的另一半：等所有 rx drop 后再退出进程
tx.closed().await;

对 send 的结果直接 unwrap，等于断言"消费者永远不会退出"——这个前提在服务的生命周期里几乎从不成立。

观测：水位、拒绝数、等待时长

策略要有数字才算落地，三个指标缺一不可：

use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
struct QueueMetrics {
    high_watermark: AtomicUsize, // len() 的历史峰值
    timeout_total: AtomicU64,    // send 等待超时次数
    wait_us_max: AtomicU64,      // 单次 send 的最长等待
}

async fn send_tracked(tx: &tokio::sync::mpsc::Sender<Job>, job: Job,
                      m: &QueueMetrics) -> Result<(), SubmitError> {
    let start = std::time::Instant::now();
    let r = timeout(std::time::Duration::from_secs(5), tx.send(job)).await;
    m.wait_us_max.fetch_max(start.elapsed().as_micros() as u64, Ordering::Relaxed);
    m.high_watermark.fetch_max(tx.len(), Ordering::Relaxed); // 水位：离失控还有多近
    match r {
        Ok(Ok(())) => Ok(()),
        Ok(Err(_)) => Err(SubmitError::Closed),
        Err(_) => {
            m.timeout_total.fetch_add(1, Ordering::Relaxed);
            Err(SubmitError::Stalled)
        }
    }
}

水位看峰值不看均值：均值 10%、峰值贴顶的队列，说明你离失控只差一次突发。

常见坑点

库代码里偷偷用 unbounded

你自己的代码都审过了，依赖返回一个 UnboundedSender——它替你做了"永不拒绝、无限缓冲"的决策。审计时对依赖 grep 一遍 unbounded，看清是谁在做决定。

Lagged 被当成普通日志

broadcast 持续 Lagged 等于持续丢数据，但它看起来只是 info 级别的一行日志。Lagged 频率必须是独立指标，不是日志噪音。

watch 丢中间态

watch 只保留最新值，消费者两次醒来之间合并了 N 次变更。需要每一个中间值（事件流、审计）就换 mpsc 或 broadcast，别指望 watch。

有界队列被 spawn 架空

从队列收一个就 spawn 一个，队列是不积压了，但同时运行的任务数没了上限。限并发是 Semaphore 的职责，之前写 Tokio 背压时展开过，这里不重复。

总结

unbounded 的问题不是性能，是它替你回答了"队列满了怎么办"——用内存回答。这个问题应该由你回答：交互请求拒绝、指标丢弃旧值、任务阻塞加超时、日志降级到冷通道。然后按语义选通道，把关闭当正常路径，把水位和拒绝数打出来。

队列满了的那一刻，用户感受到的不是你的代码，而是你替他们做的决定。