【Linux】线程全解：概念、操作、互斥与同步机制、线程池实现
🎬 个人主页：谁在夜里看海.
📖 个人专栏：《C++系列》《Linux系列》《算法系列》
⛰️ 道阻且长，行则将至
目录
📚一、线程概念
📖 回顾进程
📖 引入线程
📖 总结
📖 Linux下的线程
📚二、线程操作（phread）
📖1.线程创建
🔖语法
🔖示例
📖2.线程退出
🔖语法
🔖示例
📖3.线程取消
🔖语法
🔖示例
📖4.线程等待
🔖语法
🔖示例
📖5.分离线程
🔖语法
🔖示例
📚三、线程互斥
📖 引入
📖 互斥量的接口
🔖1.初始化
🔖2.销毁
🔖3.加锁
🔖4.解锁
🔖示例
📖 死锁
🔖四个必要条件
🔖示例
🔖解决办法
📚四、线程同步
📖 概念
📖1.条件变量
🔖基本概念
🔖语法
🔖示例：生产消费者模型
📖2.信号量
🔖基本概念
🔖语法
🔖示例：基于环形队列的PC-model
📚五、线程池
📖 概念
📖 实现
🔖1.初始化
🔖2.任务队列
🔖3.线程同步
🔖4.工作线程
🔖5.线程池退出
🔖总结
📚一、线程概念
线程与进程密不可分，要理解线程的概念，我们先来回顾一下进程：
📖 回顾进程
我们说，进程是操作系统资源分配的基本单位，是一个程序在计算机上的运行实例，对于这句话，我们分两点进行阐述：
① 一个进程被创建时，系统会它开辟一段虚拟地址空间，每个进程都有各自的虚拟内存空间，这保证了进程间的独立性。操作系统会为每个进程分配各种资源因此说，进程是资源分配的基本单位；
② 当程序被加载到内存并开始执行时，它就变成了一个进程。换句话说，程序是静态的文件，而进程是程序运行的动态实体。
📖 引入线程
程序被加载到内存中是要执行一些操作的， 但如果系统仅支持单一的进程模型，每个程序只能顺序执行（只有一个执行流），这样在面对一些复杂任务以及高并发情况时，并不能满足需求。
此时我们可以采用多进程模型执行，但是同样也会产生一些问题：由于每个进程都需要独立的虚拟内存空间，如果大量创建进程会导致巨大的内存开销；对于一些高并发的小任务，如大量 I/O 操作，并不需要让每个操作都独占一个进程，这样太浪费资源了，那有没有什么办法，可以让这些并发操作在一个进程下执行呢？
于是就引入了多线程的概念：线程是进程内的执行单元，多个线程可以在同一个进程中并发执行。由于线程共享进程的内存空间，因此相比多进程所需的内存和资源消耗更少，并且线程之间的上下文切换比进程之间的上下文切换也要更快，这使得多线程模型在I/O密集型、高并发等场景下具有非常大的优势。
📖 总结
如果说进程是操作系统资源分配的基本单元，那么线程就是资源调度的基本单元，也是程序执行的基本单元。我们可以把进程看作工厂，资源就是工厂内部的原材料与各种机械设备，而线程就是工厂里的工人。
因此如果进程只有一个执行流，那它也是一个线程，称之为主线程，我们在进程中创建线程其实就是在主线程下创建其他线程，线程与进程的关系如下图：
说完了线程的概念，我们下面来说一下线程的操作。
📖 Linux下的线程
线程和进程是两个独立的概念，在现代操作系统下（例如Windows），线程和进程都是通过内核进行管理和调度，和进程一样，每个线程也有独立的执行上下文，并且可以在内核调度器中进行独立的调度。操作系统的内核负责线程的创建、销毁和调度。
然而在Linux中，线程并不是一个独立实体，而是通过轻量级进程（LWP, Light Weight Process）来实现的，Linux用进程的概念来管理线程，其主要原因有下：
Linux内核本身是设计为面向进程的管理模式，即所有的任务和执行单元都被视为“进程”，而线程的概念是在Linux设计成型之后才提出的，因此为了避免增加内核复杂度，Linux选择将线程视为特殊类型的进程（即LWP），而非引入一个全新的抽象概念。
除了通过内核直接管理线程，我们还可以使用线程库来对线程进行管理：线程库提供了用户空间的接口，能过方便地管理和操作线程；而LWP的实现方式与POSIX线程库的设计要求高度一致，因此在Linux下，我们通常使用POSIX线程库（pthread）实现对线程的创建、销毁、同步等操作。
📚二、线程操作（phread）
POSIX线程库（通常称为 pthreads，即 POSIX Threads）是基于 POSIX 标准的一组接口，用于在程序中创建和管理线程，它提供了一个标准的、跨平台的线程管理框架，广泛应用于多种操作系统中，包括 UNIX、Linux、macOS 以及一些类 Unix 系统。
pthreads 提供了以下主要功能：
📖1.线程创建
线程的创建操作通过 pthread_create() 函数完成。该函数用来创建一个新的线程，并指定其执行的函数和相关的参数。
🔖语法
int pthread_create(pthread_t *thread, const pthread_attr_t *attr, void *(*start_routine)(void *), void *arg);
① 参数1：thread，这是一个指向 pthread_t 类型的指针，用来返回新创建线程的标识符；
❓pthread_t 类型该怎么理解
✅首先我们要清楚，虽然线程之间共享进程的资源，但是线程在被创建时，系统还是会给它分配独立的线程栈、寄存器等状态信息，以确保线程的独立执行，而 pthread_t 类型的变量则用于存储线程的唯一标识符，该标识符指向一个线程控制块（TCB），这个控制块包含了线程的执行状态、栈地址、调度信息等，通过pthread_t，操作系统能够高效地查找和管理线程的上下文。
由于线程控制块TCB存放在虚拟地址空间的共享区中，因此pthread_t实际上存放的就是一个地址，指向共享区上的一块空间。
② 参数2：attr，这是一个指向 pthread_attr_t 类型的指针，指定线程的属性。如果为NULL，则使用线程的默认属性。
③ 参数3：start_routine，这是一个函数指针，指向线程执行的函数。线程启动时将从该函数开始执行。
④ 参数4：arg，这是传递给 start_routine 函数的参数。它是一个 void* 类型的指针，可以传递任何类型的数据，在函数体内部通过强制类型转换对数据接收。
⑤ 返回值：成功时返回 0，失败时返回错误码信息。
🔖示例
这里我们创建一个线程，并循环打印线程 tid：
#include
#include
#include
#include
long gettid(void)
{
return syscall(SYS_gettid);
}
void *func1(void* arg)
{
int i = 1;
while(i)
{
printf("i am pthread 1, tid is %ld
", gettid());
sleep(1);
}
}
int main()
{
pthread_t tid;
if(pthread_create(&tid, NULL, func1, NULL) != 0)
perror("pthread_create:");
printf("thread 1 tid is %lu
",tid);
int i = 1;
while(i)
{
printf("i am main pthread, tid is %ld
", gettid());
sleep(1);
}
return 0;
} 运行结果：
📖2.线程退出
线程退出操作通过 pthread_exit() 完成。当线程执行完自己的任务时，通常会调用 pthread_exit() 来退出，确保线程的资源得到正确释放。
🔖语法
void pthread_exit(void *retval);
参数：retval，表示线程的返回值，主线程可以通过 pthread_join() 获取这个返回值，pthread_join()是线程等待函数，后面会讲解。
返回值：因为线程终止时无法获取自身的返回值，因此该函数没有返回值。
🔖示例
#include
#include
void* thread_func(void *arg) {
printf("Thread is exiting...
");
pthread_exit(NULL);
}
int main() {
pthread_t thread;
// 创建线程
pthread_create(&thread, NULL, thread_func, NULL);
// 等待线程退出
pthread_join(thread, NULL);
printf("Main thread finished.
");
return 0;
}
📖3.线程取消
终止线程还可以通过 pthread_cancel()，pthread_cancel() 用于请求取消某个线程的执行。它是异步的，不会立即终止目标线程，而是向目标线程发送取消请求，目标线程在合适的地方检查这个请求并决定是否退出。
使用流程：
① 调用 pthread_cancel() 向指定线程发送取消请求。
② 目标线程检查取消请求，如果设置了 取消点（即线程可以响应取消请求的地方），线程就会终止。
③ 如果线程不在取消点处，它可能会继续运行，直到它进入一个取消点。
🔖语法
int pthread_cancel(pthread_t thread);
① 参数1：thread，这是目标线程的标识符，即 pthread_create() 返回的线程ID。
② 返回值：成功时，返回 0；失败时，返回错误码（ESRCH 表示线程不存在，EINVAL 表示线程不支持取消等）。
🔖示例
取消线程的执行：
#include
#include
#include
void* thread_func(void *arg) {
printf("Thread started...
");
// 模拟长时间运行的任务
for (int i = 0; i < 5; i++) {
printf("Thread running: %d
", i);
sleep(1);
}
printf("Thread finished.
");
return NULL;
}
int main() {
pthread_t thread;
// 创建线程
pthread_create(&thread, NULL, thread_func, NULL);
// 睡眠2秒后请求取消线程
sleep(2);
printf("Requesting thread cancellation...
");
pthread_cancel(thread);
// 等待线程结束
pthread_join(thread, NULL);
printf("Main thread finished.
");
return 0;
}
在上述示例中，主线程创建了一个子线程，该子线程执行一个循环，每秒打印一次 "Thread running"。主线程睡眠 2 秒后调用 pthread_cancel(thread) 请求取消子线程。此时子线程如果处于取消点，可能会退出。子线程完成后，主线程通过 pthread_join() 等待子线程终止。
运行结果：
📖4.线程等待
线程等待操作通过 pthread_join() 完成。调用 pthread_join() 函数可以让主线程等待子线程执行完毕。
🔖语法
int pthread_join(pthread_t thread, void **retval);
① 参数1：thread，这是要等待的线程的标识符，即 pthread_create() 返回的线程ID。
② 参数2：retval，这是一个指向 void* 类型的指针，用来接收线程的返回值（返回值类型为void* ，所以此处为void** ）。如果不需要返回值，可以传递 NULL。
③ 返回值：成功时，返回 0；失败时，返回错误码。
🔖示例
#include
#include
#include
#include
#include
void *thread1(void *arg)
{
printf("thread1 returning...
");
int *p = (int*)malloc(sizeof(int));
*p = 1;
return (void*)p;
}
void *thread2(void *arg)
{
printf("thread2 exiting...
");
int *p = (int*)malloc(sizeof(int));
*p = 2;
pthread_exit((void*)p);
}
void *thread3(void *arg)
{
while(1)
{
printf("thread3 running...
");
sleep(1);
}
return NULL;
}
int main()
{
pthread_t tid;
void *ret;
// thread1 return
pthread_create(&tid, NULL, thread1, NULL);
pthread_join(tid, &ret);
printf("thread return, tid is %p, return code is %d
", tid, *(int*)ret);
free(ret);
// thread2 exit
pthread_create(&tid, NULL, thread2, NULL);
pthread_join(tid, &ret);
printf("thread exit, tid is %p, return code is %d
", tid, *(int*)ret);
free(ret);
// thread2 exit
pthread_create(&tid, NULL, thread3, NULL);
sleep(3);
pthread_cancel(tid);
pthread_join(tid, &ret);
if(ret == PTHREAD_CANCELED)
printf("thread exit, tid is %p, return code is PTHREAD_CANCELED
", tid);
return 0;
} 我们分别创建3个线程，每创建一个线程，及时对线程等待，回收资源，接着创建下一个线程，由于上一个线程的资源被回收，因此下一个线程可以复用上一个线程的资源，体现在它们的TCB地址是一样的，运行结果：
📖5.分离线程
默认情况下，新创建的线程是 joinable 的（支持被等待），线程退出后，需要对其进行pthread_join操作，否则无法释放资源，从而造成系统泄漏。
但如果我们并不关心线程的返回值，此时join是一种负担，这个时候我们可以告诉系统，当线程退出时自动释放线程资源，也就是将线程分离：
🔖语法
可以是线程组内其他线程对目标线程进行分离：
int pthread_detach(pthread_t thread);① 参数：thread，这是线程标识符，即通过 pthread_create() 创建的线程ID。
② 返回值：成功时，返回 0；失败时，返回错误码。
也可以是线程自己分离:
pthread_detach(pthread_self());pthread_self() 函数用于获取线程自身标识符。
注意❗️
不能同时调用 pthread_detach() 和 pthread_join()：一个线程不能既是joinable又是分离的
🔖示例
#include
#include
void *pthread(void *arg)
{
pthread_detach(pthread_self());
printf("%s
",(char*)arg);
return NULL;
}
int main()
{
pthread_t tid;
pthread_create(&tid, NULL, pthread, "pthread1 running...");
int ret = 0;
sleep(1); // 让线程先分离，再进行等待
if(pthread_join(tid, NULL) == 0)
{
printf("pthread wait success
"); // 说明线程joinable
ret = 0;
}
else
{
printf("pthread wait failed
"); // 说明线程unjoinable
ret = 1;
}
return ret;
} 线程分离之后，调用 pthread_join() 等待该线程就会失败：
📚三、线程互斥
📖 引入
理解线程互斥之前，我们先了解两个概念：临界资源和临界区，多线程执行流共享的资源，就叫做临界资源；每个线程内部，访问临界资源的代码，就叫临界区。
而线程互斥，就是在任何时刻，只能有一个执行流进入临界区，访问临界资源。为什么要这么规定呢？因为多个线程同时对临界资源访问时，可能会导致资源竞争问题，举个例子：
当多个线程同时要对变量i进行++操作时，由于i++操作在底层分为3步：① 从内存提取i到寄存器中；② 在寄存器中完成++操作；③ 将++后的值重新写入到内存中。对应三条汇编指令：
① load：将共享变量 i 从内存加载到寄存器中；
② update：更新寄存器里面的值，执行+1操作；
③ store：将新值，从寄存器写回共享变量 i 的内存地址。
这就会导致一些问题，比如线程1和线程2同时将i提取到各自的寄存器中，线程1先完成了++操作并重写会内存，但是由于这一步并不会更新到线程2的寄存器中，因此虽然变量i被执行了两次++操作，但实际上只有一次。
为了更直观地展示资源竞争现象，我们模拟了一个抢票的流程：
#include
#include
int ticket = 100;
pthread_mutex_t mutex;
void *buy(void *arg)
{
// pthread_detach(pthread_self());
char *id = (char*)arg;
while(1)
{
if(ticket > 0)
{
usleep(1000);
printf("%s sells ticket: %d
", id, ticket);
--ticket;
}
else
{
break;
}
}
}
int main()
{
pthread_t t1, t2, t3, t4;
pthread_mutex_init(&mutex, NULL);
pthread_create(&t1, NULL, buy, "thread 1");
pthread_create(&t2, NULL, buy, "thread 2");
pthread_create(&t3, NULL, buy, "thread 3");
pthread_create(&t4, NULL, buy, "thread 4");
pthread_join(t1, NULL);
pthread_join(t2, NULL);
pthread_join(t3, NULL);
pthread_join(t4, NULL);
pthread_mutex_destroy(&mutex);
// sleep(30);
return 0;
} 这里的临界资源为总票数 ticket，我们创建了4个线程来模拟抢票过程，进入临界区对ticket进行--操作，此时会出现什么结果呢：
我们发现，ticket最终变成了负数，理论上 ticket 变为0时就应该终止了，这就是由于非原子行操作ticket--所造成的资源竞争问题。
要解决上面问题，需要做到三点：
① 代码必须要有互斥行为：当代码进入临界区执行时，不允许其他线程进入该临界区。
② 如果多个线程同时要求执行临界区的代码，并且临界区没有线程在执行，那么只能允许一个线程进入该临界区。
③ 如果线程不在临界区中执行，那么该线程不能阻止其他线程进入临界区。
要做到这三点，本质上就是需要一把锁。Linux上提供的这把锁叫互斥量。
📖 互斥量的接口
🔖1.初始化
初始化互斥量有两种方法：
静态分配：
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER动态分配：
int pthread_mutex_init(pthread_mutex_t *mutex, const pthread_mutexattr_t *attr);
① 参数1：mutex，指向互斥量对象的指针，传入一个未初始化的 pthread_mutex_t 类型变量。
② 参数2：attr，用于设置互斥量属性，通常设置为 NULL，表示使用默认的互斥量属性。
③ 返回值：成功时，返回0；失败时，返回错误码。
❗️注意：如果使用 PTHREAD_MUTEX_INITIALIZER 静态分配互斥量，就不需要调用pthread_mutex_init 函数进行初始化。
🔖2.销毁
线程在不再需要互斥量时，可以调用 pthread_mutex_destroy() 来销毁互斥量。销毁互斥量之前，必须确保没有任何线程持有该互斥量的锁。
int pthread_mutex_destroy(pthread_mutex_t *mutex);
❗️注意：如果使用 PTHREAD_MUTEX_INITIALIZER 静态分配互斥量，就不需要调用pthread_mutex_destroy 函数对其销毁。
🔖3.加锁
线程在访问共享资源之前，需要通过 pthread_mutex_lock() 获得互斥量的锁。若其他线程已持有该互斥量的锁，调用线程会被阻塞，直到互斥量被释放。
int pthread_mutex_lock(pthread_mutex_t *mutex);
① 参数：mutex，指向要锁定的互斥量对象的指针。
② 返回值：成功时，返回0；失败时，返回错误码。
🔖4.解锁
当线程完成对共享资源的访问后，需要释放互斥量的锁。其他线程可以在解锁后获得该锁。
int pthread_mutex_unlock(pthread_mutex_t *mutex);
① 参数：mutex，指向要锁定的互斥量对象的指针。
② 返回值：成功时，返回0；失败时，返回错误码。
🔖示例
我们运用互斥锁来改进上面的售票模拟代码：
#include
#include
int ticket = 100;
pthread_mutex_t mutex;
void *buy(void *arg)
{
// pthread_detach(pthread_self());
char *id = (char*)arg;
while(1)
{
pthread_mutex_lock(&mutex); // 访问临界区前加锁
if(ticket > 0)
{
usleep(1000);
printf("%s sells ticket: %d
", id, ticket);
--ticket;
pthread_mutex_unlock(&mutex); // 操作完成后解锁
}
else
{
pthread_mutex_unlock(&mutex);
break;
}
}
}
int main()
{
pthread_t t1, t2, t3, t4;
pthread_mutex_init(&mutex, NULL);
pthread_create(&t1, NULL, buy, "thread 1");
pthread_create(&t2, NULL, buy, "thread 2");
pthread_create(&t3, NULL, buy, "thread 3");
pthread_create(&t4, NULL, buy, "thread 4");
pthread_join(t1, NULL);
pthread_join(t2, NULL);
pthread_join(t3, NULL);
pthread_join(t4, NULL);
pthread_mutex_destroy(&mutex);
// sleep(30);
return 0;
} ❗️注意：这里我们需要在判断 ticket 语句之前就进行加锁，而不是在++ticket前加锁，因为对ticket的判断语句也属于临界区。执行结果：
ticket在减到0时，所有线程停止访问并退出，符合预期。
📖 死锁
使用互斥锁实现线程间互斥时，如果多个线程之间在获取资源时，发生了相互依赖的情况，导致个线程在互相等待对方释放资源，从而形成了一个无限等待的状态，导致死锁的发生。
🔖四个必要条件
要发生死锁，必须满足以下四个必要条件：
① 互斥条件：一个资源只能由一个线程占用。如果有其他线程请求这个资源，那它必须等待；
② 占用且等待：一个线程至少占有了一个资源，并且等待获取被其他线程占用的资源；
③ 非抢占条件：当线程持有某个资源时，不能强行剥夺其资源，只能等待线程自己释放资源；
④ 循环等待：线程形成一个环形等待关系，其中每个线程都在等待下一个线程释放资源。
🔖示例
假设有两个互斥锁 mutex1 和 mutex2：
#include
#include
pthread_mutex_t mutex1 = PTHREAD_MUTEX_INITIALIZER;
pthread_mutex_t mutex2 = PTHREAD_MUTEX_INITIALIZER;
void* thread1_func(void* arg) {
pthread_mutex_lock(&mutex1);
printf("Thread 1: Locked mutex1
");
// 模拟一些工作
sleep(1);
pthread_mutex_lock(&mutex2);
printf("Thread 1: Locked mutex2
");
pthread_mutex_unlock(&mutex2);
pthread_mutex_unlock(&mutex1);
return NULL;
}
void* thread2_func(void* arg) {
pthread_mutex_lock(&mutex2);
printf("Thread 2: Locked mutex2
");
// 模拟一些工作
sleep(1);
pthread_mutex_lock(&mutex1);
printf("Thread 2: Locked mutex1
");
pthread_mutex_unlock(&mutex1);
pthread_mutex_unlock(&mutex2);
return NULL;
}
int main() {
pthread_t t1, t2;
pthread_create(&t1, NULL, thread1_func, NULL);
pthread_create(&t2, NULL, thread2_func, NULL);
pthread_join(t1, NULL);
pthread_join(t2, NULL);
return 0;
}
在这个例子中：
① 线程1锁住了 mutex1，然后等待 mutex2。
② 线程2锁住了 mutex2，然后等待 mutex1。
这样，两个线程就互相等待对方释放锁，造成了死锁：
🔖解决办法
为了避免死锁，可以采取以下策略：
① 避免占有且等待：尝试在一个线程内一次性获取所有必需的锁，而不是按顺序逐个获取资源。这样，线程不会在持有一个资源的同时等待其他资源。
② 锁的顺序：确保所有线程都按相同的顺序请求锁。这样可以避免循环等待。例如，如果线程都按 mutex1 -> mutex2 的顺序请求锁，则可以避免死锁。
③ 死锁检测和恢复：设计死锁检测算法，定期检查是否存在死锁现象。一旦检测到死锁，可以通过中止线程、回滚或其他方法来恢复。
④ 使用超时机制：在请求锁时，可以设置超时。如果线程在一段时间内未能获取到锁，可以放弃并重新尝试，避免无限等待。
⑤ 使用非阻塞锁：使用诸如 pthread_mutex_trylock 等非阻塞式锁操作，如果锁被占用，线程会立即返回，而不是等待，这样也能避免死锁。
📚四、线程同步
📖 概念
上面讲到，线程互斥保证了任意时刻只有一个线程访问共享资源，避免了资源竞争和冲突，但是这并不能控制线程执行的顺序。因此，它适用于线程之间相对独立、不需要互相协调的场景，例如抢票系统。在这种情况下，多个线程访问共享资源（总票数）时，只需要保证同一时刻只有一个线程修改票数，谁抢到算谁的，不需要额外的线程协作。
但是线程之间有时需要相互协作，例如生产消费者模型。在这种情况下，消费者线程需要等待生产者线程生产数据，才能进行消费；换句话说，消费线程与生产线程之间具有一定的执行顺序，此时如果只用线程互斥并不能满足需求，于是就需要引入线程同步机制：
线程同步是在线程互斥的基础上，进一步控制线程的执行顺序，确保线程之间在特定的时刻能够按规定的顺序访问资源。线程同步使用于需要线程协作的场景，例如生产消费者模型、读者写者模型等等。线程同步不仅确保线程在访问共享资源时不会发生冲突，还控制了线程之间的执行顺序，保证程序逻辑的正确性。
下面我们来介绍线程同步的实现方法。
📖1.条件变量
线程可以通过条件变量等待特定条件的发生，并在条件满足时被唤醒。条件变量常用于生产者消费者模型，生产者线程生产数据，消费者线程消费数据，确保它们在正确的时机执行。
🔖基本概念
条件变量（Condition Variable）允许线程在特定条件下进入等待状态，并在条件满足时被唤醒。它通常与 互斥锁 一起使用，以保证对共享资源的安全访问。
🔖语法
int pthread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex);
int pthread_cond_signal(pthread_cond_t *cond);
int pthread_cond_broadcast(pthread_cond_t *cond);
① pthread_cond_wait
让调用线程阻塞，直到被其他线程通过 pthread_cond_signal 或 pthread_cond_broadcast 唤醒。它需要一个互斥锁作为参数，以确保在等待条件时对共享资源的访问是安全的。
② pthread_cond_signal
唤醒一个在条件变量上等待的线程。如果有多个线程在等待，系统随机选择一个线程唤醒。
③ pthread_cond_broadcast
唤醒所有在条件变量上等待的线程。
❓ 如何理解条件变量的工作原理？
✅ 当线程调用 pthread_cond_wait 时，它会释放与条件变量相关联的互斥锁，并进入等待状态。当另一个线程调用 pthread_cond_signal 或 pthread_cond_broadcast 时，等待的线程会被唤醒。唤醒后，线程会重新获取互斥锁，然后继续执行，保证了线程之间的协调，避免了数据竞争。
🔖示例：生产消费者模型
#include
using namespace std;
#include
#include
#include
#include
#define NUM 8
class BlockQueue{
private:
queue q;
int cap = NUM;
pthread_mutex_t lock;
pthread_cond_t full;
pthread_cond_t empty;
void QueueLock(){
pthread_mutex_lock(&lock);
}
void QueueUnlock(){
pthread_mutex_unlock(&lock);
}
void ProductWait(){
pthread_cond_wait(&full, &lock);
}
void ConsumeWait(){
pthread_cond_wait(&empty, &lock);
}
void NotifyProcduct(){
pthread_cond_signal(&full);
}
void NotifyConsume(){
pthread_cond_signal(&empty);
}
bool IsEmpty()
{
if(q.size() == 0) return true;
else return false;
}
bool IsFull()
{
if(q.size() == cap) return true;
else return false;
}
public:
BlockQueue(){
pthread_mutex_init(&lock, nullptr);
pthread_cond_init(&full, nullptr);
pthread_cond_init(&empty, nullptr);
}
~BlockQueue(){
pthread_mutex_destroy(&lock);
pthread_cond_destroy(&full);
pthread_cond_destroy(&empty);
}
void PushData(const int &data){
QueueLock();
while(IsFull()){
cout << "Queue is full, product is waiting..." << endl;
ProductWait();
}
q.push(data);
NotifyConsume();
QueueUnlock();
}
void PopData(int &data){
QueueLock();
while(IsEmpty()){
NotifyProcduct();
cout << "Queue is empty, consume is waiting..." << endl;
ConsumeWait();
}
data = q.front();
q.pop();
NotifyProcduct();
QueueUnlock();
}
};
void *Procduct(void *arg){
BlockQueue *q = (BlockQueue*)arg;
srand((unsigned long)time(nullptr));
while(1){
int data = rand() % 1024;
q->PushData(data);
cout << "Produce data done: " << data << endl;
// sleep(1);
}
}
void *Consume(void *arg){
BlockQueue *q = (BlockQueue*)arg;
while(1){
int data;
q->PopData(data);
cout << "Consume data done: " << data << endl;
// sleep(1);
}
}
int main()
{
BlockQueue q;
pthread_t t1, t2;
pthread_create(&t1, nullptr, Procduct, &q);
pthread_create(&t2, nullptr, Consume, &q);
pthread_join(t1, nullptr);
pthread_join(t2, nullptr);
return 0;
} 我们创建了一个消费者线程和一个生产者线程，共享资源为容量8的队列：
① 当队列为空时，消费线程阻塞等待生产线程生产数据，一旦生产者生产了数据，就通知唤醒消费线程消费数据；
② 当队列为满时，生产线程阻塞等待消费线程消费数据，一旦消费线程消费了数据，就通知唤醒生产线程生产数据......
部分运行结果：
📖2.信号量
上面的生产消费者模型是基于queue队列模拟实现，其空间大小是动态分配的，我们判断生产消费线程阻塞等待的情况分别是队列为满与队列为空，都可以直接用内置函数size()进行判断。但是如果数据存储的空间大小是一定的，队列为空为满就不好直接判断，此时需要借助信号量，来实时记录当前空间大小。
🔖基本概念
信号量（Semaphore）维护一个计数值，线程在访问资源前需要检查信号量的值。当信号量的值大于零时，线程可以进入临界区并访问资源；当信号量的值为零时，线程会被阻塞，直到信号量的值增加。在生产消费者模型中，信号量表示为当前剩余空间以及数据量大小。
🔖语法
int pthread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex);
int pthread_cond_signal(pthread_cond_t *cond);
int pthread_cond_broadcast(pthread_cond_t *cond);
① pthread_cond_wait
让调用线程阻塞，直到被其他线程通过 pthread_cond_signal 或 pthread_cond_broadcast 唤醒。它需要一个互斥锁作为参数，以确保在等待条件时对共享资源的访问是安全的。
② pthread_cond_signal
唤醒一个在条件变量上等待的线程。如果有多个线程在等待，系统随机选择一个线程唤醒。
③ pthread_cond_broadcast
唤醒所有在条件变量上等待的线程。
❓ 如何理解条件变量的工作原理？
✅ 当线程调用 pthread_cond_wait 时，它会释放与条件变量相关联的互斥锁，并进入等待状态。当另一个线程调用 pthread_cond_signal 或 pthread_cond_broadcast 时，等待的线程会被唤醒。唤醒后，线程会重新获取互斥锁，然后继续执行，同样保证了线程之间的协调，避免了数据竞争。
❗️注意：信号量的操作（ sem_wait() 和 sem_post()）在实现时是由操作系统提供的原子操作。也就是说，信号量的计数值是通过原子操作进行修改的，这些操作是不可中断的，保证在多线程环境下操作信号量时不会出现竞争，因此在使用信号量时不需要用互斥锁来保护对信号量的修改。
🔖示例：基于环形队列的PC-model
#include
using namespace std;
#include
#include
#include
#include
#include
#define NUM 8
class RingQueue{
private:
vector q;
int cap = NUM;
sem_t data_sem;
sem_t space_sem;
int product_step;
int consume_step;
public:
RingQueue(){
q.resize(cap);
sem_init(&data_sem, 0, 0);
sem_init(&space_sem, 0, cap);
product_step = 0;
consume_step = 0;
}
~RingQueue(){
sem_destroy(&data_sem);
sem_destroy(&space_sem);
}
void PushData(const int &data){
sem_wait(&space_sem);
q[product_step] = data;
++product_step;
product_step %= cap;
sem_post(&data_sem);
}
void PopData(int &data){
sem_wait(&data_sem);
data = q[consume_step];
++consume_step;
consume_step %= cap;
sem_post(&space_sem);
}
};
void *Produce(void *arg){
RingQueue *q = (RingQueue*)arg;
srand((unsigned long)time(nullptr));
while(1){
int data = rand() % 1024;
q->PushData(data);
cout << "Produce data done: " << data << endl;
// sleep(1);
}
}
void *Consume(void *arg){
RingQueue *q = (RingQueue*)arg;
while(1){
int data;
q->PopData(data);
cout << "Consume data done: " << data << endl;
// sleep(1);
}
}
int main()
{
RingQueue q;
pthread_t t1, t2;
pthread_create(&t1, nullptr, Produce, &q);
pthread_create(&t2, nullptr, Consume, &q);
pthread_join(t1, nullptr);
pthread_join(t2, nullptr);
return 0;
} 创建了一个消费者线程和一个生产者线程，共享源为容量8的环形队列：
① 当数据信号量为0时，表示当前没有数据，消费线程阻塞等待，生产线程生产数据之后，增加数据信号量，此时消费线程会被唤醒；
② 当空间信号量为0时，表示当前没有空间，生产线程阻塞等待，消费线程消费数据之后，增加空间信号量，此时生产线程会被唤醒......
部分运行结果：
📚五、线程池
相比于多进程，多线程大大减少了空间开辟和释放的开销，但并不意味着多线程的创建和销毁的开销可以忽略不计，在面对超高并发情况，线程的频繁创建和销毁还是会带来巨大的开销，如果线程一次性创建过多，甚至会导致资源耗尽，系统崩溃的情况，如春运期间铁路12306的访问量非常大，频繁创建和销毁线程的开销会变得十分严重。为了应对这种情况，便引入了线程池技术：
📖 概念
线程池是一种线程管理机制，它通过维护一个预先创建的线程集合（线程池），来避免频繁地创建和销毁线程所带来的性能开销。线程池管理一定数量的工作线程，将任务提交给空闲线程来执行，执行完成后线程会返回池中等待下一个任务。这种方式可以提高系统性能，尤其在高并发场景下，通过复用线程来减少线程创建和销毁的成本。
线程池通常管理着以下几个重要组件：
① 线程池中的工作线程：执行实际的任务
② 任务队列：存放等待执行的任务
③ 核心线程数：线程池中始终保持活跃的线程数量
④ 核心线程数：线程池中始终保持活跃的线程数量
⑤ 线程池大小的动态调整：根据负载和任务量，线程池可能会动态增减线程数量
📖 实现
下面介绍简易线程池的具体实现：
🔖1.初始化
在创建线程池时，线程池会创建一定数量的线程，这些线程将用来执行任务。线程池的初始化通过 PoolInit() 函数完成。
bool PoolInit() {
pthread_t tid;
for (int i = 0; i < _thread_max; ++i) {
int ret = pthread_create(&tid, NULL, thr_start, this);
if (ret != 0) {
cout << "create pool thread error" << endl;
return false;
}
}
return true;
}
① PoolInit()：初始化线程池，创建指定数量的工作线程（_thread_max）。每个线程执行 thr_start() 函数。
② pthread_create()：用来创建线程，每个线程将运行 thr_start 函数。thr_start 会一直循环地从任务队列中取出任务并执行，直到线程池退出。
🔖2.任务队列
任务队列是线程池中的关键组件之一。它用于存储待处理的任务，工作线程从队列中取出任务并执行。
queue _task_queue;
这里使用了 std::queue 来实现任务队列。ThreadTask 是任务对象，每个任务都包含一个待处理的数据和对应的处理函数；当任务被推送到队列时，工作线程会从队列中弹出并执行。
队列相关操作：① PushTask：将任务加入队列；② PopTask：从队列中弹出一个任务。
bool PushTask(ThreadTask *tt) {
LockQueue();
if (_tp_isquit) {
UnLockQueue();
return false;
}
_task_queue.push(tt);
WakeUpOne();
UnLockQueue();
return true;
}
bool PopTask(ThreadTask **tt) {
*tt = _task_queue.front();
_task_queue.pop();
return true;
}
🔖3.线程同步
线程池中的多个线程需要访问共享资源（即任务队列），因此需要使用 互斥锁 来保护共享资源的访问。条件变量 用于线程间的同步，控制线程何时等待，何时被唤醒。
pthread_mutex_t _lock;
pthread_cond_t _cond;
① _lock：互斥锁，用来保护任务队列 _task_queue，避免多个线程同时修改队列；
② _cond：条件变量，用来通知线程池中的工作线程何时可以执行任务。
线程同步操作：
① LockQueue 和 UnLockQueue：用于加锁和解锁任务队列，确保对任务队列的访问是线程安全的
② WakeUpOne 和 WakeUpAll：分别唤醒一个等待的线程或所有等待的线程
③ ThreadWait：让当前线程阻塞，直到任务队列中有任务或线程池退出
void LockQueue() {
pthread_mutex_lock(&_lock);
}
void UnLockQueue() {
pthread_mutex_unlock(&_lock);
}
void WakeUpOne() {
pthread_cond_signal(&_cond);
}
void WakeUpAll() {
pthread_cond_broadcast(&_cond);
}
void ThreadWait() {
if (_tp_isquit) {
ThreadQuit();
}
pthread_cond_wait(&_cond, &_lock);
}
🔖4.工作线程
每个工作线程都执行 thr_start 函数，该函数包含一个无限循环，不断从任务队列中取出任务并执行。直到线程池退出。
static void *thr_start(void *arg) {
ThreadPool *tp = (ThreadPool*)arg;
while (1) {
tp->LockQueue();
while (tp->IsEmpty()) {
tp->ThreadWait();
}
ThreadTask *tt;
tp->PopTask(&tt);
tp->UnLockQueue();
tt->Run();
delete tt;
}
return NULL;
}
① 每个工作线程在 thr_start 中会一直等待任务；
② 如果任务队列为空，线程会通过 ThreadWait 阻塞，直到有任务被推送到队列中；
③ 任务一旦加入队列，线程会被唤醒，并执行任务；
④ 行完任务后，线程会回到等待状态，直到线程池退出。
🔖5.线程池退出
当线程池不再接受新任务并且所有任务执行完毕时，线程池需要退出。线程池的退出通过 PoolQuit 函数来完成。
bool PoolQuit() {
LockQueue();
_tp_isquit = true;
UnLockQueue();
while (_thread_cur > 0) {
WakeUpAll();
usleep(1000);
}
return true;
}
① PoolQuit：标志 _tp_isquit 被设置为 true，然后唤醒所有线程并让它们检查退出条件；
② 工作线程在执行 ThreadWait 时会检查 _tp_isquit，如果为 true，则退出。
🔖总结
1.线程池的初始化：线程池通过 PoolInit 创建多个工作线程，这些线程通过执行 thr_start 来不断地获取并处理任务。
2.任务队列：任务队列是线程池的重要组成部分，用于存储待处理的任务，任务由主线程推送到队列，工作线程从队列中取出并执行。
3.线程同步机制：互斥锁和条件变量用于确保线程安全地访问任务队列，并协调工作线程的等待和唤醒。
4.工作线程执行任务：每个工作线程都从队列中获取任务，并执行任务，直到线程池退出。
5.线程池退出：当线程池不再接收任务时，调用 PoolQuit 函数退出线程池，并通知所有工作线程退出。
完整代码如下：
// threadpool.hpp
#ifndef __M_TP_H__
#define __M_TP_H__
#include
using namespace std;
#include
#include
#include
#include
#define MAX_THREAD 5
typedef bool (*hander_t)(int);
class ThreadTask
{
private:
int _data;
hander_t _handler;
public:
ThreadTask():_data(-1), _handler(NULL){}
ThreadTask(int data, hander_t handler){
_data = data;
_handler = handler;
}
void SetTask(int data, hander_t handler){
_data = data;
_handler = handler;
}
void Run(){
_handler(_data);
}
};
class ThreadPool
{
private:
int _thread_max;
int _thread_cur;
bool _tp_isquit;
queue _task_queue;
pthread_mutex_t _lock;
pthread_cond_t _cond;
private:
void LockQueue(){
pthread_mutex_lock(&_lock);
}
void UnLockQueue(){
pthread_mutex_unlock(&_lock);
}
void WakeUpOne(){
pthread_cond_signal(&_cond);
}
void WakeUpAll(){
pthread_cond_broadcast(&_cond);
}
void ThreadQuit(){
_thread_cur--;
UnLockQueue();
pthread_exit(NULL);
}
void ThreadWait(){
if(_tp_isquit){
ThreadQuit();
}
pthread_cond_wait(&_cond, &_lock);
}
bool IsEmpty(){
return _task_queue.empty();
}
static void *thr_start(void *arg){
ThreadPool *tp = (ThreadPool*)arg;
while(1){
tp->LockQueue();
while(tp->IsEmpty()){
tp->ThreadWait();
}
ThreadTask *tt;
tp->PopTask(&tt);
tp->UnLockQueue();
tt->Run();
delete tt;
}
return NULL;
}
public:
ThreadPool(int max=MAX_THREAD):_thread_max(max), _thread_cur(max),
_tp_isquit(false){
pthread_mutex_init(&_lock, NULL);
pthread_cond_init(&_cond, NULL);
}
~ThreadPool(){
pthread_mutex_destroy(&_lock);
pthread_cond_destroy(&_cond);
}
bool PoolInit(){
pthread_t tid;
for(int i = 0; i < _thread_max; ++i){
int ret = pthread_create(&tid, NULL, thr_start, this);
if(ret != 0){
cout << "create pool thread error" << endl;
return false;
}
}
return true;
}
bool PushTask(ThreadTask *tt){
LockQueue();
if(_tp_isquit){
UnLockQueue();
return false;
}
_task_queue.push(tt);
WakeUpOne();
UnLockQueue();
return true;
}
bool PopTask(ThreadTask **tt){
*tt = _task_queue.front();
_task_queue.pop();
return true;
}
bool PoolQuit(){
LockQueue();
_tp_isquit = true;
UnLockQueue();
while(_thread_cur > 0){
WakeUpAll();
usleep(1000);
}
return true;
}
};
#endif // main.cpp
#include "threadpool.hpp"
bool handler(int data)
{
srand(time(NULL));
int n = rand() % 5;
printf("Thread: %p Run Tast: %d--sleep %d sec
", pthread_self(), data, n);
sleep(n);
return true;
}
int main()
{
ThreadPool pool;
pool.PoolInit();
for(int i = 0; i < 10; ++i){
ThreadTask *tt = new ThreadTask(i, handler);
pool.PushTask(tt);
}
pool.PoolQuit();
return 0;
}运行结果：
以上就是【深入理解线程：概念、操作、互斥与同步机制、线程池实现】的全部内容，欢迎指正~
码文不易，还请多多关注支持，这是我持续创作的最大动力！
