【Linux】线程同步与互斥

🔥铅笔小新z：个人主页 🎬博客专栏：Linux学习 💫滴水不绝，可穿石；步履不休，能至渊。

本节重点

• 深刻理解线程互斥的原理和操作
• 深刻理解线程同步
• 掌握生产消费模型
• 设计日志和线程池
• 理解线程安全和可重入，掌握锁相关概念

1. 线程互斥

1-1 进程线程间的互斥相关背景概念

共享资源

多个线程都能访问到的资源（比如全局变量）就是共享资源。

临界资源

被保护起来的共享资源，一次只允许一个线程访问，就叫做临界资源。

临界区

每个线程内部，访问临界资源的那一段代码，就叫做临界区。

互斥

任何时刻，保证有且只有一个执行流（线程）进入临界区，访问临界资源，这就是互斥。互斥的本质就是对临界资源起保护作用。

原子性

不会被任何调度机制打断的操作，叫做原子操作。原子操作只有两种状态：要么操作完成了，要么操作没完成——不存在"做了一半"这种中间状态。

📌 知识点总结：什么是临界资源、临界区、互斥和原子性？

想象一个公共卫生间只有一个坑位（临界资源），每次只能进去一个人（互斥）。排队进去、锁门、使用、出来——这一整套动作就是临界区。如果有人开门到一半突然被叫走，这就不是原子性的（被打断了）。真正的原子操作就像"锁门"这个动作，要么锁上了，要么没锁，没有中间状态。在多线程编程中，我们用互斥来保护临界资源，用原子操作来保证"检查-修改"这种组合操作不被中断。

1-2 互斥量 mutex

大部分情况下，线程使用的局部变量存储在线程栈中，这些变量归单个线程所有，其他线程无法访问。但有时候，我们需要多个线程共享一些变量（比如全局变量），这就是共享变量。

多个线程并发操作共享变量会带来问题，先看一个经典的售票系统反例：

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <pthread.h>

int ticket = 100; // 全局共享变量：总共有100张票

// 线程执行函数：每个线程都在这里抢票
void *route(void *arg)
{
char *id = (char*)arg;
while (1)
{
if (ticket > 0)
{ // 判断是否还有余票
usleep(1000); // 模拟卖票过程中的耗时操作
printf("%s sells ticket:%d\\n", id, ticket);
ticket—; // 卖出一张票，总数减1
}
else
{
break; // 没票了就退出
}
}
}

int main()
{
pthread_t t1, t2, t3, t4;
// 创建4个线程同时抢票
pthread_create(&t1, NULL, route, (void*)"thread 1");
pthread_create(&t2, NULL, route, (void*)"thread 2");
pthread_create(&t3, NULL, route, (void*)"thread 3");
pthread_create(&t4, NULL, route, (void*)"thread 4");
// 等待所有线程结束
pthread_join(t1, NULL);
pthread_join(t2, NULL);
pthread_join(t3, NULL);
pthread_join(t4, NULL);
}

一次执行结果：

thread 4 sells ticket:100
…
thread 4 sells ticket:1
thread 2 sells ticket:0
thread 1 sells ticket:-1
thread 3 sells ticket:-2

票数竟然变成了负数！这在实际生活中是不可能发生的。

为什么会出现问题？

来看 ticket– 对应的汇编代码：

mov 0x2004e3(%rip),%eax # 第1步：将ticket从内存加载到寄存器（load）
sub $0x1,%eax # 第2步：寄存器中的值减1（update）
mov %eax,0x2004da(%rip) # 第3步：将新值写回内存（store）

ticket– 对应三条汇编指令：

• load：将共享变量从内存加载到寄存器
• update：更新寄存器中的值（减1）
• store：将新值从寄存器写回内存

如果线程A执行完 load 还没执行 store 时被切走，线程B读到的还是旧值，就会出问题。

解决方案需要做到三点

要做到这三点，本质上就是需要一把锁。Linux 上提供的这把锁叫 互斥量（mutex）。

📌 知识点总结：为什么多线程操作共享变量会出问题？怎么解决？

多线程操作共享变量出问题的根因是：看似一步的操作（比如 ticket–）在CPU层面其实是多条指令（读→改→写），线程可能在任意两条指令之间被切换。这就导致数据不一致——两个线程同时读到票数是1，都觉得自己能卖，结果卖出了-1张票。解决方案就是加互斥锁（mutex），让"判断票数→卖票→减一"这一整个流程变成一个不可分割的原子操作，一个线程在做的时候，其他线程必须在门外等着。

1-3 互斥量的接口

初始化互斥量

有两种方式：

方法一：静态分配

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

方法二：动态分配

int pthread_mutex_init(pthread_mutex_t *restrict mutex, const pthread_mutexattr_t *restrict attr);
// 参数：
// mutex：要初始化的互斥量
// attr：属性，传NULL表示使用默认属性

销毁互斥量

int pthread_mutex_destroy(pthread_mutex_t *mutex);

注意：

• 使用 PTHREAD_MUTEX_INITIALIZER 初始化的互斥量不需要销毁
• 不要销毁一个已经加锁的互斥量
• 已经销毁的互斥量，要确保后面不会有线程再尝试加锁

互斥量加锁和解锁

int pthread_mutex_lock(pthread_mutex_t *mutex); // 加锁，成功返回0
int pthread_mutex_unlock(pthread_mutex_t *mutex); // 解锁，成功返回0

调用 pthread_mutex_lock 时可能遇到的情况：

• 互斥量处于未锁状态 → 锁定互斥量，返回成功
• 其他线程已经锁定互斥量 → 调用线程陷入阻塞（执行流被挂起），等待互斥量解锁

改进后的售票系统

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <pthread.h>
#include <sched.h>

int ticket = 100; // 共享资源：总票数
pthread_mutex_t mutex; // 定义一把互斥锁

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—; // 票数减1
pthread_mutex_unlock(&mutex); // 卖完票后解锁
} else {
pthread_mutex_unlock(&mutex); // 没票了也要解锁
break;
}
}
return nullptr;
}

int main(void)
{
pthread_t t1, t2, t3, t4;
pthread_mutex_init(&mutex, NULL); // 初始化互斥锁
pthread_create(&t1, NULL, route, (void*)"thread 1");
pthread_create(&t2, NULL, route, (void*)"thread 2");
pthread_create(&t3, NULL, route, (void*)"thread 3");
pthread_create(&t4, NULL, route, (void*)"thread 4");
pthread_join(t1, NULL);
pthread_join(t2, NULL);
pthread_join(t3, NULL);
pthread_join(t4, NULL);
pthread_mutex_destroy(&mutex); // 销毁互斥锁
}

📌 知识点总结：互斥量（mutex）怎么用？

互斥量就是一把"线程锁"，用法很简单：在访问共享资源之前调用 pthread_mutex_lock 上锁，用完以后调用 pthread_mutex_unlock 解锁。如果一个线程已经锁了，另一个线程再来 lock 就会被阻塞住，直到锁被释放。这有点像公共厕所——进去的人把门锁上（lock），外面的人只能排队等着，里面的人出来后再开门（unlock），下一个才能进去。注意，锁用完之后要销毁（destroy），别忘了。

1-4 互斥量实现原理探究

i++ 或 ++i 都不是原子的，它们对应多条汇编指令（load → modify → store），线程切换可能导致数据不一致。

为了实现互斥锁操作，大多数 CPU 都提供了 swap 或 exchange 指令，它的作用是把寄存器和内存单元的数据交换。由于这是一条指令，保证了原子性——即使是多处理器平台，访问内存的总线周期也有先后顺序，一个处理器上的交换指令执行时，另一个处理器的交换指令只能等待总线周期。

简单来说，加锁的底层原理是这样的：

📌 知识点总结：互斥锁的底层原理是怎么实现的？

互斥锁的底层依赖 CPU 提供的原子交换指令（swap/exchange）。这条指令可以一口气完成"把内存里的值和寄存器里的值互换"，中间不可被打断。加锁的过程本质上是线程之间"抢"一个标志位——谁通过原子交换把标志位从"空闲"改成"占用"，谁就抢到了锁。因为这个交换操作是一条 CPU 指令，所以是原子性的，不会出现"两个线程同时抢到锁"的情况。

1-5 互斥量的封装（RAII 风格）

直接用 pthread_mutex_lock/unlock 容易出问题——万一忘记解锁或者中间抛出异常，锁就永远解不开了。更优雅的方式是使用 RAII（资源获取即初始化） 风格，把锁的生命周期和作用域绑定。

#pragma once
#include <iostream>
#include <string>
#include <pthread.h>

namespace LockModule
{
// 对锁进行封装，可以独立使用
class Mutex
{
public:
// 删除拷贝构造函数和赋值运算符，防止锁被复制
Mutex(const Mutex &) = delete;
const Mutex &operator =(const Mutex &) = delete;

Mutex()
{
int n = pthread_mutex_init(&_mutex, nullptr);
(void)n; // 忽略返回值，生产代码应该检查
}

void Lock()
{
int n = pthread_mutex_lock(&_mutex);
(void)n;
}

void Unlock()
{
int n = pthread_mutex_unlock(&_mutex);
(void)n;
}

// 获取原始互斥量指针，供条件变量等接口使用
pthread_mutex_t *GetMutexOriginal()
{
return &_mutex;
}

~Mutex()
{
int n = pthread_mutex_destroy(&_mutex);
(void)n;
}

private:
pthread_mutex_t _mutex;
};

// 采用RAII风格进行锁管理：构造时加锁，析构时解锁
class LockGuard
{
public:
LockGuard(Mutex &mutex)
: _mutex(mutex)
{
_mutex.Lock(); // 构造时加锁
}

~LockGuard()
{
_mutex.Unlock(); // 析构时解锁
}

private:
Mutex &_mutex; // 持有锁的引用
};
}

使用 RAII 锁改造抢票代码

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <pthread.h>
#include "Lock.hpp"

using namespace LockModule;

int ticket = 1000; // 全局共享变量：总票数
Mutex mutex; // 全局互斥锁

void *route(void *arg)
{
char *id = (char *)arg;
while (1)
{
LockGuard lockguard(mutex); // 构造时加锁，出作用域自动解锁
if (ticket > 0)
{
usleep(1000);
printf("%s sells ticket:%d\\n", id, ticket);
ticket—;
}
else
{
break;
}
}
return nullptr;
}

int main(void)
{
pthread_t t1, t2, t3, t4;
pthread_create(&t1, NULL, route, (void*)"thread 1");
pthread_create(&t2, NULL, route, (void*)"thread 2");
pthread_create(&t3, NULL, route, (void*)"thread 3");
pthread_create(&t4, NULL, route, (void*)"thread 4");
pthread_join(t1, NULL);
pthread_join(t2, NULL);
pthread_join(t3, NULL);
pthread_join(t4, NULL);
}

📌 知识点总结：什么是 RAII 风格的锁？为什么用它？

RAII 锁就是"构造时自动加锁，析构时自动解锁"。你只需要在临界区开头创建一个 LockGuard 对象，剩下的就不用管了——变量出作用域时会自动调用析构函数释放锁。这样做最大的好处是永远不会忘记解锁，即使函数中间 return 或者抛出异常，C++ 的栈展开机制也会确保析构函数被调用。C++11 标准库也提供了类似的东西：std::lock_guard<std::mutex>。

2. 线程同步

2-1 条件变量

当一个线程互斥地访问某个变量时，它可能发现在其他线程改变状态之前，自己什么也做不了。

例如：一个线程访问队列时发现队列是空的，它只能等待，直到其他线程往队列中添加了元素。这种情况下就需要用到条件变量。

条件变量不是锁，它是一种等待通知机制：一个线程可以在某个条件不满足时主动休眠（等待），其他线程在条件满足时把它唤醒。

2-2 同步概念与竞态条件

同步

在保证数据安全的前提下，让线程能够按照某种特定的顺序访问临界资源，从而有效避免饥饿问题，叫做同步。

饥饿问题：一个线程一直在等资源，但总是轮不到它，就像排队买饭时总有人插队。

简单理解互斥和同步的区别：

• 互斥：保证同一时间只有一个人在厕所里（安全）
• 同步：保证大家上厕所的顺序是公平的，不会有人憋死（公平）

竞态条件

因为时序（执行顺序）问题而导致程序异常，叫做竞态条件。比如线程A先执行和线程B先执行，结果完全不同，这就是竞态条件。

📌 知识点总结：线程同步是什么？和互斥有什么区别？

互斥解决的是"能不能同时用"的问题——保证同一时刻只有一个线程访问共享资源。同步解决的是"按照什么顺序用"的问题——让线程按照某种约定的顺序执行，避免有的线程一直抢不到资源（饥饿）。打个比方：厕所只有一个坑位（互斥保证不会两个人同时用），但如果有人一直占着坑不出来，其他人就会饿死（这就是饥饿），同步机制就是确保大家轮着来，每个人都有机会。

2-3 条件变量函数

初始化

int pthread_cond_init(pthread_cond_t *restrict cond, const pthread_condattr_t *restrict attr);
// 参数：
// cond：要初始化的条件变量
// attr：属性，传NULL表示默认

也可以静态初始化：

pthread_cond_t cond = PTHREAD_COND_INITIALIZER;

销毁

int pthread_cond_destroy(pthread_cond_t *cond);

等待条件满足

int pthread_cond_wait(pthread_cond_t *restrict cond, pthread_mutex_t *restrict mutex);
// 参数：
// cond：在这个条件变量上等待
// mutex：互斥量（后面会详细解释为什么需要）

唤醒等待

int pthread_cond_broadcast(pthread_cond_t *cond); // 唤醒所有等待线程
int pthread_cond_signal(pthread_cond_t *cond); // 唤醒一个等待线程

简单示例

#include <iostream>
#include <string.h>
#include <unistd.h>
#include <pthread.h>

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);
}
}

int main(void)
{
pthread_t t1, t2;
pthread_create(&t1, NULL, active, (void*)"thread-1");
pthread_create(&t2, NULL, active, (void*)"thread-2");

sleep(3); // 确保两个线程已经在运行并等待

while (true) {
// pthread_cond_signal(&cond); // 每次唤醒一个线程
pthread_cond_broadcast(&cond); // 每次唤醒所有线程
sleep(1);
}

pthread_join(t1, NULL);
pthread_join(t2, NULL);
}

执行逻辑：

1. 初始化阶段

2. 线程抢锁与挂起阶段（在前 3 秒内）

在 main 线程睡觉的这 3 秒内，t1 和 t2 开始并发执行：

此时的状态： t1 和 t2 全都躺在等待队列里，而互斥锁 mutex 处于**空闲（无人占有）**状态。

3. 唤醒与执行阶段（3 秒之后）

main 线程醒来，进入它的 while(true) 死循环，根据你开启的代码不同，会有两种执行结果：

场景 A：如果执行的是 pthread_cond_broadcast(&cond);

• 最终现象： 屏幕上每隔 1 秒，会同时打印出 thread-1 活动… 和 thread-2 活动…（两者的先后顺序随机）。

场景 B：如果执行的是 pthread_cond_signal(&cond);

• 最终现象： 屏幕上每隔 1 秒，只会打印一行。可能是 thread-1 连续打印，也可能是两者交替，完全取决于操作系统的随机调度。

运行结果：

thread-1 活动…
thread-2 活动…
thread-1 活动…
thread-1 活动…
thread-2 活动…

📌 知识点总结：条件变量是什么？怎么用？

条件变量是一种等待-通知机制。当一个线程发现"条件不满足"（比如队列为空），它可以在条件变量上等待（休眠），而不是空转浪费 CPU。当其他线程改变了条件（比如往队列里放了数据），就通过 signal 或 broadcast 通知等待的线程醒来干活。关键点：pthread_cond_wait 会自动释放你传入的互斥锁，醒来时又会重新获取锁。这样就可以避免"休眠时还占着锁，别人进不去"的问题。

2-4 生产者消费者模型

321 原则（便于记忆）

3 种关系：

• 生产者 vs 生产者：互斥（不能同时往同一个位置放数据）
• 消费者 vs 消费者：互斥（不能同时从同一个位置取数据）
• 生产者 vs 消费者：互斥 + 同步（不能同时访问缓冲区，且缓冲区满时生产者等，空时消费者等）

2 种角色：生产者（生产数据）和 消费者（处理数据）

1 个交易场所：缓冲区（如阻塞队列、环形队列）

为什么使用生产者消费者模型

生产者和消费者模式通过一个容器（缓冲区）来解决生产者和消费者的强耦合问题：

• 生产者和消费者彼此之间不直接通讯
• 生产者把数据扔给阻塞队列即可，不用等消费者处理
• 消费者直接从阻塞队列取数据，不用找生产者要
• 阻塞队列就像一个缓冲区，平衡了生产者和消费者的处理能力

生产者消费者模型的优点

📌 知识点总结：生产者消费者模型的 321 原则是什么？有什么好处？

321原则是：3种关系（生产-生产互斥、消费-消费互斥、生产-消费互斥+同步）、2种角色（生产者和消费者）、1个缓冲区（交易场所）。好处有3点：一是解耦——生产者和消费者不需要直接打交道；二是支持并发——生产和消费可以同时进行；三是能应对忙闲不均——生产快了数据先存着，消费快了从缓冲区拿。就像一个快递中转站：快递员（生产者）把包裹放进驿站，客户（消费者）自己去取，不用双方约时间见面。

2-5 基于 BlockingQueue 的生产者消费者模型

什么是 BlockingQueue

阻塞队列是多线程编程中常用的数据结构，用于实现生产者消费者模型。与普通队列的区别：

• 队列为空时，尝试取元素的消费者线程会被阻塞，直到有数据放入
• 队列满时，尝试放元素的生产者线程会被阻塞，直到有数据被取出

BlockingQueue 封装代码

#ifndef __BLOCK_QUEUE_HPP__
#define __BLOCK_QUEUE_HPP__

#include <iostream>
#include <string>
#include <queue>
#include <pthread.h>

// 模板类，支持存放任意类型的数据
template <typename T>
class BlockQueue
{
private:
// 判断队列是否已满
bool IsFull()
{
return _block_queue.size() == _cap;
}

// 判断队列是否为空
bool IsEmpty()
{
return _block_queue.empty();
}

public:
// 构造函数，指定队列容量上限
BlockQueue(int cap) : _cap(cap)
{
_productor_wait_num = 0; // 等待的生产者数量
_consumer_wait_num = 0; // 等待的消费者数量
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()) { // 队列满了就等待
// 注意：pthread_cond_wait 会做三件事：
// a. 让调用线程在条件变量上等待
// b. 自动释放持有的 _mutex 锁
// c. 被唤醒后，重新竞争 _mutex 锁，竞争成功才返回
_productor_wait_num++;
pthread_cond_wait(&_product_cond, &_mutex); // 生产者等待
_productor_wait_num—;
}
// 进行生产
_block_queue.push(in);
// 通知消费者来消费
if (_consumer_wait_num > 0)
pthread_cond_signal(&_consum_cond);
pthread_mutex_unlock(&_mutex); // 解锁
}

// 消费者调用的接口：从队列中取出数据
void Pop(T *out)
{
pthread_mutex_lock(&_mutex); // 加锁
while (IsEmpty()) { // 队列空了就等待
_consumer_wait_num++;
pthread_cond_wait(&_consum_cond, &_mutex); // 消费者等待
_consumer_wait_num—;
}
// 进行消费
*out = _block_queue.front();
_block_queue.pop();
// 通知生产者来生产
if (_productor_wait_num > 0)
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:
std::queue<T> _block_queue; // 底层容器：标准库队列
int _cap; // 队列容量上限
pthread_mutex_t _mutex; // 保护队列的互斥锁
pthread_cond_t _product_cond; // 生产者条件变量
pthread_cond_t _consum_cond; // 消费者条件变量
int _productor_wait_num; // 等待的生产者数量
int _consumer_wait_num; // 等待的消费者数量
};

#endif

任务类型

#pragma once
#include <iostream>
#include <string>
#include <functional>

// 任务类型：使用函数对象，可以是任何可调用对象
using Task = std::function<void()>;

注意：这里使用模板，意味着队列中不仅可以存放 int 等内置类型，对象也可以作为任务参与生产消费过程。

📌 知识点总结：阻塞队列（BlockingQueue）在生产消费者模型中是怎么工作的？

阻塞队列就是一个带容量限制的线程安全队列。生产者放数据时，如果队列满了就阻塞等待（通过条件变量休眠），消费者取数据时，如果队列空了也阻塞等待。双方各有一个条件变量——生产者等 _product_cond（被通知"有空间了"），消费者等 _consum_cond（被通知"有数据了"）。这样就实现了"生产者-消费者"之间的同步：满时停生产，空时停消费。同时用互斥锁保护队列本身，保证任何时候只有一个线程在操作队列。

2-6 为什么 pthread_cond_wait 需要互斥量？

核心原因

条件等待是线程间同步的一种手段。如果只有一个线程，条件不满足就只能一直等——永远等不到满足的一天。所以必须要有另一个线程通过某些操作改变共享变量，使条件变得满足，并且通知等待的线程。

条件不会无缘无故突然变得满足，必然会牵扯到共享数据的变化，所以一定要用互斥锁来保护。没有互斥锁，就无法安全地获取和修改共享数据。

错误的设计

// 错误的设计：先解锁再等待
pthread_mutex_lock(&mutex);
while (condition_is_false) {
pthread_mutex_unlock(&mutex); // 解锁
// 在解锁之后、等待之前，条件可能已经满足，信号已经发出——但信号被错过了！
pthread_cond_wait(&cond, &mutex);
pthread_mutex_lock(&mutex); // 重新加锁
}
pthread_mutex_unlock(&mutex);

为什么这是错的？因为解锁和等待不是原子操作。解锁之后、pthread_cond_wait 之前，如果其他线程获取到互斥量，改变了条件并发送了信号，那么 pthread_cond_wait 将错过这个信号，导致线程永远阻塞。

正确的做法

pthread_cond_wait 的设计本身就是原子的：进入该函数后，会原子性地释放互斥量并开始等待。当被唤醒返回时，又会自动重新获取互斥量。这样，解锁和等待就是一个不可分割的操作，不会出现"信号被错过"的情况。

📌 知识点总结：为什么 pthread_cond_wait 需要传一个互斥量进去？

因为 pthread_cond_wait 要保证"释放锁 + 进入休眠"这两个操作是原子的。如果不传互斥量，你需要手动先解锁再等待——但解锁和等待之间有空隙，别的线程可能在这期间发送了信号，而你的线程还没开始等，信号就丢失了。pthread_cond_wait 内部会原子性地帮你完成：先释放锁让其他人能改条件，然后把自己挂起等待通知；被唤醒后又重新获取锁，保证你在读条件时是安全的。

2-7 条件变量使用规范

等待条件（消费者侧）

pthread_mutex_lock(&mutex); // 先加锁
while (条件为假) { // 用 while 而不是 if！
pthread_cond_wait(cond, mutex); // 等待条件满足
}
// 修改条件（如从队列取数据）
pthread_mutex_unlock(&mutex); // 解锁

为什么用 while 而不是 if？ 因为可能存在伪唤醒（spurious wakeup）——线程被唤醒了，但条件并不满足（比如多个消费者同时被唤醒，只有一个抢到了数据）。用 while 可以在唤醒后重新检查条件，不满足就继续等。

发送信号（生产者侧）

pthread_mutex_lock(&mutex); // 先加锁
设置条件为真 // 如往队列放数据
pthread_cond_signal(cond); // 通知等待的线程
pthread_mutex_unlock(&mutex); // 解锁

📌 知识点总结：使用条件变量的标准写法是什么？

标准写法分两边：等待方先加锁，然后用 while 循环判断条件（防止伪唤醒），条件不满足就 pthread_cond_wait（自动释放锁、休眠、醒来重新加锁），条件满足就执行操作，最后解锁。通知方先加锁，修改条件（比如往队列放数据），发送信号（signal或broadcast），最后解锁。用 while 而不是 if 是因为线程可能被"假唤醒"——明明条件没满足也被叫起来了，while 循环能再检查一次。

2-8 条件变量的封装

为了让条件变量更通用，封装时不要在 Cond 类内部引用对应的 Mutex，否则组合时难以初始化（因为 Mutex 和 Cond 基本是一起创建的）。

#pragma once
#include <iostream>
#include <string>
#include <pthread.h>
#include "Lock.hpp"

namespace CondModule
{
using namespace LockModule;

class Cond
{
public:
Cond()
{
int n = pthread_cond_init(&_cond, nullptr);
(void)n; // 生产代码应加错误处理
}

// 等待条件满足，需要传入互斥量
void Wait(Mutex &mutex)
{
int n = pthread_cond_wait(&_cond, mutex.GetMutexOriginal());
(void)n;
}

// 唤醒一个等待线程
void Notify()
{
int n = pthread_cond_signal(&_cond);
(void)n;
}

// 唤醒所有等待线程
void NotifyAll()
{
int n = pthread_cond_broadcast(&_cond);
(void)n;
}

~Cond()
{
int n = pthread_cond_destroy(&_cond);
(void)n;
}

private:
pthread_cond_t _cond;
};
}

📌 知识点总结：封装条件变量时需要注意什么？

封装条件变量时，不要把 Mutex 和 Cond 绑死在同一个类里，因为 Mutex 和 Cond 虽然经常一起用，但它们是不同的概念——一个负责互斥，一个负责同步。把 Cond 设计成独立类，使用时通过参数传入 Mutex，这样更灵活。这就像遥控器和电池——它们需要配合使用，但各自独立生产，用时再组装。

2-9 POSIX 信号量

POSIX 信号量和 SystemV 信号量作用相同，都是用于同步操作，达到无冲突访问共享资源的目的。POSIX 信号量可以用于线程间同步。

初始化

#include <semaphore.h>

int sem_init(sem_t *sem, int pshared, unsigned int value);
// 参数：
// sem：信号量对象
// pshared：0表示线程间共享，非0表示进程间共享
// value：信号量的初始值

销毁

int sem_destroy(sem_t *sem);

等待信号量（P操作）

int sem_wait(sem_t *sem); // 将信号量的值减1，如果值为0则阻塞

发布信号量（V操作）

int sem_post(sem_t *sem); // 将信号量的值加1，唤醒等待的线程

简单封装

#pragma once
#include <iostream>
#include <semaphore.h>

// 随手做一下封装
class Sem
{
public:
Sem(int n)
{
sem_init(&_sem, 0, n); // 初始值设为n，线程间共享
}

void P() // 等待资源（减1），相当于申请资源
{
sem_wait(&_sem);
}

void V() // 释放资源（加1），相当于归还资源
{
sem_post(&_sem);
}

~Sem()
{
sem_destroy(&_sem);
}

private:
sem_t _sem;
};

📌 知识点总结：什么是 POSIX 信号量？P/V 操作代表什么？

信号量本质上就是一个资源计数器。P 操作（sem_wait）申请一个资源——如果计数器>0就减1然后继续，如果计数器=0就阻塞等待。V 操作（sem_post）释放一个资源——计数器加1，如果有线程在等待就唤醒它。可以理解为停车场门口的显示屏：P 操作是"看看有没有空车位，有就进去（数字减1），没有就排队等"；V 操作是"车开走了，空出一个车位（数字加1），通知排队的人进来"。

2-10 基于环形队列的生产消费模型

基于固定大小的环形队列（用数组模拟），配合 POSIX 信号量实现生产消费模型。

设计思路

• 环形队列用数组模拟，通过取模运算实现环状特性
• 用两个信号量分别管理"剩余空间"和"已有数据"
• _room_sem：剩余空间数量（生产者关心）
• _data_sem：已有数据数量（消费者关心）

多生产多消费的 “321” 分析

• 3种关系：生产-生产（互斥）、消费-消费（互斥）、生产-消费（互斥+同步）
• 2种角色：生产者和消费者
• 1个交易场所：环形队列

解决方案：

• 需要 2 把锁：一把给生产者互斥，一把给消费者互斥
• 信号量本身是原子的，所以 P/V 操作不需要额外保护

完整实现

#pragma once
#include <iostream>
#include <string>
#include <vector>
#include <semaphore.h>
#include <pthread.h>

// 单生产单消费 / 多生产多消费 都适用
template<typename T>
class RingQueue
{
private:
// 加锁和解锁的辅助方法
void Lock(pthread_mutex_t &mutex)
{
pthread_mutex_lock(&mutex);
}

void Unlock(pthread_mutex_t &mutex)
{
pthread_mutex_unlock(&mutex);
}

public:
// 构造函数，指定队列容量
RingQueue(int cap)
: _ring_queue(cap),
_cap(cap),
_room_sem(cap), // 初始有cap个空位
_data_sem(0), // 初始有0个数据
_productor_step(0),
_consumer_step(0)
{
pthread_mutex_init(&_productor_mutex, nullptr); // 生产者互斥锁
pthread_mutex_init(&_consumer_mutex, nullptr); // 消费者互斥锁
}

// 生产者：向队列中放入数据
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(); // 数据数量加1，通知消费者
}

// 消费者：从队列中取出数据
void Pop(T *out)
{
// 消费行为
_data_sem.P(); // 申请一个数据（没有数据就等待）
Lock(_consumer_mutex); // 多个消费者互斥
*out = _ring_queue[_consumer_step++]; // 取出当前位置的数据
_consumer_step %= _cap; // 下标移动到下一个位置（环形）
Unlock(_consumer_mutex);
_room_sem.V(); // 空位数量加1，通知生产者
}

~RingQueue()
{
pthread_mutex_destroy(&_productor_mutex);
pthread_mutex_destroy(&_consumer_mutex);
}

private:
// 1. 环形队列的存储
std::vector<T> _ring_queue; // 底层数组
int _cap; // 容量上限

// 2. 生产和消费的下标
int _productor_step; // 生产者当前写入位置
int _consumer_step; // 消费者当前读取位置

// 3. 信号量
Sem _room_sem; // 剩余空间（生产者关心）
Sem _data_sem; // 已有数据（消费者关心）

// 4. 互斥锁（多生产多消费时需要）
pthread_mutex_t _productor_mutex; // 生产者之间的互斥锁
pthread_mutex_t _consumer_mutex; // 消费者之间的互斥锁
};

📌 知识点总结：基于环形队列的生产消费模型怎么设计？

环形队列 + 信号量是一个经典组合。用两个信号量：_room_sem 表示剩余空间（初始值=容量），_data_sem 表示已有数据（初始值=0）。生产者先 P(room) 申请空位，然后放数据，最后 V(data) 增加数据量；消费者先 P(data) 申请数据，然后取数据，最后 V(room) 释放空位。信号量本身的 P/V 是原子的，所以单生产单消费不需要加锁。多生产多消费场景下，需要额外加两把锁：一把锁住生产者之间的竞争，一把锁住消费者之间的竞争。因为生产者和消费者操作的是不同的下标，不需要同一把锁。

3. 线程池

3-1 日志与策略模式

什么是设计模式

设计模式是前辈们针对一些经典、常见的场景，总结出来的通用解决方案。它不是代码，而是一种编程思想。

日志认识

日志是记录系统和软件运行中发生事件的文件，主要作用是监控运行状态、记录异常信息，帮助快速定位问题。

日志格式中以下几个指标是必须有的：

• 时间戳
• 日志等级
• 日志内容

以下指标是可选的：

• 文件名和行号
• 进程ID、线程ID等

设计目标格式：

[2024-08-04 12:27:03] [DEBUG] [202938] [main.cc] [16] – hello world
[2024-08-04 12:27:03] [WARNING] [202938] [main.cc] [23] – hello world

策略模式日志实现

这里采用策略模式来设计日志系统，策略模式的核心思想是：定义一系列算法（策略），把它们封装起来，使它们可以互相替换。

#pragma once
#include <iostream>
#include <string>
#include <fstream>
#include <memory>
#include <ctime>
#include <sstream>
#include <filesystem> // C++17，需要高版本编译器 -std=c++17
#include <unistd.h>
#include "Lock.hpp"

namespace LogModule
{
using namespace LockModule;

// 默认路径和日志名称
const std::string defaultpath = "./log/";
const std::string defaultname = "log.txt";

// 日志等级枚举
enum class LogLevel
{
DEBUG,
INFO,
WARNING,
ERROR,
FATAL
};

// 将日志等级转换为字符串
std::string LogLevelToString(LogLevel level)
{
switch (level)
{
case LogLevel::DEBUG:
return "DEBUG";
case LogLevel::INFO:
return "INFO";
case LogLevel::WARNING:
return "WARNING";
case LogLevel::ERROR:
return "ERROR";
case LogLevel::FATAL:
return "FATAL";
default:
return "UNKNOWN";
}
}

// 获取可读性较强的时间字符串
std::string GetCurrTime()
{
time_t tm = time(nullptr);
struct tm curr;
localtime_r(&tm, &curr);
char timebuffer[64];
snprintf(timebuffer, sizeof(timebuffer), "%4d-%02d-%02d %02d:%02d:%02d",
curr.tm_year + 1900,
curr.tm_mon,
curr.tm_mday,
curr.tm_hour,
curr.tm_min,
curr.tm_sec);
return timebuffer;
}

// 策略模式：策略接口（抽象基类）
class LogStrategy
{
public:
virtual ~LogStrategy() = default;
virtual void SyncLog(const std::string &message) = 0; // 不同模式核心是刷新方式不同
};

// 具体策略1：控制台日志（打印到屏幕，方便调试）
class ConsoleLogStrategy : public LogStrategy
{
public:
void SyncLog(const std::string &message) override
{
LockGuard LockGuard(_mutex); // 显示器也是临界资源
std::cerr << message << std::endl;
}
private:
Mutex _mutex; // 保证输出线程安全
};

// 具体策略2：文件日志（写入到文件）
class FileLogStrategy : public LogStrategy
{
public:
FileLogStrategy(const std::string logpath = defaultpath,
std::string logfilename = defaultname)
: _logpath(logpath), _logfilename(logfilename)
{
LockGuard lockguard(_mutex);
if (std::filesystem::exists(_logpath)) // 如果目录已存在
return;
try
{
std::filesystem::create_directories(_logpath); // 创建目录
}
catch (const std::filesystem::filesystem_error &e)
{
std::cerr << e.what() << '\\n';
}
}

void SyncLog(const std::string &message) override
{
LockGuard lockguard(_mutex);
std::string log = _logpath + _logfilename;
std::ofstream out(log.c_str(), std::ios::app); // 追加方式写入
if (!out.is_open())
return;
out << message << "\\n";
out.close();
}

private:
std::string _logpath;
std::string _logfilename;
Mutex _mutex; // 保证文件写入线程安全
};

// 日志类本身
class Logger
{
public:
Logger()
{
UseConsoleStrategy(); // 默认使用控制台策略
}

~Logger() {}

// 切换策略：控制台输出
void UseConsoleStrategy()
{
_strategy = std::make_unique<ConsoleLogStrategy>();
}

// 切换策略：文件输出
void UseFileStrategy()
{
_strategy = std::make_unique<FileLogStrategy>();
}

// 内部类：表示一条完整的日志（RAII风格）
class LogMessage
{
public:
// 构造时格式化好日志头部信息
LogMessage(LogLevel type, std::string &filename, int line, Logger &logger)
: _type(type),
_curr_time(GetCurrTime()),
_pid(getpid()),
_filename(filename),
_line(line),
_logger(logger)
{
std::stringstream ssbuffer;
ssbuffer << "[" << _curr_time << "] "
<< "[" << LogLevelToString(type) << "] "
<< "[" << _pid << "] "
<< "[" << _filename << "] "
<< "[" << _line << "]"
<< " – ";
_loginfo = ssbuffer.str();
}

// 支持 C++ 风格的 << 连续输入
template <typename T>
LogMessage &operator<<(const T &info)
{
std::stringstream ssbuffer;
ssbuffer << info;
_loginfo += ssbuffer.str();
return *this; // 返回自身，支持链式调用
}

// 析构时（RAII）将完整的日志信息持久化
~LogMessage()
{
if (_logger._strategy)
{
_logger._strategy->SyncLog(_loginfo);
}
}

private:
LogLevel _type; // 日志等级
std::string _curr_time; // 日志时间
pid_t _pid; // 进程ID
std::string _filename; // 对应的文件名
int _line; // 对应的文件行号
Logger &_logger; // 引用外部Logger，方便使用策略刷新
std::string _loginfo; // 完整的日志信息
};

// 返回临时对象，形成 RAII 管理
LogMessage operator()(LogLevel type, std::string filename, int line)
{
return LogMessage(type, filename, line, *this);
}

private:
std::unique_ptr<LogStrategy> _strategy; // 日志写入策略
};

// 全局 Logger 对象
Logger logger;

// 宏：方便获取文件名和行号
#define LOG(type) logger(type, __FILE__, __LINE__)
// 宏：切换策略
#define ENABLE_CONSOLE_LOG_STRATEGY() logger.UseConsoleStrategy()
#define ENABLE_FILE_LOG_STRATEGY() logger.UseFileStrategy()
}

使用示例

#include <iostream>
#include "Log.hpp"

using namespace LogModule;

void fun()
{
int a = 10;
LOG(LogLevel::FATAL) << "hello world" << 1234 << ", 3.14" << 'c' << a;
}

int main()
{
LOG(LogLevel::DEBUG) << "hello world";
LOG(LogLevel::DEBUG) << "hello world";
LOG(LogLevel::WARNING) << "hello world";
fun();
return 0;
}

📌 知识点总结：策略模式在日志系统中是怎么体现的？

策略模式定义一系列算法，把它们封装成可互换的类。在日志系统中，我们定义了一个 LogStrategy 接口（纯虚基类），然后派生出两个具体策略：ConsoleLogStrategy（打印到控制台）和 FileLogStrategy（写入文件）。Logger 类持有策略对象的指针，运行时可以随时切换策略（比如调试时用控制台，部署后用文件）。这样，日志格式化和日志输出就解耦了——以后想加网络日志、数据库日志，只需再写一个策略子类，完全不用改现有代码。

3-2 线程池设计

什么是线程池

线程过多会带来调度开销，影响性能。线程池维护着多个线程，等待着监督管理者分配可并发执行的任务。这就避免了在处理短时间任务时，频繁创建和销毁线程的代价。

适用场景

线程池的种类

• 固定数量线程池：创建固定数量的线程，循环从任务队列取任务执行
• 浮动线程池：线程数量可以动态调整

此处我们选择固定线程个数的线程池。

线程池完整实现

#pragma once
#include <iostream>
#include <vector>
#include <queue>
#include <memory>
#include <pthread.h>
#include "Log.hpp" // 引入日志
#include "Thread.hpp" // 引入线程封装
#include "Lock.hpp" // 引入锁
#include "Cond.hpp" // 引入条件变量

using namespace ThreadModule;
using namespace CondModule;
using namespace LockModule;
using namespace LogModule;

const static int gdefaultthreadnum = 10; // 默认线程数

template <typename T>
class ThreadPool
{
private:
// 线程要执行的任务处理函数
void HandlerTask()
{
std::string name = GetThreadNameFromNptl();
LOG(LogLevel::INFO) << name << " is running…";
while (true)
{
// 1. 加锁保护任务队列
_mutex.Lock();
// 2. 队列为空且线程池在运行，则等待
while (_task_queue.empty() && _isrunning)
{
_waitnum++; // 等待的线程数加1
_cond.Wait(_mutex); // 等待条件变量（有任务来）
_waitnum—; // 等待的线程数减1
}
// 2.1 如果线程池已退出 && 任务队列为空
if (_task_queue.empty() && !_isrunning)
{
_mutex.Unlock();
break; // 退出循环，线程结束
}
// 2.2 线程池在运行 && 队列有任务：正常执行
// 2.3 线程池已退出 && 队列还有任务：处理完再退出
// 3. 一定有任务，取出任务
T t = _task_queue.front();
_task_queue.pop();
_mutex.Unlock(); // 取完任务立即解锁

LOG(LogLevel::DEBUG) << name << " get a task";
// 4. 处理任务（任务在线程中独占执行）
t();
}
}

public:
// 构造函数（私有，后面会改成单例）
ThreadPool(int threadnum = gdefaultthreadnum)
: _threadnum(threadnum), _waitnum(0), _isrunning(false)
{
LOG(LogLevel::INFO) << "ThreadPool Construct()";
}

// 初始化：创建所有线程对象（不启动）
void InitThreadPool()
{
for (int num = 0; num < _threadnum; num++)
{
_threads.emplace_back(
std::bind(&ThreadPool::HandlerTask, this)); // 绑定成员函数
LOG(LogLevel::INFO) << "init thread "
<< _threads.back().Name() << " done";
}
}

// 启动所有线程
void Start()
{
_isrunning = true;
for (auto &thread : _threads)
{
thread.Start();
LOG(LogLevel::INFO) << "start thread " << thread.Name() << "done";
}
}

// 停止线程池（发出退出信号）
void Stop()
{
_mutex.Lock();
_isrunning = false; // 标记线程池退出
_cond.NotifyAll(); // 唤醒所有等待的线程
_mutex.Unlock();
LOG(LogLevel::DEBUG) << "线程池退出中…";
}

// 等待所有线程退出
void Wait()
{
for (auto &thread : _threads)
{
thread.Join();
LOG(LogLevel::INFO) << thread.Name() << " 退出…";
}
}

// 向任务队列中添加任务
bool Enqueue(const T &t)
{
bool ret = false;
_mutex.Lock();
if (_isrunning)
{
_task_queue.push(t);
if (_waitnum > 0)
{
_cond.Notify(); // 有线程在等任务，唤醒一个
}
LOG(LogLevel::DEBUG) << "任务入队列成功";
ret = true;
}
_mutex.Unlock();
return ret;
}

~ThreadPool() {}

private:
int _threadnum; // 线程数量
std::vector<Thread> _threads; // 线程对象容器
std::queue<T> _task_queue; // 任务队列
Mutex _mutex; // 保护任务队列的锁
Cond _cond; // 条件变量（等待/通知任务）
int _waitnum; // 当前等待任务的线程数
bool _isrunning; // 线程池是否在运行
};

编译命令：

g++ main.cc -std=c++17 -lpthread

运行日志示例：

[2024-08-04 15:09:29] [INFO] [206342] [ThreadPool.hpp] [62] – ThreadPool Construct()
[2024-08-04 15:09:29] [INFO] [206342] [ThreadPool.hpp] [70] – init thread Thread-0 done
[2024-08-04 15:09:29] [INFO] [206342] [ThreadPool.hpp] [79] – start thread Thread-0done
[2024-08-04 15:09:29] [INFO] [206342] [ThreadPool.hpp] [28] – Thread-0 is running…
[2024-08-04 15:09:29] [DEBUG] [206342] [ThreadPool.hpp] [109] – 任务入队列成功
[2024-08-04 15:09:29] [DEBUG] [206342] [ThreadPool.hpp] [52] – Thread-0 get a task
this is a task
[2024-08-04 15:09:39] [DEBUG] [206342] [ThreadPool.hpp] [88] – 线程池退出中…
[2024-08-04 15:09:44] [INFO] [206342] [ThreadPool.hpp] [95] – Thread-0 退出…

📌 知识点总结：线程池是怎么设计的？为什么要用线程池？

线程池的核心思想是：预先创建一批线程，它们不干活时就阻塞在条件变量上等待任务；当有任务进来时，唤醒一个线程去执行，执行完再回来继续等。这样避免了频繁创建/销毁线程的开销。设计要点：用一个互斥锁保护任务队列，一个条件变量让线程在没任务时休眠，用 _isrunning 标志控制线程池的启停。退出时先把标志设为 false，然后广播唤醒所有线程，让它们自己检查到"要退出了"就主动结束。

3-3 线程安全的单例模式

什么是单例模式

某些类只应该有一个对象（实例），这就叫单例。比如一个男人只能有一个媳妇。在很多服务器开发场景中，经常需要让服务器加载大量数据到内存中，此时往往用一个单例类来管理这些数据。

饿汉方式和懒汉方式

用一个洗碗的例子来理解：

• 饿汉：吃完饭立刻洗碗。下一顿吃的时候可以立刻拿碗吃饭（提前准备好）
• 懒汉：吃完饭把碗放下，下一顿用到碗了再洗（用到时才创建）

懒汉方式最核心的思想是延时加载，可以优化服务器的启动速度。

饿汉方式实现

template <typename T>
class Singleton {
static T data; // 程序启动时就创建好
public:
static T* GetInstance() {
return &data;
}
};

优点：实现简单，线程安全（初始化发生在 main 之前） 缺点：不管用不用，都会创建对象，影响启动速度

懒汉方式实现（线程不安全版）

template <typename T>
class Singleton {
static T* inst; // 初始为nullptr
public:
static T* GetInstance() {
if (inst == NULL) {
inst = new T(); // 第一次调用时才创建
}
return inst;
}
};

问题：线程不安全。第一次调用时，如果两个线程同时进入，可能会创建两份对象。

懒汉方式实现（线程安全版）

// 懒汉模式，线程安全
template <typename T>
class Singleton {
volatile static T* inst; // volatile 防止编译器过度优化
static std::mutex lock;
public:
static T* GetInstance() {
if (inst == NULL) { // 第一层判断：避免不必要的锁竞争
lock.lock(); // 加锁保证线程安全
if (inst == NULL) { // 第二层判断：防止重复创建
inst = new T();
}
lock.unlock();
}
return inst;
}
};

关键点：

📌 知识点总结：懒汉单例和饿汉单例有什么区别？

饿汉是程序启动时就创建好对象，好处是实现简单、天然线程安全（初始化在 main 之前），坏处是拖慢启动速度，且如果一直不用就浪费了资源。懒汉是用到时才创建，好处是启动快、不浪费资源，坏处是多线程环境下要考虑线程安全。线程安全的懒汉单例使用"双重判定 + 加锁"的方案：外层 if 判断是否为 nullptr，避免每次调用都拿锁（性能优化）；内层 if 在锁保护下再判断一次，确保只创建一次。volatile 关键字防止编译器把变量优化到寄存器导致判断出错。

3-4 单例式线程池

将上面的线程池改造为单例模式，保证整个进程只有一个线程池实例。

#pragma once
#include <iostream>
#include <vector>
#include <queue>
#include <memory>
#include <pthread.h>
#include "Log.hpp"
#include "Thread.hpp"
#include "Lock.hpp"
#include "Cond.hpp"

using namespace ThreadModule;
using namespace CondModule;
using namespace LockModule;
using namespace LogModule;

const static int gdefaultthreadnum = 10;

template <typename T>
class ThreadPool
{
private:
// 构造函数私有化，外部不能直接创建
ThreadPool(int threadnum = gdefaultthreadnum)
: _threadnum(threadnum), _waitnum(0), _isrunning(false)
{
LOG(LogLevel::INFO) << "ThreadPool Construct()";
}

void InitThreadPool()
{
for (int num = 0; num < _threadnum; num++)
{
_threads.emplace_back(
std::bind(&ThreadPool::HandlerTask, this));
LOG(LogLevel::INFO) << "init thread "
<< _threads.back().Name() << " done";
}
}

void Start()
{
_isrunning = true;
for (auto &thread : _threads)
{
thread.Start();
LOG(LogLevel::INFO) << "start thread " << thread.Name() << "done";
}
}

void HandlerTask()
{
std::string name = GetThreadNameFromNptl();
LOG(LogLevel::INFO) << name << " is running…";
while (true)
{
_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();
LOG(LogLevel::DEBUG) << name << " get a task";
t();
}
}

// 禁止拷贝和赋值
ThreadPool<T> &operator=(const ThreadPool<T> &) = delete;
ThreadPool(const ThreadPool<T> &) = delete;

public:
// 获取单例对象的静态方法
static ThreadPool<T> *GetInstance()
{
// 双判断方式，有效减少加锁成本
if (nullptr == _instance) // 第一次之后，不需加锁直接返回
{
LockGuard lockguard(_lock);
if (nullptr == _instance) // 真正创建对象
{
_instance = new ThreadPool<T>();
_instance->InitThreadPool();
_instance->Start();
LOG(LogLevel::DEBUG) << "创建线程池单例";
return _instance;
}
}
LOG(LogLevel::DEBUG) << "获取线程池单例";
return _instance;
}

void Stop()
{
_mutex.Lock();
_isrunning = false;
_cond.NotifyAll();
_mutex.Unlock();
LOG(LogLevel::DEBUG) << "线程池退出中…";
}

void Wait()
{
for (auto &thread : _threads)
{
thread.Join();
LOG(LogLevel::INFO) << thread.Name() << " 退出…";
}
}

bool Enqueue(const T &t)
{
bool ret = false;
_mutex.Lock();
if (_isrunning)
{
_task_queue.push(t);
if (_waitnum > 0)
{
_cond.Notify();
}
LOG(LogLevel::DEBUG) << "任务入队列成功";
ret = true;
}
_mutex.Unlock();
return ret;
}

~ThreadPool() {}

private:
int _threadnum;
std::vector<Thread> _threads;
std::queue<T> _task_queue;
Mutex _mutex;
Cond _cond;
int _waitnum;
bool _isrunning;

// 单例相关静态成员
static ThreadPool<T> *_instance;
static Mutex _lock;
};

// 静态成员初始化
template <typename T>
ThreadPool<T> *ThreadPool<T>::_instance = nullptr;

template <typename T>
Mutex ThreadPool<T>::_lock;

测试用例

#include <iostream>
#include <functional>
#include <unistd.h>
#include "ThreadPool.hpp"

using task_t = std::function<void()>;

void DownLoad()
{
std::cout << "this is a task" << std::endl;
}

int main()
{
ENABLE_CONSOLE_LOG_STRATEGY();

int cnt = 10;
while (cnt)
{
// 通过单例获取线程池对象，添加任务
ThreadPool<task_t>::GetInstance()->Enqueue(DownLoad);
sleep(1);
cnt—;
}
ThreadPool<task_t>::GetInstance()->Stop();
sleep(5);
ThreadPool<task_t>::GetInstance()->Wait();
return 0;
}

📌 知识点总结：单例式线程池有什么好处？怎么保证线程安全？

单例式线程池保证整个程序只有一个线程池实例，避免重复创建线程池浪费资源。通过双重判定 + 加锁实现线程安全的懒汉单例：外层 if 判断是否已创建（避免每次请求都加锁），内层 if 在锁保护下真正创建对象（保证只创建一次）。构造函数私有化，外部只能通过 GetInstance() 静态方法获取线程池对象。在测试代码中，多次调用 GetInstance() 返回的都是同一个对象，添加任务时直接 Enqueue 即可。

4. 线程安全和重入问题

基本概念

线程安全

多个线程在访问共享资源时，能够正确地执行，不会相互干扰或破坏彼此的执行结果。多个线程并发执行同一段只有局部变量的代码时，不会出现不同的结果。但操作全局变量或静态变量且没有锁保护时，就容易出现线程安全问题。

重入

同一个函数被不同的执行流调用，当前一个流程还没有执行完，就有其他执行流再次进入，叫做重入。一个函数在重入的情况下，运行结果不会出现任何不同或问题，则该函数叫做可重入函数，否则叫做不可重入函数。

重入分为两种情况：

常见的线程不安全情况

• 不保护共享变量的函数
• 函数状态随着调用而发生变化的函数
• 返回指向静态变量指针的函数
• 调用了线程不安全函数的函数

常见的不可重入情况

• 调用了 malloc/free 函数（用全局链表管理堆）
• 调用了标准 I/O 库函数（很多实现以不可重入方式使用全局数据结构）
• 函数体内使用了静态的数据结构

常见的线程安全情况

• 每个线程对全局变量或静态变量只有读取权限，没有写入权限
• 类或接口对于线程来说都是原子操作
• 多个线程之间的切换不会导致执行结果有二义性

常见的可重入情况

• 不调用不可重入函数
• 不返回静态或全局数据，所有数据由调用者提供
• 使用本地数据，或通过制作全局数据的本地拷贝来保护全局数据

可重入与线程安全的联系

• 函数是可重入的，那就是线程安全的
• 函数是不可重入的，就不能由多个线程使用，可能引发线程安全问题
• 如果一个函数中有全局变量，那么这个函数既不是线程安全也不是可重入的

可重入与线程安全的区别

• 可重入函数是线程安全函数的一种
• 线程安全不一定是可重入的，而可重入函数一定是线程安全的
• 如果对临界资源的访问加了锁，这个函数是线程安全的，但如果重入时锁还未释放，就会产生死锁，因此是不可重入的
• 线程安全侧重描述线程访问公共资源的安全情况，表现的是并发线程的特点
• 可重入描述一个函数是否能被重复进入，表现的是函数本身的特点

📌 知识点总结：线程安全和可重入有什么区别和联系？

联系：可重入的函数一定是线程安全的（因为可重入要求更严格，它甚至不能依赖锁）。区别：线程安全侧重"多个线程同时用不会出问题"（可以通过加锁实现），可重入侧重"一个执行流还没执行完，另一个执行流再次进入也不会出问题"（不能依赖锁，因为锁可能已经被自己持有了）。比如一个加锁的保护共享变量的函数，它是线程安全的，但如果它在持有锁时被信号处理函数再次调用，就会死锁——所以它不是可重入的。

5. 常见锁概念

5-1 死锁

什么是死锁

申请一把锁是原子的，但是申请两把就不一定了。

造成的结果是：

死锁是指在一组进程中的各个进程均占有不会释放的资源，但因互相申请被其他进程占用的不会释放的资源，而处于的一种永久等待状态。

例如：线程 A 持有锁 1，线程 B 持有锁 2，A 想申请锁 2，B 想申请锁 1，双方都不释放自己手里的锁——于是就"死"在那了。

5-2 死锁的四个必要条件

5-3 避免死锁

破坏死锁的四个必要条件

核心思路：破坏循环等待条件。常见方法：

• 资源一次性分配：要么一次拿到所有需要的锁，要么一个都不拿
• 使用超时机制：申请锁时设置超时时间，超时了就放弃已持有的锁
• 加锁顺序一致：所有线程按照相同的顺序加锁（比如总是先锁 mtx1 再锁 mtx2）

示例：不加锁顺序一致会导致的问题

#include <iostream>
#include <mutex>
#include <thread>
#include <vector>
#include <unistd.h>

// 两个共享资源和两把锁
int shared_resource1 = 0;
int shared_resource2 = 0;
std::mutex mtx1, mtx2;

// 同时访问两个共享资源的函数
void access_shared_resources()
{
int cnt = 10000;
while (cnt)
{
++shared_resource1;
++shared_resource2;
cnt—;
}
}

// 模拟多线程并发访问
void simulate_concurrent_access()
{
std::vector<std::thread> threads;
for (int i = 0; i < 10; ++i)
{
threads.emplace_back(access_shared_resources);
}
for (auto &thread : threads)
{
thread.join();
}
std::cout << "Shared Resource 1: " << shared_resource1 << std::endl;
std::cout << "Shared Resource 2: " << shared_resource2 << std::endl;
}

int main()
{
simulate_concurrent_access();
return 0;
}

如果不按顺序加锁，结果可能不正确（资源1=94416，资源2=94536，不是期望的100000）。

使用 std::lock 一次性锁定多个互斥量，或者使用 std::unique_lock 配合 std::defer_lock 延迟加锁，可以避免死锁。

// 正确做法：同时锁定两个互斥量（原子操作）
std::unique_lock<std::mutex> lock1(mtx1, std::defer_lock);
std::unique_lock<std::mutex> lock2(mtx2, std::defer_lock);
std::lock(lock1, lock2); // 一次性锁定两个锁

5-4 避免死锁算法（了解）

• 死锁检测算法：允许死锁发生，但检测到时进行恢复
• 银行家算法：资源分配前判断是否会导致不安全状态

📌 知识点总结：什么是死锁？怎么避免？

死锁就是两个或多个线程互相等待对方释放资源，结果谁也没法继续执行。产生死锁需要同时满足四个条件：互斥、请求与保持、不剥夺、循环等待。只要破坏其中一个就能避免死锁。最实用的办法是让所有线程按相同的顺序加锁（比如都先锁 A 再锁 B），或者使用 std::lock 一次性申请所有锁。就像两个人要互相交换礼物——最安全的做法是双方同时松手再同时接住，而不是你先拿着自己的礼物等对方先给。

6. STL、智能指针和线程安全

6-1 STL 容器是否是线程安全的？

不是。

原因是 STL 的设计初衷是将性能挖掘到极致，而加锁保证线程安全会对性能造成巨大影响。不同容器的加锁方式也不同（比如哈希表有"锁表"和"锁桶"之分），所以 STL 默认不提供线程安全保证。在多线程环境下使用 STL 容器，需要调用者自行加锁保护。

6-2 智能指针是否是线程安全的？

• unique_ptr：只在当前代码块范围内生效，不存在线程安全问题（它是独占所有权的，不共享）
• shared_ptr：多个对象共用一个引用计数变量，存在线程安全问题。但标准库实现时已经考虑到了这一点，基于**原子操作（CAS）**的方式保证 shared_ptr 能够高效、原子地操作引用计数

📌 知识点总结：STL 容器和智能指针是线程安全的吗？

STL 容器不是线程安全的——这是设计上的取舍，为了性能不引入锁的开销。你自己用 STL 容器时需要手动加锁。unique_ptr 是独占的，不存在线程安全问题。shared_ptr 的引用计数的修改是原子的（通过 CAS 实现），所以引用计数本身是线程安全的，但 shared_ptr 指向的数据并不是自动保护的——多个线程通过 shared_ptr 读写指向的对象时仍然需要加锁。

7. 其他常见的锁

悲观锁

在每次取数据时，总是担心数据会被其他线程修改，所以会在取数据前先加锁（读锁、写锁、行锁等）。其他线程想要访问数据时会被阻塞挂起。互斥锁就是一种典型的悲观锁。

乐观锁

每次取数据时，总是乐观地认为数据不会被其他线程修改，因此不上锁。但在更新数据前，会判断数据在更新前有没有被其他线程修改过。主要采用两种方式：

CAS 操作

当需要更新数据时，判断当前内存值和之前取到的值是否相等：

• 如果相等，说明数据没有被修改过，用新值更新
• 如果不相等，说明数据已经被别人修改了，更新失败
• 失败后重试，一般是一个自旋的过程（不断重试）

CAS 通常对应 CPU 的一条原子指令（如 cmpxchg），所以它是原子的。

📌 知识点总结：悲观锁和乐观锁有什么区别？

悲观锁认为一定会有人抢，所以每次操作前先加锁——就像上车前先抢座，把座位占住，别人不能坐。乐观锁认为不太会有人抢，所以先不上锁，但在提交修改时检查一下——就像在座位上留个纸条写了自己的名字，更新时看看名字有没有被改过。如果改了说明有人坐了，那就重试（重新找座位）。CAS 是实现乐观锁的经典方式，它利用 CPU 提供的原子比较-交换指令来判断"值是否被改过"。悲观锁适合写多的场景，乐观锁适合读多的场景。

总结：线程同步与互斥全景图

本讲内容可以归纳为以下几个层次：