【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 <stdio.h>
#include <pthread.h>
#include <unistd.h>
#include <sys/syscall.h>

long gettid(void)
{
    return syscall(SYS_gettid);
}

void *func1(void* arg)
{
    int i = 1;
    while(i)
    {
        printf("i am pthread 1, tid is %ld\n", 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\n",tid);
    int i = 1;
    while(i)
    {
        printf("i am main pthread, tid is %ld\n", gettid());
        sleep(1);
    }
    return 0;
}

运行结果：

📖2.线程退出

线程退出操作通过 pthread_exit() 完成。当线程执行完自己的任务时，通常会调用 pthread_exit() 来退出，确保线程的资源得到正确释放。

🔖语法

void pthread_exit(void *retval);

参数：retval，表示线程的返回值，主线程可以通过 pthread_join() 获取这个返回值，pthread_join()是线程等待函数，后面会讲解。

返回值：因为线程终止时无法获取自身的返回值，因此该函数没有返回值。

🔖示例

#include <pthread.h>
#include <stdio.h>

void* thread_func(void *arg) {
    printf("Thread is exiting...\n");
    pthread_exit(NULL);
}

int main() {
    pthread_t thread;
    
    // 创建线程
    pthread_create(&thread, NULL, thread_func, NULL);
    
    // 等待线程退出
    pthread_join(thread, NULL);
    
    printf("Main thread finished.\n");
    return 0;
}

📖3.线程取消

终止线程还可以通过 pthread_cancel()，pthread_cancel() 用于请求取消某个线程的执行。它是异步的，不会立即终止目标线程，而是向目标线程发送取消请求，目标线程在合适的地方检查这个请求并决定是否退出。

使用流程：

① 调用 pthread_cancel() 向指定线程发送取消请求。

② 目标线程检查取消请求，如果设置了 取消点（即线程可以响应取消请求的地方），线程就会终止。

③ 如果线程不在取消点处，它可能会继续运行，直到它进入一个取消点。

🔖语法

int pthread_cancel(pthread_t thread);

① 参数1：thread，这是目标线程的标识符，即 pthread_create() 返回的线程ID。

② 返回值：成功时，返回 0；失败时，返回错误码（ESRCH 表示线程不存在，EINVAL 表示线程不支持取消等）。

🔖示例

取消线程的执行：

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

void* thread_func(void *arg) {
    printf("Thread started...\n");
    
    // 模拟长时间运行的任务
    for (int i = 0; i < 5; i++) {
        printf("Thread running: %d\n", i);
        sleep(1);
    }
    
    printf("Thread finished.\n");
    return NULL;
}

int main() {
    pthread_t thread;
    
    // 创建线程
    pthread_create(&thread, NULL, thread_func, NULL);
    
    // 睡眠2秒后请求取消线程
    sleep(2);
    printf("Requesting thread cancellation...\n");
    pthread_cancel(thread);
    
    // 等待线程结束
    pthread_join(thread, NULL);
    
    printf("Main thread finished.\n");
    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 <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <pthread.h>

void *thread1(void *arg)
{
    printf("thread1 returning...\n");
    int *p = (int*)malloc(sizeof(int));
    *p = 1;
    return (void*)p;
}

void *thread2(void *arg)
{
    printf("thread2 exiting...\n");
    int *p = (int*)malloc(sizeof(int));
    *p = 2;
    pthread_exit((void*)p);
}

void *thread3(void *arg)
{
    while(1)
    {
        printf("thread3 running...\n");
        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\n", 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\n", 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\n", 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 <stdio.h>
#include <pthread.h>

void *pthread(void *arg)
{
    pthread_detach(pthread_self());
    printf("%s\n",(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\n"); // 说明线程joinable
        ret = 0;
    }
    else
    {
        printf("pthread wait failed\n"); // 说明线程unjoinable
        ret = 1;        
    }
    return ret;
}

线程分离之后，调用 pthread_join() 等待该线程就会失败：

📚三、线程互斥

📖 引入

理解线程互斥之前，我们先了解两个概念：临界资源和临界区，多线程执行流共享的资源，就叫做临界资源；每个线程内部，访问临界资源的代码，就叫临界区。

而线程互斥，就是在任何时刻，只能有一个执行流进入临界区，访问临界资源。为什么要这么规定呢？因为多个线程同时对临界资源访问时，可能会导致资源竞争问题，举个例子：

当多个线程同时要对变量i进行++操作时，由于i++操作在底层分为3步：① 从内存提取i到寄存器中；② 在寄存器中完成++操作；③ 将++后的值重新写入到内存中。对应三条汇编指令：

① load：将共享变量 i 从内存加载到寄存器中；

② update：更新寄存器里面的值，执行+1操作；

③ store：将新值，从寄存器写回共享变量 i 的内存地址。

这就会导致一些问题，比如线程1和线程2同时将i提取到各自的寄存器中，线程1先完成了++操作并重写会内存，但是由于这一步并不会更新到线程2的寄存器中，因此虽然变量i被执行了两次++操作，但实际上只有一次。

为了更直观地展示资源竞争现象，我们模拟了一个抢票的流程：

#include <stdio.h>
#include <pthread.h>

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\n", 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 <stdio.h>
#include <pthread.h>

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\n", 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 <pthread.h>
#include <stdio.h>

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\n");

    // 模拟一些工作
    sleep(1);

    pthread_mutex_lock(&mutex2);
    printf("Thread 1: Locked mutex2\n");

    pthread_mutex_unlock(&mutex2);
    pthread_mutex_unlock(&mutex1);

    return NULL;
}

void* thread2_func(void* arg) {
    pthread_mutex_lock(&mutex2);
    printf("Thread 2: Locked mutex2\n");

    // 模拟一些工作
    sleep(1);

    pthread_mutex_lock(&mutex1);
    printf("Thread 2: Locked mutex1\n");

    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 <iostream>
using namespace std;
#include <queue>
#include <pthread.h>
#include <stdlib.h>
#include <unistd.h>

#define NUM 8

class BlockQueue{
private:
    queue<int> 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 <iostream>
using namespace std;
#include <stdlib.h>
#include <unistd.h>
#include <semaphore.h>
#include <pthread.h>
#include <vector>

#define NUM 8

class RingQueue{
private:
    vector<int> 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<ThreadTask *> _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 <iostream>
using namespace std;
#include <queue>
#include <pthread.h>
#include <stdlib.h>
#include <unistd.h>

#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<ThreadTask *> _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\n", 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;
}

运行结果：