不能直接裸写std::thread拼流水线，因缺乏有界缓冲区、反压机制和生命周期同步；需用blocking_queue实现stage间通信，统一stage签名并支持级联关闭与异常处理。
为什么不能直接用 std::thread 拼流水线
直接裸写 std::thread 启动多个阶段，很容易卡死或数据错乱——因为没统一的缓冲区控制、没反压机制、没生命周期同步。比如 stage2 消费太快，stage1 还没产出，就只能轮询或 sleep，浪费 CPU；反过来 stage1 疯狂 push，stage2 来不及处理，队列爆内存。这不是线程问题，是背压和协调问题。
- 必须为每个 stage 间引入有界缓冲区（
std::queue+std::mutex+std::condition_variable），且容量可配 - 每个 stage 线程需支持“空闲时阻塞等待”，而非忙等
- pipeline 启动/停止必须原子：不能出现 stage3 已退出，stage2 还往它队列里 push 的情况
用 blocking_queue 实现 stage 间通信
别自己从零写锁队列，用封装好的 blocking_queue —— 它内部已处理好生产者-消费者等待逻辑。核心行为：push 时若满则阻塞，pop 时若空则阻塞，析构时自动唤醒所有等待线程并拒绝新操作。
示例结构：
template<typename T>
class blocking_queue {
std::queue<T> q_;
mutable std::mutex mtx_;
std::condition_variable not_empty_;
std::condition_variable not_full_;
size_t capacity_;
public:
blocking_queue(size_t cap) : capacity_(cap) {}
void push(T&& item);
T pop(); // 阻塞直到有数据
bool try_pop(T& out); // 非阻塞
bool closed() const;
};
- capacity_ 设为 1 时退化为同步 channel（类似 Go chan）；设为较大值可缓解阶段间速度差
-
pop()必须用 unique_lock + wait，不能用try_lock轮询 - 析构时调用
not_empty_.notify_all()并置closed_ = true，让所有阻塞线程能安全退出
如何定义 stage 并串联成 pipeline
每个 stage 是一个可调用对象（lambda / functor / function pointer），接受输入类型、返回输出类型。pipeline 不关心具体业务逻辑，只管调度和传数据。
立即学习“C++免费学习笔记（深入）”；
组合式C++代码评审方案，融合静态分析、AI推理、多轮迭代评审和C++专项检查，适用于PR审查、增量代码审查、全项目评审和代码质量评分，触发词包括review cpp、cpp代码评审、C++review、代码审查。
关键设计点：
- stage 类型签名统一为
std::function<Output(Input)>，但实际中 Input/Output 可能是std::optional<T>或带错误码的 wrapper，用于表达“结束信号” - pipeline 构造时传入 stage 列表，自动创建对应数量的
blocking_queue（n 个 stage → n−1 个队列） - 每个 stage 线程循环执行：
input_q.pop()→ 计算 →output_q.push(result)；若 pop 返回空（如std::nullopt），则向下游发终止信号并退出
启动后，首 stage 从外部 push 初始数据，末 stage 的输出由用户通过 final_queue.pop() 获取。
stop() 必须触发级联关闭
调用 pipeline.stop() 不能只 join 线程——必须先通知第一个 stage 停止接收新数据，再逐级传递 shutdown 信号，否则中间队列可能卡住未消费项，导致后续 stage 永远阻塞在 pop 上。
- 推荐方案：每个
blocking_queue提供close()方法，设 flag + notify，之后所有push失败，pop在空时立即返回std::nullopt - pipeline.stop() 顺序：先
input_queue.close()，再join()所有 stage 线程；线程内检测到 pop 返回 nullopt 就主动 close 下游 queue 并 return - 务必确保所有线程都已 join 完，才析构 queues 和 stages，否则析构时可能访问已销毁的 mutex
最易被忽略的是异常路径：某个 stage 函数 throw，必须保证异常不逃逸出线程函数，且能触发下游关闭——通常用 std::current_exception() 捕获后存入特殊 error token 推向下一级。
