线程同步与互斥：抢票负数、锁的原理、生产消费模型与死锁
📝 前言
四个线程一起抢票，跑完一看：thread 1 sells ticket:-1。
票怎么会是负数？ticket 不是原子性的，判断和 ticket-- 之间随时可能被切走。
那加把锁就完事了？纯互斥下还有问题：1. 竞争锁是无序的；2. 得到锁的线程有可能反复申请锁，却无法做动作（其他线程的饥饿问题）。
引入新的技术必然导致新的问题，为了进一步的解决问题，必然有新技术。从一把锁一路走到死锁：
| 卡在哪 | 拿什么解 |
|---|---|
| 共享资源数据不一致、抢票出负数 | 互斥量 mutex |
| 竞争无序、别的线程一直饿着 | 条件变量 做线程同步 |
| 生产与消费强耦合、忙闲不均 | BlockingQueue |
| 资源按"块"分批使用 | POSIX 信号量 |
| 短任务反复创建销毁线程 | 线程池 + 单例 |
| 锁没释放又被同一个执行流重复进来 | 死锁 |
通过本文，你将掌握：
| 技能 | 应用场景 |
|---|---|
| 互斥量 | 保护临界资源，解决数据不一致 |
| 条件变量 | 让线程按照特定的顺序访问临界资源 |
| 生产消费模型 | 生产与消费过程解耦合、支持忙闲不均 |
| POSIX 信号量 | 基于环形队列的生产消费模型 |
| 线程池 / 单例 / 可重入 | 日志与线程安全的常见写法 |
🔍 问题来了： 一把锁就能保证安全，为什么还要同步、条件变量、信号量这一整套东西？
答案就是——互斥只管"同一时刻只能进一个"，不管"谁先谁后"！
文章目录
一、📂 先把几个词摆清楚
由于线程共享地址空间，所以线程会共享大部分资源，这会造成各种数据不一致问题。这就要靠 同步与互斥。
先把几个词记一下（后面一直要用）：
| 词 | 意思 |
|---|---|
| 共享资源 | 多个线程都能摸到的资源 |
| 临界资源 | 多线程执行流被保护的共享资源 |
| 临界区 | 每个线程内部，访问临界资源的那段代码 |
| 互斥 | 任何时刻，有且只有一个执行流进入临界区，通常对临界资源起保护作用 |
| 原子性 | 不会被任何调度机制打断的操作，只有两态，要么完成，要么未完成 |
大部分情况，线程用的数据都是局部变量，变量地址在线程栈空间内，归属单个线程，别的线程拿不到。但有时候很多变量必须线程间共享，靠共享完成交互。多个线程并发操作共享变量，就会带来问题。
要做到安全，本质上得满足三点：
① 代码进入临界区执行时，不允许其他线程进入；
② 临界区没人、多个线程同时要求进入时，只允许一个进；
③ 线程不在临界区里执行时，不能阻止其他线程进。
三点凑齐，本质上就是需要一把锁。
💡 局部变量不用保护，共享变量才要保护。先分清哪些是共享的，再谈加锁。
线程安全 说的是多个线程访问共享资源时能够正确执行，不会相互干扰或破坏彼此的执行结果。并发跑同一段只有局部变量的代码，一般不会跑出不同结果；翻车基本都在全局 / 静态变量没保护上。
程序员写的时候有可能出现这个问题：一个函数本来只用局部变量，跑得好好的；后来为了"记一下状态"改成全局 / 静态变量，多个线程一起进来，结果就不对了。变量一改成共享的，锁就得跟着补上。
二、🔍 抢票为什么抢出负数
多线程下抢票为什么会出现负数？ticket 不是原子性的，非一条汇编（一次动作）。
代码摆出来：
// ========== 无锁下：四个线程抢票，会抢出负数 ==========
int ticket = 1000;
void *route(void *arg)
{
char *id = (char *)arg;
while (1) {
if (ticket > 0) { // ① 判断：为真以后可以随时被切走
usleep(1000); // ② 模拟漫长业务过程，很多线程都能进来
printf("%s sells ticket:%d\n", id, ticket);
ticket--; // ③ 不是原子操作
} else {
break;
}
}
return nullptr;
}
拆成汇编看，ticket--; 是三条汇编组成的，是一种计算：
// ========== ticket-- 对应的三条汇编 ==========
mov 0x2004e3(%rip),%eax // ① load ：共享变量从内存加载到寄存器
sub $0x1,%eax // ② update ：寄存器里执行 -1
mov %eax,0x2004da(%rip) // ③ store ：新值从寄存器写回 ticket 的内存地址
① 从内存把 ticket 读到 CPU；
② 在 CPU 里算；
③ 写回内存。
可能在算完到写回之间被调度，之后再恢复就会出问题！
ticket > 0; 才是出现负数原因：判断为真以后未写入内存，同时其他线程判断，之后多个线程同时 ticket--，造成出现负数。
竞态条件 —— 因为时序问题，而导致程序异常。抢票出负数就是它。
💡 要想复现这种情况，制造更多并发、更多切换。切换通过：1. 时间片 2. 阻塞式 IO 3. sleep 等（陷入内核）；切回到线程：从内核态返回用户态，进行检查。
程序员写的时候有可能出现这个问题：不是每次都错，跑十次错一次，翻日志看到 -1 才反应过来。而且 usleep 模拟的那个"漫长业务过程"越长，能挤进来的线程越多。别指望"我这次跑对了"就算安全。
三、🛠️ 引入互斥量：定义、接口、加锁
解决这种情况需要锁。引入锁：pthread_mutex_t 互斥锁 / 互斥量，#include <pthread.h>。
① 全局锁（静态分配）：该锁使用后无需释放，程序结束后会自动释放；
② 局部定义（动态分配）：不同于全局锁，局部定义需要手动释放。
#include <pthread.h>
// ========== ① 全局锁：静态分配，不需要销毁 ==========
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
// ========== ② 局部定义：动态分配，需要手动销毁 ==========
int pthread_mutex_init(pthread_mutex_t *restrict mutex,
const pthread_mutexattr_t *restrict attr); // attr：锁属性，一般 NULL
int pthread_mutex_destroy(pthread_mutex_t *mutex);
⚠️ 销毁要注意三条：静态分配的互斥量不需要销毁；不要销毁一个已经加锁的互斥量；已经销毁的，要确保后面不会有线程再尝试加锁。
申请锁：
int pthread_mutex_lock(pthread_mutex_t *mutex); // 阻塞式申请
int pthread_mutex_trylock(pthread_mutex_t *mutex); // 非阻塞申请，申请失败会去执行其他
int pthread_mutex_unlock(pthread_mutex_t *mutex); // 成功返回 0，失败返回错误号
竞争申请锁，多线都必须先看到锁；锁本身就是临界资源，所以申请锁的过程，必须是原子的（硬件支持）。调用 lock 时两种情况：互斥量未锁，锁定并返回成功；别人已经锁了、或同时申请没竞争到，就陷入阻塞（执行流被挂起），等解锁。
成功：继续向后运行，访问临界区代码、访问临界资源；失败：阻塞挂起，申请执行流。
💡 加锁注意，尽量加锁的范围粒度要比较细，尽可能的不要包含太多非临界区代码。
加锁以后的抢票，其余不变，只在 while 里包一层：
// ========== 加锁后的抢票：临界区由并行变为串行 ==========
void *route(void *arg)
{
char *id = (char *)arg;
while (1) {
pthread_mutex_lock(&mutex);
if (ticket > 0) {
usleep(1000);
printf("%s sells ticket:%d\n", id, ticket);
ticket--;
pthread_mutex_unlock(&mutex);
} else {
pthread_mutex_unlock(&mutex); // break 前别忘了解锁
break;
}
}
return nullptr;
}
手写 unlock 容易漏。用 RAII 封一层，出作用域自动解锁：
// ========== RAII 风格的锁：构造加锁，析构解锁 ==========
class Mutex {
public:
Mutex() { pthread_mutex_init(&_mutex, nullptr); }
void Lock() { pthread_mutex_lock(&_mutex); }
void Unlock() { pthread_mutex_unlock(&_mutex); }
pthread_mutex_t *GetMutexOriginal() { return &_mutex; }
~Mutex() { pthread_mutex_destroy(&_mutex); }
private:
pthread_mutex_t _mutex;
};
class LockGuard {
public:
LockGuard(Mutex &mutex) : _mutex(mutex) { _mutex.Lock(); }
~LockGuard() { _mutex.Unlock(); }
private:
Mutex &_mutex;
};
// ========== 用起来：一行搞定，不会漏解锁 ==========
while (1) {
LockGuard lockguard(mutex);
if (ticket > 0) { usleep(1000); ticket--; }
else break;
}
C++11 里也有现成的 std::mutex + std::lock_guard，一样是 RAII；自己封一遍，是为了后面能和线程、条件变量一起成套用。
还有个顺序上的小细节：动态分配的锁，main 里先 pthread_mutex_init，再 create 线程，pthread_join 全收回来，最后才 pthread_mutex_destroy，别乱。
⚠️ 容易漏掉的三点：
①break/return/ 中间退出，忘了 unlock；
② 一把锁把半个函数都包进去，粒度太粗，并发直接变串行；
③ 锁还锁着就去 destroy，或者 init 完忘了 destroy。
线程如此，进程呢？进程间互斥如何处理？把锁放进共享内存 (pthread_mutex_t *)shm，用的还是线程锁。
四、⚙️ 理解锁：加锁凭什么就原子
加锁以后，临界区内部允许切换线程吗？切换了会怎么样？
🔍 允许切换，但是因为有锁存在，其它所有线程都要申请锁，而由于锁还未解锁，必然失败，被阻塞挂起或"离开"临界区域。所有线程必须等解锁！
对于竞争的进程（线程），只有等待和进入两种状态，这就是 原子性！
顺带一个效果：执行临界区代码由并行变为串行（执行期间不会被打扰，也是一种变相原子性的表现）。
锁的原理：1. 硬件级实现：关闭时钟中断，出问题会死机；2. 软件级实现。
前提：进程 / 线程切换，CPU 内寄存器硬件只有一套，但 CPU 寄存器内数据有多份，各自一份，是当前执行流的上下文。换句话说，是把一个变量内容，交换到 CPU 寄存器内部。本质是把该变量的内容，获取到当前执行流的硬件上下文中。当前 CPU 寄存器的硬件上下文（各个寄存器内容）是线程私有的！
我们用 swap、exchange 将内存中变量交换到 CPU 寄存器中；本质是进程 / 线程在获取锁，因为是交换，不是拷贝！所以，只有一份，所以，谁申请到，谁持有锁，相当于一把钥匙。
内存里的 mutex 加锁前值为 1；CPU 里先放一个 0（%al），一条指令把两者交换：换到 1 就是拿到锁，换到 0 就是锁在别人手里。xchgb 只有一条指令，所以原子性有保证（多处理器下总线周期也有先后，一个在换，另一个只能等）。
// ========== lock：把 %al 里的 0 和内存 mutex 交换 ==========
lock:
movb $0, %al
xchgb %al, mutex
if (al 寄存器的内容 > 0)
return 0; // 换到 1，申请成功
else
挂起等待; // 换到 0，锁被别人持有
goto lock;
// ========== unlock：把 mutex 还回 1 ==========
unlock:
movb $1, mutex
唤醒等待 mutex 的线程;
return 0;
💡
i++/++i同样不是原子的，有数据一致性问题；锁本身是靠一条交换指令把"多步"变成"一步"。
常见误解：加锁 = 不让线程切换。答：不是。切换照旧，只是别人申请锁必然失败、进不来。所以别在临界区里塞很长的 IO，切走是切走了，锁一直被你攥着，后面全在等。
五、🔄 纯互斥不够：线程同步与条件变量
引入新的技术必然导致新的问题，为了进一步的解决问题，必然有新技术。
纯互斥下有问题：
- 竞争锁是无序的
- 得到锁的线程有可能反复申请锁，却无法做动作（其他线程的饥饿问题）
处理：进行线程的使有序操作，也就是排队，这就是 线程同步。
同步：在保证数据安全情况下，让线程能够按照特定的顺序访问临界资源，从而有效避免饥饿。
典型场景：一个线程互斥地访问队列，发现队列是空的，在其它线程放数据进来之前，它什么也做不了，只能等。这种情况就需要条件变量。
条件变量 接口：
// ========== 全局 / 局部 ==========
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
int pthread_cond_init(pthread_cond_t *restrict cond,
const pthread_condattr_t *restrict attr);
int pthread_cond_destroy(pthread_cond_t *cond);
// ========== 队列等待（固定时间等待用 timedwait）==========
int pthread_cond_wait(pthread_cond_t *restrict cond,
pthread_mutex_t *restrict mutex);
int pthread_cond_timedwait(pthread_cond_t *restrict cond,
pthread_mutex_t *restrict mutex,
const struct timespec *restrict abstime);
// ========== 唤醒线程 ==========
int pthread_cond_broadcast(pthread_cond_t *cond); // 唤醒该条件变量下所有线程
int pthread_cond_signal(pthread_cond_t *cond); // 唤醒该条件变量下等待的一个线程
那什么时候等待呢？在资源条件不满足的时候，要对条件资源进行判定！由于判定本身就是在访问临界资源，所以判定结果也必然在临界资源内部。即，当条件不满足，线程立刻要休眠，这个过程也是在临界区内部（锁未解开的）。所以 wait 必须把锁打开，这就是 mutex 参数的原因。这也意味着，下次醒来要再次申请锁，在 wait 内自动执行，失败了就阻塞等待！
⚠️ 解锁和等待必须是一个原子操作。先解锁、再 wait，中间别的线程可能已经把条件改了、信号也发了，这个信号被错过，线程就永远阻塞。
为什么非得带把锁？条件不会无缘无故突然变满足，必然牵扯到共享数据的变化；没有互斥锁就没法安全地获取和修改共享数据。先上锁、发现不满足、解锁、再等，看着挺顺，其实是错的：
// ========== 错误设计：解锁和等待不是原子的 ==========
pthread_mutex_lock(&mutex);
while (条件为假) {
pthread_mutex_unlock(&mutex);
// 解锁之后、等待之前：别的线程可能已经改了条件、发了信号，信号被错过
pthread_cond_wait(&cond, &mutex);
pthread_mutex_lock(&mutex);
}
pthread_mutex_unlock(&mutex);
所以 wait 干的事是三件：a. 让调用线程等待；b. 自动释放曾经持有的锁；c. 条件满足被唤醒后，必须重新竞争锁，竞争成功才返回。之前安全，之后也安全。
条件变量使用规范：
// ========== 等待条件：判定必须用 while，不能用 if ==========
pthread_mutex_lock(&mutex);
while (条件为假) // 伪唤醒：醒来条件可能又不满足了
pthread_cond_wait(&cond, &mutex);
// 修改条件
pthread_mutex_unlock(&mutex);
// ========== 给条件发信号：改条件、发信号，都在锁内 ==========
pthread_mutex_lock(&mutex);
设置条件为真;
pthread_cond_signal(&cond);
pthread_mutex_unlock(&mutex);
条件变量，可以允许线程等待；可以允许一个线程唤醒在 cond 等待的其他线程，实现同步过程。
// ========== 条件变量简单案例：主线程按节奏唤醒 ==========
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
void *active(void *arg)
{
std::string name = static_cast<const char*>(arg);
while (true) {
pthread_mutex_lock(&mutex);
pthread_cond_wait(&cond, &mutex); // 等待 + 放锁，返回时已重新持有锁
std::cout << name << " 活动..." << std::endl;
pthread_mutex_unlock(&mutex);
}
}
// 主线程：两个线程起来之后，每秒唤醒一次
// pthread_cond_signal(&cond); // 唤醒一个线程
pthread_cond_broadcast(&cond); // 唤醒所有线程
把条件变量封一层，Wait / Notify / NotifyAll 各包一个接口就行：
// ========== 条件变量封装：锁从外面传，别在类内绑死 ==========
class Cond {
public:
Cond() { pthread_cond_init(&_cond, nullptr); }
void Wait(Mutex &mutex) { pthread_cond_wait(&_cond, mutex.GetMutexOriginal()); }
void Notify() { pthread_cond_signal(&_cond); }
void NotifyAll() { pthread_cond_broadcast(&_cond); }
~Cond() { pthread_cond_destroy(&_cond); }
private:
pthread_cond_t _cond;
};
⚠️ Cond 类里别直接引用封装好的 Mutex 对象，两个类一起创建时会互相耦合、初始化很难摆平；锁从外面传进来就好。还有，线程刚 create 完别急着 signal，等它真跑起来再发，不然第一次唤醒容易丢。
程序员写的时候有可能出现这个问题：程序偶尔卡死不动，怎么等都不醒；查到最后是 if 写成了 while（伪唤醒）、自己拆开了解锁和等待，或者信号发早了。
知道了"队列为空要等"，接着就是生产消费模型。
六、📦 生产和消费模型
1. 三种要素：生产者 s，消费者 s，一个交易场所，就是一种临界资源。
| 谁和谁 | 什么关系 |
|---|---|
| 生产者 ↔ 生产者 | 竞争关系，互斥关系 |
| 生产者 ↔ 消费者 | 互斥关系，和同步关系 |
| 消费者 ↔ 消费者 | 竞争关系，互斥关系 |
三种关系，两种角色（线程承担），1 种交易场所（以特定结构构成的一种"内存空间"）——“321” 原则。
2. 为什么要有生产消费模型：好处有三条：
① 生产与消费过程解耦合
② 支持忙闲不均
③ 提高效率，获取任务与处理任务是并发的。
生产者和消费者彼此不直接通讯，全靠队列传话：生产者生产完不用等消费者处理，直接扔给队列；消费者不找生产者要数据，直接从队列里取。队列就是个缓冲区，用来解耦，也平衡了两边处理能力。
基于 Blockingqueue 的生产消费模型：阻塞队列是一个容量具有上限的队列，不满足读写条件时，阻塞对应线程。和普通队列的区别就在这：为空时取元素的操作被阻塞，直到有人放进来；为满时放元素的操作被阻塞，直到有人取走。
Thread1 --put--> Blockingqueue --take--> Thread2；为空时 Thread2 被阻塞（读），满时 Thread1 被阻塞（写）。
空 / 满 就是上一节 wait 的那个"条件"，判定用 while，if 挡不住伪唤醒。
// ========== C++ queue 模拟阻塞队列 ==========
template <typename T>
class BlockQueue {
public:
BlockQueue(int cap) : _cap(cap) {
pthread_mutex_init(&_mutex, nullptr);
pthread_cond_init(&_product_cond, nullptr);
pthread_cond_init(&_consum_cond, nullptr);
}
void Enqueue(T &in) // 生产者用的接口
{
pthread_mutex_lock(&_mutex);
while (IsFull()) {
// 生产线程去等待，是在临界区中休眠的！你现在还持有锁呢
// wait 自动放锁，唤醒后必须重新竞争锁，竞争成功才返回
pthread_cond_wait(&_product_cond, &_mutex);
}
_block_queue.push(in); // 进行生产
pthread_cond_signal(&_consum_cond); // 通知消费者来消费
pthread_mutex_unlock(&_mutex);
}
void Pop(T *out) // 消费者用的接口
{
pthread_mutex_lock(&_mutex);
while (IsEmpty()) {
pthread_cond_wait(&_consum_cond, &_mutex); // 伪唤醒，用 while 兜住
}
*out = _block_queue.front(); // 进行消费
_block_queue.pop();
pthread_cond_signal(&_product_cond); // 通知生产者来生产
pthread_mutex_unlock(&_mutex);
}
~BlockQueue() {
pthread_mutex_destroy(&_mutex);
pthread_cond_destroy(&_product_cond);
pthread_cond_destroy(&_consum_cond);
}
private:
bool IsFull() { return _block_queue.size() == _cap; }
bool IsEmpty() { return _block_queue.empty(); }
private:
std::queue<T> _block_queue; // 阻塞队列，被整体使用
int _cap; // 总上限
pthread_mutex_t _mutex; // 保护队列的锁
pthread_cond_t _product_cond, _consum_cond; // 专门给生产者 / 消费者提供的条件变量
};
先写单生产、单消费，然后改成多生产、多消费，这里代码其实不变 —— 锁就一把，两个条件变量各管各的。
还有两个细节：
① 记一下有多少线程正在等（_productor_wait_num / _consumer_wait_num），有人在等才去 signal，省掉没必要的唤醒；
② 队列里放的不一定是 int，任务本身也可以是对象，比如 using Task = std::function<void()>;，取出来 t() 就跑。
💡 用模板就是为了这个：内置类型能放，对象也能作为任务参与生产消费的过程。
程序员写的时候有可能出现这个问题：队列空着还在 pop、信号发给了根本没人在等的条件变量、或者把任务只当成一个数字。空 / 满判定、signal 的对象、任务的类型，三样都要对上。
七、📊 POSIX 信号量
1. 回顾信号量：本质是一个计数器，是对特定资源的预订机制。多线程使用资源，有两种场景：
- ① 将目标资源整体使用
[mutex + 二元信号量]，也就是基于 blockingqueue 的生产消费者模型 - ② 将目标资源按照不同的"块"，分批使用
[信号量]，也就是基于环形队列的生产消费者模型
POSIX 信号量和 System V 信号量作用相同，都是用来同步、达到无冲突访问共享资源；但 POSIX 这套可以用在线程间同步。
上一节是 ①，这一节换 ②：环形队列用数组模拟，模运算模拟环状特性；起始状态和结束状态一样，本来不好判断空 / 满，但现在有信号量这个计数器，就很简单地进行多线程间的同步过程。
2. 模型
前置约定：
① 为空，生产者先运行；
② 为满，消费者先运行；
③ 生产者不能生产过多（套圈）；
④ 消费者不能超过生产者。
结论：
① 不为空 && 不为满时，可同时进行；
② 为空 | 为满时，生产消费应同步互斥。
用几个信号量？两个就够：生产者 关注空位置 sem_blank = N；消费者 关注数据 sem_data = 0。
环形队列还有个老问题：起始状态和结束状态长得一样，不好判空 / 满。以前得用三种办法：
① 加计数器
② 加标记位
③ 预留一个空位置当满
现在手里有信号量这个计数器，直接就判了。
需要几把锁？2 把。生产一把、消费一把，维护多生产多消费之间的互斥；同步交给两个信号量。还是先写单生产单消费，再改多生产多消费，代码不用动。
3. 接口：
#include <semaphore.h>
// ========== 初始化 / 销毁 ==========
int sem_init(sem_t *sem, int pshared, unsigned int value);
// pshared：0 表示线程间共享，非 0 表示进程间共享；value：信号量初始值
int sem_destroy(sem_t *sem);
// ========== 等待：将信号量值减 1，P 操作 ==========
int sem_wait(sem_t *sem);
// ========== 发布：表示资源使用完毕可以归还，值加 1，V 操作 ==========
int sem_post(sem_t *sem);
信号量的另一个本质，是把对临界资源是否存在、就就绪等的条件，以原子性的形式，呈现在访问临界资源之前就判断了。
// ========== 随手做一下封装 ==========
class Sem {
public:
Sem(int n) { sem_init(&_sem, 0, n); }
void P() { sem_wait(&_sem); }
void V() { sem_post(&_sem); }
~Sem() { sem_destroy(&_sem); }
private:
sem_t _sem;
};
// ========== 基于环形队列的生产消费 ==========
void Enqueue(const T &in)
{
_room_sem.P(); // 生产者关注空位置，一定有空间
Lock(_productor_mutex);
_ring_queue[_productor_step++] = in;
_productor_step %= _cap; // 模运算模拟环状
Unlock(_productor_mutex);
_data_sem.V(); // 数据多了一块
}
void Pop(T *out)
{
_data_sem.P(); // 消费者关注数据
Lock(_consumer_mutex);
*out = _ring_queue[_consumer_step++];
_consumer_step %= _cap;
Unlock(_consumer_mutex);
_room_sem.V(); // 空位置多了一块
}
生产 / 消费各一把锁，是维护多生产多消费之间的互斥关系；两个信号量负责同步。
| 对比项 | BlockingQueue | 环形队列 + 信号量 |
|---|---|---|
| 资源用法 | 整体使用，一把锁保护 | 按"块"分批使用 |
| 空 / 满判定 | 在锁内 while 判队列状态 | 信号量这个计数器提前判好 |
| 同步靠什么 | 条件变量 + 互斥量 | 两个信号量 |
| 互斥靠什么 | 同一把锁 | 生产者 / 消费者各一把锁 |
程序员写的时候有可能出现这个问题：P / V 顺序写反（先 V 再放数据，别人拿到的是旧值）、下标忘了 %= _cap、或者生产者跑太快套了圈。先 P 再写、写完再 V，这个顺序别翻。
八、🧵 线程池、日志、单例、可重入
线程池：一种线程使用模式。线程过多会带来调度开销，进而影响缓存局部性和整体性能；线程池维护着多个等着接任务的线程，省掉处理短任务时反复创建 / 销毁线程的代价，也防止过分调度。
场景：单个任务小、任务数量巨大的场合合适；长时间任务（比如一个 Telnet 连接请求）优点就不明显了。种类：a. 创建固定数量线程池，循环从任务队列取任务执行；b. 浮动线程池。这里选固定线程个数。
具体看这几种场合：
① 大量线程完成短任务，比如 WEB 服务器处理网页请求、热门网站的点击次数；
② 对响应要求苛刻的服务；
③ 突发大量请求，但不想因此把内存撑爆 —— 没线程池时短时间内创建一大堆线程，内存可能直接到极限。
反过来，长连接那种会话时间比创建线程的时间长得多，池子没啥优势。可用线程数看什么？并发处理器、核数、内存、网络 sockets 这些。
但要看得见线程在干嘛，得先有日志。
日志
-
概念：
① 日志等级：
Debug调试、Info常规、Warning告警、Error错误、Fatal致命② 日志功能：a. 形成完整日志 b. 刷新到目标文件（策略模式）
-
代码。格式大致是：
[时间] [日志等级] [pid] [文件名] [行号] - 消息内容。
一条日志必须有：时间戳、日志等级、日志内容；可选的：文件名行号、进程 / 线程 id。凑齐这些，出问题时才好定位。日志本来就是拿来监控运行状态、记录异常、快速定位问题的。
不同模式核心是刷新方式的不同；显示器也是临界资源，输出要加锁：
// ========== 策略模式：日志策略接口 ==========
class LogStrategy {
public:
virtual ~LogStrategy() = default;
virtual void SyncLog(const std::string &message) = 0;
};
// ========== 控制台日志策略：只向显示器打印，方便 debug ==========
class ConsoleLogStrategy : public LogStrategy {
public:
void SyncLog(const std::string &message) override {
LockGuard lockguard(_mutex);
std::cerr << message << std::endl;
}
private:
Mutex _mutex; // 显示器也是临界资源，保证输出线程安全
};
// ========== 文件日志策略：追加方式刷新到目标文件 ==========
class FileLogStrategy : public LogStrategy {
public:
void SyncLog(const std::string &message) override {
LockGuard lockguard(_mutex);
std::ofstream out((_logpath + _logfilename).c_str(), std::ios::app);
if (!out.is_open()) return;
out << message << "\n";
out.close();
}
private:
Mutex _mutex; // 文件也是共享的，同样要保护
std::string _logpath, _logfilename;
};
RAII 也能拿来写日志 —— 构造时把 [时间] [等级] [pid] [文件名] [行号] - 拼好，重载 << 收内容，临时对象析构时自动按策略刷新；LOG(type) 这个宏就是把文件名和行号塞进去。
程序员写的时候有可能出现这个问题：多线程日志输出交叉成一团、文件策略忘了加锁、或者一行日志里缺了时间 / 等级，事后完全没法查。
线程池设计：1. 代码 2. 单例模式。
创建、启动、停止三步；退出时要把队列里剩余任务处理完：
// ========== 固定数量线程池 ==========
void InitThreadPool()
{
for (int num = 0; num < _threadnum; num++) {
_threads.emplace_back(std::bind(&ThreadPool::HandlerTask, this));
}
}
void Start()
{
_isrunning = true;
for (auto &thread : _threads) thread.Start();
}
void Stop()
{
_mutex.Lock();
_isrunning = false;
_cond.NotifyAll(); // 叫醒所有等着取任务的线程
_mutex.Unlock();
}
线程取任务那一层，得把三种情况分清楚：
// ========== HandlerTask：取任务前先把三种情况判明白 ==========
_mutex.Lock();
while (_task_queue.empty() && _isrunning) { // 队列不一定有数据
_waitnum++;
_cond.Wait(_mutex);
_waitnum--;
}
// ① 已退出 && 队列空：解锁，退出线程
if (_task_queue.empty() && !_isrunning) { _mutex.Unlock(); break; }
// ② 没退出 && 队列非空：正常取任务
// ③ 已退出 && 队列非空：把剩下任务全处理完再退
T t = _task_queue.front();
_task_queue.pop();
_mutex.Unlock();
t(); // 任务是线程独占的
程序员写的时候有可能出现这个问题：Stop 一调就直接退出，队列里剩的任务全丢了；或者只 Notify 一个，还有一堆线程在睡。把队列里剩下的任务处理完再退，这一条最容易被漏。
单例模式：特点，一个类只有一个对象。如何实现？私有所有可创建对象的方法，保留一个静态成员，创建唯一对象。
吃完饭立刻洗碗，下顿直接拿来用，叫饿汉；吃完先放着，下顿要用才洗，叫懒汉。核心是"延时加载"，优化启动速度。
什么时候会想用单例？服务器经常要把上百 G 数据加载进内存，这种事交给一个类管一份就够。饿汉的实现里，只要都通过 Singleton 包一层去用 T，一个进程里就只有一个 T 实例。
两种方式：
- ① 饿汉模式：直接在类内
static修饰成员变量（全局 data 区，一旦加载，可以直接用）；成员太大，加载速度变慢 - ② 懒汉模式（常使用）：在类内
static修饰成员变量指针，再给一个静态方法来new空间；核心是"延时加载"，优化启动速度
全局 data 区 ↑ static 成员变量（相当于）：当类被加载到内存，会有一个对象被创建。
懒汉不加锁行不行？答：不行。第一次 GetInstance 两个线程同时进来，可能 new 出两份；后面再调用倒是没问题了。
// ========== 饿汉方式：static 成员变量 ==========
template <typename T>
class Singleton {
static T data;
public:
static T* GetInstance() { return &data; }
};
// ========== 懒汉方式（线程安全版本）：static 指针 + 双判断 ==========
template <typename T>
class Singleton {
volatile static T* inst;
static std::mutex lock;
public:
static T* GetInstance() {
if (inst == NULL) { // 双重判定空指针，降低锁冲突概率
lock.lock(); // 保证多线程情况下也只调用一次 new
if (inst == NULL) inst = new T();
lock.unlock();
}
return inst;
}
};
⚠️ 多线程同时访问线程池，要加锁；三个点别漏：加锁解锁的位置、双重 if 判定避免不必要的锁竞争、
volatile防止过度优化。
程序员写的时候有可能出现这个问题：单例在多线程里被创建了两次，还是偶发的，测试环境根本复现不出来。
关于线程安全与可重入函数：可重入函数一般包括两种情况：
① 多线程重入；
② 信号导致同一执行流重复进入函数。
线程安全侧重说明线程访问公共资源的安全情况，表现的是并发线程的特点（指线程执行方面）；可重入描述的是一个函数是否能被重复进入，表示的是函数的特点（指数据方面）。
可重入函数必然线程安全，而线程安全不一定可重入。
| 对比项 | 线程安全 | 可重入 |
|---|---|---|
| 说的是谁 | 并发线程的执行 | 函数本身 |
| 常见情况 | 每个线程对全局变量或静态变量只有读取、没有写入 | 不使用全局或静态数据结构，不调用不可重入函数 |
| 常见坑 | 多个线程同时写共享数据 | 调用了 malloc/free、标准 IO 等以不可重入方式使用全局结构的接口 |
| 关系 | 不一定可重入 | 一定是线程安全的 |
常见不安全的例子：
① 不保护共享变量的函数；
② 函数状态随被调用而变化的；
③ 返回指向静态变量指针的；
④ 调用了线程不安全函数的。
反过来，常见可重入的例子：
① 不碰全局 / 静态变量；
② 不用 malloc / new 开出的全局空间；
③ 不返回静态或全局数据，数据全由调用者提供；
④ 只用本地数据，或者把全局数据做一份本地拷贝；
⑤ 不调用不可重入函数。
函数里有个全局变量还去写它，那它既不是线程安全、也不是可重入。不考虑"信号让同一执行流重复进入"这种情况时，线程安全和可重入在安全角度其实不做区分。
加锁可以使不可重入函数在多线程执行时表现出可重入函数的特性，但无法保证信号重复执行时正常，这时就撞上 死锁 了。
程序员写的时候有可能出现这个问题：在信号处理函数里调了 malloc/free 或标准 IO，或者函数里塞了个 static 计数器 —— 单线程跑得好好的，一上并发 / 信号就出问题。
锁还没释放，又被同一个执行流重复进来，就卡住了。
九、⚠️ 死锁
死锁：指在一组执行流中均占有不会释放的资源，但因互相申请被其他执行流占用的资源而处于的一种永久等待状态。
申请一把锁是原子的，但是申请两把锁就不一定了（线程 A 拿着锁 1 等锁 2，线程 B 拿着锁 2 等锁 1，才能进行后续资源的访问）。
产生死锁的必要条件：
| 条件 | 意思 |
|---|---|
| ① 互斥条件 | 资源每次只能被一个执行流使用 |
| ② 请求与保持条件 | 一个执行流因请求资源而阻塞，对已获得的资源不放 |
| ③ 不剥夺条件 | 一个执行流已获得的资源，在未使用完之前不能强行剥夺 |
| ④ 循环等待条件 | 若干执行流之间形成头尾相接的循环等待资源关系 |
怎么办？四个条件破坏掉任意一个就行。常用的：资源一次性分配（破 ②）、加锁顺序一致（破 ④）、超时机制和使用 trylock（破 ③）。避免锁未释放的场景，也算一条。
想再往深走，还有两个算法（了解就行）：死锁检测算法、银行家算法。
⚠️ 四个条件是"必要"不是"充分"，凑齐才可能死锁；但真卡住的时候，四个一定都在。
程序员写的时候有可能出现这个问题：两处代码都要拿锁 1 和锁 2，一处先拿 1、另一处先拿 2，循环等待就成了。还有个直观的例子：两个共享计数各加 10000 次、跑 10 个线程，分开两次申请锁，最后常常对不上（94416 / 94536 这种）；一次性把两把锁都申请到再算，才是 100000 / 100000。
十、📌 接口速查：一张表收尾
| 要干的事 | 接口 |
|---|---|
| 定义 / 初始化锁 | PTHREAD_MUTEX_INITIALIZER / pthread_mutex_init |
| 销毁锁 | pthread_mutex_destroy |
| 加锁 / 非阻塞加锁 / 解锁 | pthread_mutex_lock / pthread_mutex_trylock / pthread_mutex_unlock |
| 定义 / 销毁条件变量 | PTHREAD_COND_INITIALIZER / pthread_cond_init / pthread_cond_destroy |
| 等待 / 定时等待 | pthread_cond_wait / pthread_cond_timedwait |
| 唤醒一个 / 唤醒全部 | pthread_cond_signal / pthread_cond_broadcast |
| 初始化 / 销毁信号量 | sem_init / sem_destroy |
| P / V | sem_wait / sem_post |
一条主线再顺一遍：共享资源出问题，抢票抢出负数，先拿 锁 来管；锁凭什么原子？swap / exchange 一条指令把变量换进寄存器。纯互斥又会饿着别的线程，接着是 条件变量 + 同步；落到队列上就是 BlockingQueue，改成按"块"用资源就是 环形队列 + 信号量；一批线程来跑任务，有了 线程池 + 单例；锁没解开又被重复进来，最后撞上 死锁。
表里的接口不用背，用的时候照着查；真正要记的是每一层在解决上一层的什么问题。
十一、🧰 几个常被问到的
写到这儿，还有几个平时一定会碰到的问题。
① STL 容器线程安全吗？
答：不是。设计初衷是把性能挖到极致，一加锁性能掉得厉害；而且不同容器加锁方式还不一样（hash 表就有锁表、锁桶的区别）。多线程里想用，调用者自己保证。
程序员写的时候有可能出现这个问题：多个线程一起往同一个容器 push，偶发崩、或者数据对不上，还查不出规律。
② 智能指针呢？
unique_ptr只在当前代码块范围内生效，不涉及线程安全问题shared_ptr多个对象要共用一个引用计数变量，本来会有线程安全问题；但标准库实现里已经用 CAS 原子操作 保证了，能高效、原子地改引用计数
③ 还有哪些锁？（了解）
| 名字 | 一句话 |
|---|---|
| 悲观锁 | 取数据前先加锁，总担心别人会改；别人要访问就阻塞挂起 |
| 乐观锁 | 取数据不上锁，更新前才判断期间有没有被改过；靠版本号机制 / CAS |
| CAS | 更新时看内存里的值还等不等于刚读到的，相等就写新值，不等就失败重试（一般是个自旋过程） |
| 自旋锁 | 拿不到锁就一直重试，不挂起 |
| 读写锁 / 行锁 | 按读 / 写分开加锁，粒度更细 |
CAS 和前面 xchgb 是同一路的想法 —— 把"判断 + 写入"压成一次原子动作；只不过 CAS 失败是重试，锁失败是挂起。
程序员写的时候有可能出现这个问题：以为 STL 容器自带线程安全、以为 shared_ptr 随便并发用没事、或者自旋锁等太久把 CPU 吃满。
本节完
📝 作者：say-fall | 编辑：say-fall | 🌟 原创不易，如果对你有帮助，记得 👍 点赞 + ⭐ 收藏 哦！
转载自 CSDN-专业IT技术社区
原文链接：https://blog.csdn.net/say_fall/article/details/170242362
