1. 从"裸奔"的共享变量到消息队列：先看一个真实案例
我接手过一个采集程序，最初版本用的是最直白的做法：一个生产者线程不断往全局链表里塞数据，消费者线程定时醒来遍历链表，为了防冲突给链表挂了一把大锁。前两个月一切正常，后来数据量上来，问题陆续暴露：消费者为了不漏数据，只能高频轮询，CPU空转严重；生产者偶尔拿到锁却发现链表被消费者清空了，数据没丢但时延忽高忽低；最头疼的是业务逻辑后来要求"同一批数据必须一起处理"，链表+锁完全没法表达这种批量语义。
后面被迫重构，把"共享变量+锁"这种裸奔方式换成了消息队列。这里说的消息队列不是中间件那种跨进程的MQ，而是线程内部、进程地址空间里的消息传递机制。改完以后最直观的变化：生产者只需要入队，消费者只需要出队，两边完全不感知对方的存在，同步、解耦、削峰一次解决。Linux下实现线程间消息队列的常见途径就两条，一条是内核提供的POSIX消息队列（mq_open/mq_send/mq_receive），另一条是自己用互斥锁+条件变量在用户态实现。本篇标题带着"(1)"，我计划从实用角度出发，先用mutex + condition_variable手写一个够用、可扩展的线程消息队列，把核心机制吃透，后面再展开批量消息、优先级、超时、无锁等进阶话题。
这个系列定位是"实用功能代码集"，所以不会去贴几千行的大框架，而是给可以直接搬进项目的核心代码和相关注意事项。适合谁看？如果你正在写多线程采集、日志异步落盘、任务分发这类程序，或者面试前想踏踏实实搞懂条件变量和生产者消费者模型，这篇值得花二十分钟读完，代码可以直接抄进你的工程改造。
2. 为什么不自造轮子也要先搞懂这套组合拳
2.1 说在前面：为什么不直接用std::queue配锁？
一段烂大街的代码如下面的写法，不少第一次写多线程的人第一反应都是这样：
std::queue<int> q; std::mutex mtx; void producer() { std::lock_guard<std::mutex> lock(mtx); q.push(1); } void consumer() { std::lock_guard<std::mutex> lock(mtx); if (!q.empty()) { int v = q.front(); q.pop(); // 处理v } }问题马上来了。消费者这端，当队列为空时它什么都做不了，只能反复lock、检查、unlock，这就是空转。更隐蔽的问题是：代码根本没处理"生产者唤醒消费者"这件事。如果消费者在q.empty()的瞬间被切走，生产者又push了新数据，消费者醒来后可能一直等不到下一次调度——虽然现实中因为锁的存在，这种情况的概率不算高，但谁都不想靠运气写并发代码。
条件变量（condition_variable）就是用来解决"等待+通知"这两个核心问题的。生产者入队后notify，消费者在条件不满足时把自己挂起，等通知来了再醒来，既避免了空转，又避免了唤醒丢失——当然，前提是使用姿势要对。
2.2 一个容易栽的细节：unique_lock和lock_guard的取舍
条件变量的等待操作必须搭配unique_lock，这是C++标准写死的。原因很朴素：wait()内部需要原子的完成"把线程阻塞"和"释放互斥锁"两个动作，等线程被唤醒后再自动重新持有锁。lock_guard的锁粒度是构造析构固定的，没法在中间释放再获取，所以只能由unique_lock来承担。这个点属于解释过一万遍但还是有人踩的基础知识，如果你之前只写过lock_guard，需要先转换一下思维。
3. 手写一个线程安全消息队列：核心实现与逐步拆解
3.1 接口设计的几个关键取舍
我实现的这个队列，基础版本只提供入队、出队、停止三个语义，没做批量接口，刻意保持小。下面的结构体是整个队列的核心：
template <typename T> class ThreadSafeQueue { public: explicit ThreadSafeQueue(size_t capacity) : capacity_(capacity), stopped_(false) {} bool push(T&& value); bool pop(T& value); void stop(); size_t size() const; private: mutable std::mutex mtx_; std::condition_variable not_empty_cv_; std::condition_variable not_full_cv_; std::queue<T> queue_; size_t capacity_; bool stopped_; };几个接口为什么要这样设计：
入队用右值引用（T&&），是为了减少复制。如果队列里装的是大对象（比如带缓冲区的数据包），每次push都深拷贝一次，开销非常明显，后面在优化部分我会专门测试带move和不带move的差别。
pop的语义是"阻塞直到拿到一个元素，或者队列被停止"。返回值用bool，false代表队列已经停止且没有残留数据，调用方可以借此退出循环。停止机制是这个队列和教科书版本最大的差异点，很多线上事故都源于消费者线程无法优雅退出，我单独用一节讲。
3.2 入队实现：notify放在锁内还是锁外？
bool push(T&& value) { std::unique_lock<std::mutex> lock(mtx_); not_full_cv_.wait(lock, [this]() { return stopped_ || queue_.size() < capacity_; }); if (stopped_) { return false; } queue_.emplace(std::forward<T>(value)); not_empty_cv_.notify_one(); return true; }wait(lock, predicate)是条件变量的标准写法，它等价于：
while (!predicate()) { cv.wait(lock); }重点是"自我唤醒后要重新检查条件"，也就是所谓的伪唤醒（spurious wakeup）防护。用if判断条件的写法在极端情况下会出问题，虽然概率小，但多线程程序出事就是事故，所以一律用带谓词的wait重载。
关于notify_one是放在锁内还是解锁后再调用，C++标准没有强制要求，但实测在锁内notify有个微小的问题：被唤醒的线程要等当前线程释放锁才能继续，多了一层无谓的调度延迟。如果是高吞吐场景，建议这样写：
bool push(T&& value) { { std::unique_lock<std::mutex> lock(mtx_); not_full_cv_.wait(lock, [this]() { return stopped_ || queue_.size() < capacity_; }); if (stopped_) { return false; } queue_.emplace(std::forward<T>(value)); } not_empty_cv_.notify_one(); return true; }这段是经典优化后的版本。先加锁、等条件、入队，然后立刻解锁，解锁后再通知消费者。这样消费者被唤醒时锁已经释放了，可以直接进入临界区，延迟更低。需要注意的坑是：必须保证notify时队列状态确实变化了，不能想着"我统一在函数末尾notify一次"，如果wait条件不满足早退，后续就会漏通知，造成生产者/消费者互相傻等。
3.3 弹出实现：pop的阻塞与超时语义
bool pop(T& value) { std::unique_lock<std::mutex> lock(mtx_); not_empty_cv_.wait(lock, [this]() { return stopped_ || !queue_.empty(); }); if (stopped_ && queue_.empty()) { return false; } value = std::move(queue_.front()); queue_.pop(); not_full_cv_.notify_one(); return true; }这里有两个细节很多人会忽视。一个是pop返回值的判定，必须同时判断stopped_和queue_.empty()。因为停止信号发出时，队列里可能还有残留数据，正确的语义是"先把存量数据消费完，再退出"，所以stop()之后生产者虽然被拦住了，但消费者的循环仍然能取完最后的元素，取完下次进入pop时queue_空了，才返回false。
另一个是pop里notify的时机。为什么出队后要notify生产者？因为队列容量是有限的，消费者拿走一个元素，生产者可能正卡在not_full等待上，必须通知它"有空位了"，否则队列满了以后生产者和消费者都会卡死。这个点反映的是生产者消费者模型里两类条件变量各自独立、各管各的。
如果想支持非阻塞式弹出，可以加一个tryPop：
bool tryPop(T& value) { std::lock_guard<std::mutex> lock(mtx_); if (queue_.empty()) { return false; } value = std::move(queue_.front()); queue_.pop(); not_full_cv_.notify_one(); return true; }使用场景是：消费者线程除了消费消息，还需要定期做其他事情（比如心跳），不能无限期阻塞在pop上，那就用tryPop轮询配合sleep，或者后面我们再扩展带超时的pop版本。
3.4 stop和析构：最容易被忽略的"停止机制"
void stop() { { std::lock_guard<std::mutex> lock(mtx_); stopped_ = true; } not_empty_cv_.notify_all(); not_full_cv_.notify_all(); }朴素版本常常漏了stop方面的工作。没有stop，线程退出就只能靠"发一个特殊消息"或者干脆detach让线程自生自灭，前者污染业务逻辑，后者在程序退出时容易崩溃。stop加了两道保险：把stopped_置为true，然后同时唤醒两个条件的等待者。注意这里是notify_all而不是notify_one，因为可能同时有多个生产者和多个消费者在等待，只唤醒一个会导致其他人无法退出。
析构函数里也需要调用stop()，防止派生类先析构了条件变量，消费者还挂在上面：
~ThreadSafeQueue() { stop(); }这个类析构时thread可能还在pop里wait，如果不通知，程序会直接卡在析构处——实际调试时的表现就是主线程退出时hang住，用gdb看到一堆线程在__pthread_cond_wait里。
4. 生产消费者场景的完整构建与实测验证
4.1 一个能跑的完整示例
代码只看不用是学不会的，下面给出一个可以直接编译运行的示例，包含两个生产者线程、两个消费者线程，以及主线程在3秒后触发停止：
#include <condition_variable> #include <cstdio> #include <mutex> #include <queue> #include <thread> #include <chrono> #include <vector> template <typename T> class ThreadSafeQueue { public: explicit ThreadSafeQueue(size_t capacity) : capacity_(capacity), stopped_(false) {} bool push(T&& value) { { std::unique_lock<std::mutex> lock(mtx_); not_full_cv_.wait(lock, [this]() { return stopped_ || queue_.size() < capacity_; }); if (stopped_) { return false; } queue_.emplace(std::forward<T>(value)); } not_empty_cv_.notify_one(); return true; } bool pop(T& value) { std::unique_lock<std::mutex> lock(mtx_); not_empty_cv_.wait(lock, [this]() { return stopped_ || !queue_.empty(); }); if (stopped_ && queue_.empty()) { return false; } value = std::move(queue_.front()); queue_.pop(); not_full_cv_.notify_one(); return true; } void stop() { { std::lock_guard<std::mutex> lock(mtx_); stopped_ = true; } not_empty_cv_.notify_all(); not_full_cv_.notify_all(); } size_t size() const { std::lock_guard<std::mutex> lock(mtx_); return queue_.size(); } private: mutable std::mutex mtx_; std::condition_variable not_empty_cv_; std::condition_variable not_full_cv_; std::queue<T> queue_; size_t capacity_; bool stopped_; }; void producer(ThreadSafeQueue<int>& q, int id) { for (int i = 0; i < 10; ++i) { bool ok = q.push(i + id * 100); if (ok) { printf("producer[%d] push %d, queue size=%zu\n", id, i + id * 100, q.size()); } else { printf("producer[%d] stopped, give up\n", id); break; } std::this_thread::sleep_for(std::chrono::milliseconds(30)); } } void consumer(ThreadSafeQueue<int>& q, int id) { while (true) { int value = 0; bool ok = q.pop(value); if (!ok) { printf("consumer[%d] queue stopped, exit\n", id); break; } printf("consumer[%d] pop %d, queue size=%zu\n", id, value, q.size()); std::this_thread::sleep_for(std::chrono::milliseconds(50)); } } int main() { ThreadSafeQueue<int> q(5); std::vector<std::thread> producers; std::vector<std::thread> consumers; for (int i = 0; i < 2; ++i) { producers.emplace_back(producer, std::ref(q), i + 1); consumers.emplace_back(consumer, std::ref(q), i + 1); } std::this_thread::sleep_for(std::chrono::seconds(3)); q.stop(); for (auto& t : producers) t.join(); for (auto& t : consumers) t.join(); printf("main exit, final queue size=%zu\n", q.size()); return 0; }这个示例的关键点在于验证两个行为：第一，消费者不会空转，测试时可以在top里看到消费线程处于睡眠状态，CPU占用率几乎为零；第二，主线程调用stop后，所有生产者消费者线程都能在短时间内退出，程序能顺利跑完，不会卡死。
提示：编译时务必加 -pthread，g++ 11 以上直接
g++ -std=c++11 -O2 -pthread test.cpp -o test即可。
4.2 实测中需要盯的三个观察点
跑这个程序，我建议不要把目光只放在"能跑"上，要盯三个细节：
第一个是队列长度变化。把printf打好，能看到q.size()在0到5之间波动，一旦到达5，生产者就会阻塞在not_full上，消费者的速度决定了能否把队列填满，这直观体现了流量控制的效果。如果去掉capacity限制，生产速度远大于消费时内存会无上限增长，这是分布式采集程序最常见的隐患之一。
第二个是消费顺序。生产者的i取值是0-9，消费者的打印顺序并不严格按照生产者入队顺序排列，因为两个生产者是并发的，谁先抢到锁不一定。如果你的业务要求严格的全局顺序，这个队列就不适用了，需要用带序号的消息或者单一生产者模型，这是消息队列选型时最容易忽略的点。
第三个是stop后队列里的残留数据。我在测试时故意让生产者比消费者快得多，程序退出前队列里还剩几条数据。观察到的行为是：stop后生产者们立即退出，但消费者们仍会继续pop，直到队列被清空后才返回false退出。这正是我们期待的语义——先消费完存量，再干干净净地终止。
5. 容量、唤醒与线程安全的三个细节补课
5.1 为什么需要两个条件变量而不是一个？
很多初版实现只用一个条件变量，然后无论入队还是出队都notify这一个，能工作，但效率很差。考虑一种情况：队列满了，所有生产者都在not_full上等待，此时一个消费者拿走一个元素，它呼叫notify，本意是通知生产者"有空位了"，但用单一条件变量时消费者自己也在这个变量上等待，notify完全可能唤醒另一个消费者，而消费者发现队列是空的（或者数据被其他消费者抢走了），只能再次睡过去。
这说明单一条件变量会造成"错误的唤醒"和"惊群"问题。用两个独立条件变量，生产者等not_full，消费者等not_empty，各等各的，唤醒才能精确送达。这一设计是教科书级别的，在实际多线程项目里尤其重要，因为它直接决定了高并发下锁竞争的激烈程度。
5.2 队列容量和"真满/假满"的判断
capacity语义上代表着内存占用的上限。在push的wait里，条件是queue_.size() < capacity_，注意这里用的是小于，也就是说容量为5时，队列最多能放5个元素。想要允许队列装到6个，判断就得改成<=，这种边界条件一旦写错，配合生产者消费者速度差异，往往在压力测试时才暴露。
另外，queue_.size()返回的是size_t类型，是无符号数，和capacity_比较时类型不一致，但两者都是非负的，不存在负数比较的坑。不过如果后续扩展成支持"无限容量"，这里条件会变成永真式，就失去控流的意义了。生产环境里我一般建议初始容量保守一点，比如处理网络包的线程队列设成几千个，处理磁盘IO的设成几百个，再根据实际q.size()的长期均值调整。
5.3 多消费者场景下的"惊群"与负载均衡
上面代码用的 multiple consumers 模式下，每次pop只会唤醒一个消费者，天然避免了惊群。但要注意：notify_one选择唤醒哪个线程由操作系统调度决定，不能保证"唤醒的一定是等待最久的线程"。如果希望多个消费者更公平地分摊任务，需要用更复杂的调度策略，比如每个消费者维护一个专属子队列，生产者按某种策略（轮询或负载最小的方式）选择队列投递。这种设计在单队列模式下是做不到的。
另外，pop操作本身的移动赋值（value = std::move(queue_.front())）意味着队列里存储的对象必须是可移动构造/可移动赋值的。如果你的T类型是const成员或者禁用了move，这里直接编译不过。C++11之后大多数标准库类型（string、vector、shared_ptr）都是可移动的，所以实际影响不大，但如果存放自定义对象，需要确认一下。
6. 从实测到工程：锁粒度、move语义与更优的消息传递方案
6.1 push一个对象要复制多少次？聊聊move语义
很多初学者写push时会把参数写成const T&，然后调用q.push(item)，看起来没毛病，但里外里会多出好几次拷贝。以std::string为例：
std::string msg = "hello"; q.push(msg); // msg被拷贝进临时对象，再移动进队列 q.push(std::move(msg)); // 直接把msg的资源转移进队列，msg变成空串 q.push("hello" + std::to_string(id)); // 临时对象直接被移动，零拷贝我的push签名用了T&&，配合std::forward，保证了对象以"移动"的方式进入容器。实测对比：一个包含10KB缓冲区的自定义结构，普通拷贝版每次push要拷贝10KB内存；改用move后只需要拷贝几十字节的指针和长度，性能差距在几十倍以上。这对网络数据包、日志行这类大对象尤其重要。
有个使用上的细节：调用q.push(std::move(msg))之后，msg就处于有效但未指定的状态，不能再依赖它的内容。如果后续还需要使用msg，建议先复制一份再move，或者干脆传const引用交给队列内部拷贝。这是move语义的天生代价，谈不上坑，但新手容易踩。
6.2 批量接口与锁粒度优化：一个预告性思路
单条push/pop在高吞吐下最明显的瓶颈是锁竞争。每一笔消息都要经历"加锁->等待条件->入队/出队->解锁->通知"的完整流程，当生产者和消费者数量都很多时，锁的竞争会非常激烈。
一种优化思路是把锁粒度放大，用"批量"来摊薄锁开销：
template <typename InputIt> bool pushBulk(InputIt first, InputIt last);接口实现时一次性拿锁，把一批元素全部入队，然后只做一次notify。消费者端的对应优化是批量弹出，也就是一次取走队列中所有可用的元素，处理完这一批后再回到wait状态。这里有个收益点：批量操作让同一批消息的处理共享了一次锁的开销，同时队列的吞吐量会因为锁竞争减少而显著提升，这个方向我计划在"(2)"里专门展开。
6.3 自旋锁和条件变量怎么选？
不是所有场景都适合用条件变量。如果队列里消息到达的频率极高，每个元素之间的间隔只有几十纳秒，条件变量上下文切换的开销反而会成为瓶颈，因为在互斥量和条件变量上进行线程阻塞唤醒的代价大约在几微秒级别。这种情况下可以把"阻塞等待"换成"自旋等待"——忙等一个原子变量。
Linux下常用的自旋锁方案是用std::atomic_flag，或者干脆用while (flag.test_and_set(std::memory_order_acquire));。自旋锁适合临界区极短且锁持有时间远小于线程切换时间的场景，但如果等待时间长了，CPU空转损耗同样可怕。实际工程选择时，我个人的经验法则是：消息到达平均间隔少于10微秒的，值得考虑自旋锁；更多的情况用互斥锁+条件变量锁定胜局，因为它在等待时不消耗CPU。
7. 排查卡死问题：一个完整的实战排查链路
7.1 现象：程序卡住不动了，怎么定位
某天你的程序突然不干活了，top里进程还在，但业务停止推进，日志也不打印。结合消息队列的使用场景，我倾向于按下面的顺序排查：
先看CPU。如果进程占用率在多个线程间跳来跳去，说明线程在空转，很可能进入了忙等循环，检查有没有把阻塞写法误写成while(queue.empty());这种自旋。如果CPU很低，线程基本都睡着了，那就是真的阻塞了，大概率所有线程都挂在了某个wait上。
再用gdb挂上去：
gdb -p <pid> thread apply all bt如果多个线程的调用栈同时卡在__pthread_cond_wait或者std::condition_variable::wait，说明大家都在等通知，那问题就变成"谁该通知但没通知"。我见过三种常见原因：
- 生产者线程崩了或者提前退出了，没人再入队，消费者等空了。
- 某个生产者线程长期持有mutex不释放（比如在持锁期间调用了阻塞IO），其他线程全在mutex上排队。
- 条件变量的notify和wait之间出现了竞争（用法不对），例如wait前先检查了一次条件，但没在锁内检查。
其中第三种最隐蔽，很多人会在wait外面先写if (queue_.empty())再进入wait，这在多线程下是经典的竞态条件。正确做法就是上面代码那样，把条件判断完全交给wait的谓词处理。
7.2 队列泄漏：consumer少了怎么办
检查程序里new出来的消息是否只入队不出队，malloc的缓冲是否只分配不释放，这种"队列泄漏"和内存泄漏很像，只是泄漏的是逻辑消息而不是内存。最直观的验证方法：程序跑一段时间后看q.size()是否持续增长不回落。如果一直增长，要么生产者速度远超消费者（容量不够引起的内存增长），要么消费者逻辑里遇到某些消息不处理就丢弃了，导致队列里的对象永远得不到释放。
对于前者，可以把队列容量调小，让生产者在队列满时阻塞更久，观察消费速度能不能跟上。对于后者，需要给消息加一个累计计数器，消费端每处理一条就自增，和生产端入队数对比，差多少一目了然。
7.3 一个亲测有效的验证手法：打点记录唤醒次数
排查完之后，怎么确信问题真的修好了？除了跑通业务，我建议给队列加一对调试计数器，一个统计pop成功次数，一个统计push成功次数，并配合时间戳打点打印。写一个小脚本，每100毫秒cat一次/proc/ /status里的voluntary_ctxt_switches值，对比正常和异常场景下的线程切换频率。手动模拟停住场景，逐步回放排查链路，比盲目改代码高效得多。
8. 再说几个绕不开的使用注意点
用完这个队列，有一些边角问题我实际踩过，顺手汇总在这里，能帮你少走弯路。
第一，pop里边执行value = std::move(queue_.front())，对T类型是有要求的。如果T是std::unique_ptr这种只移动类型，它的默认拷贝构造是被删除的，直接写在代码里能编过，但万一某个分支触发了拷贝路径（比如value是const的），编译直接报错。我给生产环境写的模板参数一律要求T可移动，而且显式用static_assert(std::is_move_constructible<T>::value)卡住，把错误尽早在编译期暴露出来。
第二，用std::ref(q)把队列传给线程时，生命周期管理一定要小心。让线程持有队列的引用或裸指针，调用方必须保证队列存活时间超过所有线程。我的习惯是把队列和线程一起打包进一个管理类，管理类析构时先stop再join，顺序不能反：一旦先join消费者线程再stop，消费者会永远挂在pop上；先stop再join则丝滑退出。
第三，停止后不应该允许继续push。代码里stopped_一旦为true，push即使等到了not_full也会返回false，这是有意的设计。但调用方要记得响应这个false，不然可能产生"想退出退不掉"的循环重试。我通常在生产者的循环里加一个if (!q.push(...)) break;，双方配合才能干净结束。
第四，条件变量的谓词尽量不要做耗时操作。我看到有人的谓词里直接调了queue_.size() >= 1000000这种比较，本身没问题，如果有人把size()实现成O(n)的遍历，那每次wait都会反复计算，性能会比较难看。C++标准里queue的size()不同实现有差异，不过常见的std::queue基于deque，size()是O(1)。即便如此，也不建议在谓词里做业务级计算，比如统计所有消息的大小总和，这个动作放在消费端做更合理。
9. 进步路线：这个队列后面还能怎么扩展
本系列标题是"(1)"，说明后面还有计划。我梳理一下这个线程消息队列可扩展的几个方向，有些是我自己项目里打磨过的，有些是看到的业界方案，供你参考：
一是批量弹出接口。一次取走队列中全部或最多N个元素，把锁竞争的开销摊薄到每个元素上，这在高吞吐采集场景收益很大。实现上注意返回一个std::vector ，移动整个vector出去比逐个move再push高效得多，但要小心弹出后通知生产者的次数，建议只notify一次。
二是超时接口。pop支持wait_for或wait_until，这样消费者可以"最多等100毫秒，没数据就去干点别的"或"等到某个绝对时间点"再回来。实现逻辑是把wait(lock, pred)换成wait_for(lock, timeout, pred)，注意条件变量和超时版本之间是"谓词满足或超时任一即可返回"的关系，别把两个逻辑搞混。
三是多队列路由。消息队列从"一个队列一对消费者"变成"多个队列，按业务类型分发"。这背后就是Linux网络协议栈里per-CPU队列思想的线程版，只是把per-CPU换成了per-worker或per-priority。核心收益是减少了锁竞争，代价是分发逻辑复杂了。对于日志系统按级别分开处理、对于网络包按连接分流的场景，都很值得做。
四是无锁队列。boost::lockfree::queue和moodycamel::ConcurrentQueue是Linux下常见的两个无锁实现。无锁队列不是万能药，它牺牲了"有界等待"的确定性，换来极低的锁开销。我实测过moodycamel的ConcurrentQueue在单生产者单消费者场景下，吞吐比mutex+条件变量高出一个数量级，但它的内存模型和ABA问题的处理更复杂，编码稍有不慎就把自己坑了。
作为系列第一篇，主要把mutex+条件变量这套基础打牢。后续我会按计划更新批量接口、性能测试对比、以及更高阶的无锁方案。如果你对哪个方向更感兴趣，可以按照自己的需要先自行扩展，期待在评论区看到你的实现思路和踩坑记录。
最后分享一个我自己贴了三年多的调试习惯：给每个线程起好名字（用pthread_setname_np，或者在类里维护一个name_成员），打印日志时带上线程名和队列残量。线上查问题时，一眼扫日志就能看出是哪个线程卡住、队列是在堆积还是被清空，比重新拉gdb要快得多。习惯虽小，排查效率至少提升一个档次。
