Crossbeam 有界通道流水线

示例说明

无界通道在消费端变慢时无限吃内存；有界通道的 send 在队满时阻塞——上游自动匹配下游速率，背压沿流水线反向传导。crossbeam 的 select! 宏让线程同时等待多个事件：输入到达、通道关闭或中断信号，任一就绪即处理——优雅停机时先排空再退出，数据零丢失。

技术要点

• 有界通道满即阻塞形成背压
• select 多路等待输入与中断
• drop 发送端是标准关闭信号

代码解读

三个线程三阶段：生产者按五毫秒节奏产出原始行；解析线程 select 等待输入或中断信号，收到中断后用 try_iter 排空残余再退出；聚合线程迭代输出通道累加。每阶段的通道容量都是八，慢阶段自动让上游降速。drop 发送端让下游迭代器自然结束。

使用建议

中断演示的排空逻辑要在真实场景与事务边界对齐。通道容量按阶段处理速率差设定。跨线程错误传递考虑 channel 传 Result。

代码示例

use crossbeam_channel::{bounded, never, select};
use std::thread;
use std::time::Duration;

fn main() {
    let (raw_tx, raw_rx) = bounded::<String>(8);
    let (parsed_tx, parsed_rx) = bounded::<u64>(8);

    // 阶段一：读取与清洗
    thread::spawn(move || {
        for i in 0..20 {
            let line = format!("metric:{}:{}", i % 5, i * 10);
            if raw_tx.send(line).is_err() {
                break;
            }
            thread::sleep(Duration::from_millis(5));
        }
        drop(raw_tx);
    });

    // 阶段二：解析与转换（select 监听中断信号）
    thread::spawn(move || {
        let (stop_tx, stop_rx) = bounded::<()>(1);
        thread::spawn(move || {
            thread::sleep(Duration::from_millis(200));
            let _ = stop_tx.send(());
        });

        loop {
            select! {
                recv(raw_rx) -> line => match line {
                    Ok(text) => {
                        let value: u64 = text
                            .rsplit(':')
                            .next()
                            .and_then(|v| v.parse().ok())
                            .unwrap_or(0);
                        if parsed_tx.send(value).is_err() {
                            break;
                        }
                    }
                    Err(_) => break,
                },
                recv(stop_rx) -> _ => {
                    println!("interrupt requested, draining remaining input");
                    for leftover in raw_rx.try_iter() {
                        let _ = parsed_tx.send(parse_value(&leftover));
                    }
                    break;
                }
                recv(never()) -> _ => unreachable!(),
            }
        }
        drop(parsed_tx);
    });

    // 阶段三：聚合消费
    let mut total: u64 = 0;
    for value in parsed_rx.iter() {
        total += value;
    }
    println!("pipeline total: {total}");
}

fn parse_value(text: &str) -> u64 {
    text.rsplit(':').next().and_then(|v| v.parse().ok()).unwrap_or(0)
}