基于linux的基础线程知识大总结

文章目录

- - 1.线程的基础概念认知
  - - - 1.1什么是线程
      - 1.2线程的优缺点
      - 1.3一些页表知识的额外补充
      - 1.4进程和线程的对比
  - 2.线程的基本控制
  - - - 2.1POSIX线程库
      - 2.2创建一个新的线程
      - 2.3有关线程id的解释和线程栈区的地址空间布局
      - 2.4线程终止
      - 2.5线程等待
      - 2.6线程分离
  - 3.线程间互斥
  - - - 3.1基本概念
      - 3.2互斥的情景模拟
      - 3.3互斥量
      - 3.3.1初始化互斥量
        3.3.2互斥量销毁
        3.3.3互斥量加锁与解锁
        3.3.4演示例子
      - 3.4互斥量的实现原理
      - 3.5死锁
  - 4.linux线程同步
  - - - 4.1基本概念
      - 4.2条件变量函数
      - 初始化条件变量
        等待条件变量的满足
        唤醒等待
        销毁条件变量对象，并释放相关的资源。
        简单实例的应用
  - 4.生产者消费者模型
  - - - 4.1基本原理
      - 4.2基于BlockingQueue的生产者消费者模型
      - 4.3POSIX信号量
      - 初始化信号量
        销毁信号量
        等待信号量
        发布信号量
      - 4.4基于环形队列的生产消费模型
  - 5.线程池
  - - - 5.1基本概念
      - 5.2经典线程池代码案例

1.线程的基础概念认知

1.1什么是线程

好一上来我们也别像网上其他文章一样巴拉巴拉一堆废话，之间给出我总结后最为精简核心的概念：

线程在进程内部执行是进程调度的基本单位。

我知道大家看到这句话肯定还是会有很多疑问那么接下了我将为大家进一步讲解其中核心：
在这里插入图片描述

如图所示操作系统创建了一个个进程结构体来公用一片地址空间，而这第一个创建帮忙分配虚拟地址空间的进程，成为主线程，其下管理同一片虚拟地址空间的称之为轻量级别线程（注意linux下进程和线程公用同一数据结构而windows下会区分开来）。

ps：线程是进程的执行分支，线程出异常，就类似进程出异常，进而触发信号机制，终止进程，进程终止，该

进程内的所有线程也就随即退出

1.2线程的优缺点

线程优点：并发执行、资源共享、数据共享、轻量级。
线程缺点：同步问题、线程安全性、调试困难、资源竞争。

1.3一些页表知识的额外补充

在这里插入图片描述

正常来说我们会用虚拟内存来通过页表（通常大小为4kb）来映射实际内存，而当发生这一切时我们的页表实际上是要加载到物理内存上的，所以大小不能太大，因此采用了多级页表的形式，如上图所示，利用这样一个位图的结构存储可以表达出足够的映射的关系，从而来完成映射过程。

1.4进程和线程的对比

进程是资源分配的基本单位

线程是调度的基本单位

线程共享进程数据，但也拥有自己的一部分数据:

线程ID
一组寄存器
栈
errno
信号屏蔽字
调度优先级

进程的多个线程共享同一地址空间,因此Text Segment、Data Segment都是共享的,如果定义一个函数,在各线程

中都可以调用,如果定义一个全局变量,在各线程中都可以访问到,除此之外,各线程还共享以下进程资源和环境:

文件描述符表
每种信号的处理方式(SIG_ IGN、SIG_ DFL或者自定义的信号处理函数)
当前工作目录
用户id和组id

2.线程的基本控制

2.1POSIX线程库

首先要使用一些有关线程的基本函数，我们就不得不提到POSIX库。

与线程有关的函数构成了一个完整的系列，绝大多数函数的名字都是以“pthread_”打头的
要使用这些函数库，要通过引入头文<pthread.h>
链接这些线程函数库时要使用编译器命令的“-lpthread”选项

2.2创建一个新的线程

#include <pthread.h>

int pthread_create(pthread_t *thread, const pthread_attr_t *attr, void *(*start_routine) (void *), void *arg);


参数解释：

thread：指向pthread_t类型的指针，用于存储新线程的标识符。
attr：指向pthread_attr_t类型的指针，用于指定线程的属性。可以传递NULL，表示使用默认属性。
start_routine：指向线程函数的指针，该函数是新线程启动后要执行的函数。函数指针的定义为void *(*start_routine) (void *)，它接受一个void*类型的参数并返回一个void*类型的指针。
arg：传递给线程函数的参数，可以是任意类型的指针。
返回值：

pthread_create函数成功时返回0，表示线程创建成功。
如果出现错误，返回一个非零的错误代码，指示错误类型。

基础的使用例子：

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

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

void* thread_function(void* arg) {
    int thread_id = *(int*)arg;
    printf("Hello from thread %d\n", thread_id);
    pthread_exit(NULL);
}

int main() {
    pthread_t thread_id;
    int thread_arg = 1;
    int result = pthread_create(&thread_id, NULL, thread_function, (void*)&thread_arg);
    if (result != 0) {
        printf("Thread creation failed\n");
        return 1;
    }

    printf("Hello from the main thread\n");

    pthread_join(thread_id, NULL);  // 等待子线程结束

    return 0;
}

输出：
Hello from the main thread
Hello from thread 1

2.3有关线程id的解释和线程栈区的地址空间布局

相信聪明的大家不难发现，个线程间虽然共用同一块虚拟地址空间但他们的栈从设计上来讲必须被封开，这是怎么做到的呢？请看下面这张图：

在这里插入图片描述

原来在Linux下，有一个帮助线程划分栈区空间的动态库，它被叫作libpthread，也称为POSIX线程库

前面所说的函数pthread_ create函数会产生一个线程ID，存放在第一个参数指向的地址中。该线程ID和前面说的线程ID 不是一回事。

前面讲的线程ID属于进程调度的范畴。因为线程是轻量级进程，是操作系统调度器的最小单位，所以需要

一个数值来唯一表示该线程。

pthread_ create函数第一个参数指向一个虚拟内存单元，该内存单元的地址即为新创建线程的线程ID，

属于NPTL线程库的范畴。线程库的后续操作，就是根据该线程ID来操作线程的。

ps：NPTL是Linux特定的线程库实现，它是为了改进Linux上线程模型的性能而开发的。而POSIX线程是一个跨平台的线程标准，定义了线程的API和行为，允许在不同的操作系统上开发具有一致性的多线程程序。就比如上述那个开始创建线程的函数，pthread_create函数是属于POSIX线程库的函数，可在符合POSIX标准的操作系统上使用。在Linux系统中，NPTL库实现了POSIX线程库的功能，并提供了更好的性能和扩展性，所以pthread_create函数亦是NPTL库实现的一部分，可以用于创建新线程。

当然linux也提供了查看自己线程id的函数：

pthread_t pthread_self(void);
//pthread_self函数没有参数，它返回一个pthread_t类型的值，表示当前线程的线程ID（实际上就是上面说的虚拟地址空间上的一个地址（这是基础linux的NPTL这个库具体的实现方式而言））。

2.4线程终止

如果想要只终止某个线程而不是整个进程，那么有三种方法：

从线程函数return。这种方法对主线程不适用,从main函数return相当于调用exit。
线程可以调用pthread_ exit终止自己。
一个线程可以调用pthread_ cancel终止同一进程中的另一个线程。

//用于终止当前线程的执行并返回一个线程退出状态。
void pthread_exit(void *value_ptr);

参数：value_ptr不要指向一个局部变量(算是一种退出状态)。
返回值：无返回值，跟进程一样，线程结束的时候无法返回到它的调用者（自身）

需要注意,pthread_exit或者return返回的指针所指向的内存单元必须是全局的或者是用malloc分配的,不能在线程函数的栈上分配,因为当其它线程得到这个返回指针时线程函数已经退出了。

功能：取消一个执行中的线程

原型
int pthread_cancel(pthread_t thread);

参数
thread:线程ID   //可以使用上面介绍的pthread_self函数获取

返回值：成功返回0；失败返回错误码

2.5线程等待

线程的等待机制是一种有效的方法，可以使多个线程之间协调执行，避免竞争条件和资源冲突。它允许线程在合适的时机进行等待和唤醒，以便实现线程间的同步和通信。

相关函数：

//功能：调用该函数的线程将挂起等待,直到id为thread的线程终止。

#include <pthread.h>

int pthread_join(pthread_t thread, void **retval);
 
  pthread_join() 函数接受两个参数：thread 是要等待的线程的标识符，retval 是一个指向指针的指针，它用于接收被等待线程的返回值。

//下面是 pthread_join() 函数的一些重要特点和用法：

1.阻塞等待：调用 pthread_join() 函数的线程将被阻塞，直到被等待的线程结束。一旦被等待的线程结束，调用线程将继续执行。

2.获取返回值：通过 retval 参数，可以获取被等待线程的返回值。被等待线程的返回值必须是一个指针类型（void*），因此 retval 是一个指向指针的指针。

3.线程终止：被等待的线程可以通过 pthread_exit() 函数或从线程函数中返回来终止。在线程终止时，它可以通过 pthread_exit() 传递一个返回值。

4.线程标识符：pthread_t 类型的 thread 参数表示要等待的线程标识符。在线程创建时，可以通过 pthread_create() 函数获取线程标识符。

5.线程同步：pthread_join() 函数常用于线程同步，以确保在主线程中等待所有子线程结束后再继续执行后续的操作。

2.6线程分离

在多线程编程中，线程分离（Thread Detach）是一种机制，用于告诉操作系统在线程结束时自动释放其资源，而无需调用 pthread_join() 来等待线程的结束和获取返回值。(一般用于不关注返回值的情况)

（Joinable（可连接）状态：当一个线程被创建时，默认情况下它是可连接状态。可连接状态的线程可以被其他线程通过调用 pthread_join() 函数等待其结束，并获取其返回值，如果不进行pthread_join操作，否则无法释放资源，从而造成系统泄漏。）

相关函数：

//pthread_detach() 函数用于将一个线程设置为分离状态，从而告诉操作系统在线程结束时自动释放其资源

#include <pthread.h>

int pthread_detach(pthread_t thread);
函数返回值为 0 表示成功，非零值表示失败
可以是线程组内其他线程对目标线程进行分离，也可以是线程自己分离
Joinable状态和Detached状态是冲突的，一个线程不能既是Joinable状态又是Detached状态

3.线程间互斥

3.1基本概念

临界资源：多线程执行流共享的资源就叫做临界资源
临界区：每个线程内部，访问临界资源的代码，就叫做临界区
互斥：任何时刻，互斥保证有且只有一个执行流进入临界区，访问临界资源，通常对临界资源起保护作用
原子性（后面讨论如何实现）：不会被任何调度机制打断的操作，该操作只有两态，要么完成，要么未完
成

3.2互斥的情景模拟

当同一时间创建多个线程时，操作系统会通过时间片轮转调度的方式，在各个线程之间进行快速切换，使得它们看起来是同时执行的。

想象一下以下情景,有两个线程T1,T2访问同一个临界资源ticket进行–操作，当T1正在把ticket的数据放到cpu的寄存器上运算而没有将其更新到内存，正在这时T2访问了这个临界资源，其将cpu寄存器上的数据在进行了一遍运算，并将其更新到了内存上，造成了问题的出现。

这里所说的–操作并不是原子性的（下面附上–对应的三条汇编指令）

取出ticket--部分的汇编代码
objdump -d a.out > test.objdump
152 40064b: 8b 05 e3 04 20 00 mov 0x2004e3(%rip),%eax # 600b34 <ticket>
153 400651: 83 e8 01 sub $0x1,%eax
154 400654: 89 05 da 04 20 00 mov %eax,0x2004da(%rip) # 600b34 <ticket>

load ：将共享变量ticket从内存加载到寄存器中
update : 更新寄存器里面的值，执行-1操作
store ：将新值，从寄存器写回共享变量ticket的内存地址

3.3互斥量

那么如何解决这一问题呢？
linux提供了一把锁，我们称之为互斥量

在这里插入图片描述

相信看了上面这一张图，大家因该已经大致明白了互斥量的作用了吧，那么废话不多说我们接着介绍一下有关互斥量的相关接口。

3.3.1初始化互斥量

互斥量的初始化可以通过两种方式进行：静态初始化和动态初始化。

1.静态初始化：
静态初始化是在定义互斥量时直接进行的，不需要在运行时进行额外的初始化操作。静态初始化使用PTHREAD_MUTEX_INITIALIZER 来初始化互斥量。

#include <pthread.h>

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

静态初始化的优点是简单方便，无需额外的调用和错误处理。但是，静态初始化的互斥量的属性是默认的，并且无法在运行时动态修改。

2.动态初始化：

动态初始化是在运行时通过函数调用来初始化互斥量，可以在初始化时设置互斥量的属性。动态初始化需要使用 pthread_mutex_init() 函数来初始化互斥量，并在不再使用时使用 pthread_mutex_destroy() 函数来销毁互斥量。

#include <pthread.h>

int pthread_mutex_init(pthread_mutex_t *mutex, const pthread_mutexattr_t *attr);
 
    mutex 参数是指向互斥量对象的指针，attr 参数是指向 pthread_mutexattr_t 类型的互斥量属性对象的指针。如果将 attr 参数设置为 NULL，则使用默认的属性进行初始化。

函数返回值为 0 表示成功，非零值表示失败。

3.3.2互斥量销毁

int pthread_mutex_destroy(pthread_mutex_t *mutex)；
    
使用 PTHREAD_ MUTEX_ INITIALIZER 初始化的互斥量不需要销毁
不要销毁一个已经加锁的互斥量
已经销毁的互斥量，要确保后面不会有线程再尝试加锁

3.3.3互斥量加锁与解锁

int pthread_mutex_lock(pthread_mutex_t *mutex);
int pthread_mutex_unlock(pthread_mutex_t *mutex);

返回值:成功返回0,失败返回错误号

3.3.4演示例子

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <pthread.h>
#include <sched.h>
int tickets = 10000; // 在并发访问的时候，导致了我们数据不一致的问题！临界资源

#define THREAD_NUM 800

class ThreadData
{
public:
    ThreadData(const std::string &n,pthread_mutex_t *pm):tname(n), pmtx(pm)
    {}
public:
    std::string tname;
    pthread_mutex_t *pmtx;
};

void *getTickets(void *args)
{
    // int myerrno = errno;
    ThreadData *td = (ThreadData*)args;
    while(true)
    {
        // 抢票逻辑
        int n = pthread_mutex_lock(td->pmtx);
        assert(n == 0);
        // 临界区
        if(tickets > 0) // 1. 判断的本质也是计算的一种
        {
            usleep(rand()%1500);
            printf("%s: %d\n", td->tname.c_str(), tickets);
            tickets--; // 2. 也可能出现问题
            n = pthread_mutex_unlock(td->pmtx);
            assert(n == 0);
        }
        else{
            n = pthread_mutex_unlock(td->pmtx);
            assert(n == 0);
            break;
        }
        
        // 抢完票，其实还需要后续的动作
        usleep(rand()%2000);
        //  errno = myerrno;
    }
    delete td;
    return nullptr;
}

int main()
{
    time_t start = time(nullptr);
    pthread_mutex_t mtx;
    pthread_mutex_init(&mtx, nullptr);

    srand((unsigned long)time(nullptr) ^ getpid() ^ 0x147);
    pthread_t t[THREAD_NUM];
    // 多线程抢票的逻辑
    for(int i = 0; i < THREAD_NUM; i++)
    {
        std::string name = "thread ";
        name += std::to_string(i+1);
        ThreadData *td = new ThreadData(name, &mtx);
        pthread_create(t + i, nullptr, getTickets, (void*)td);
    }

    for(int i = 0; i < THREAD_NUM; i++)
    {
        pthread_join(t[i], nullptr);
    }

    pthread_mutex_destroy(&mtx);

    time_t end = time(nullptr);

    cout << "cast: " << (int)(end - start) << "S" << endl;
}

3.4互斥量的实现原理

伪代码及演示图：
在这里插入图片描述

在这里插入图片描述

按照上面lock的伪代码进行讲解：假设有一个进程A，此时将cpu中一个名叫al的寄存器中存入数据0，再将al寄存器中的这个0与内存当中mutex存储的这个1的数据进行互换。这时进程B突然切换，进程A就带着它的上下文（就是此时寄存器中存储的这个1）溜之大吉。B按照这一套操作下来发现卡在了if等待这一步，就只能无奈挂起等待A进程执行完。

根据上述的描述其实加锁的本质就是交换寄存器与内存数据这一步（该操作是原子的），解锁看伪代码就明白这里就不过多解释了。

3.5死锁

死锁是指在一组进程中的各个进程均占有不会释放的资源，但因互相申请被其他进程所站用不会释放的资

源而处于的一种永久等待状态。

###死锁四个必要条件

互斥条件：一个资源每次只能被一个执行流使用
请求与保持条件：一个执行流因请求资源而阻塞时，对已获得的资源保持不放
不剥夺条件:一个执行流已获得的资源，在末使用完之前，不能强行剥夺
循环等待条件:若干执行流之间形成一种头尾相接的循环等待资源的关系

###避免死锁

破坏死锁的四个必要条件
加锁顺序一致
避免锁未释放的场景
资源一次性分配

###避免死锁算法

死锁检测算法(了解)
银行家算法（了解）

在这里插入图片描述

上述将了这么多概念，我们就上图来简单聊聊死锁的概念，比如线程A和线程B在申请完锁1和锁2之后分别在申请锁2和锁1，但是双方都不给，这样就会形成一个循环导致最后什么都没干成，这就是死锁的一个抽象例子

4.linux线程同步

4.1基本概念

当一个线程访问临界资源但是发现这个临界资源已经在被访问，这个进程只能无奈等待什么都做不了，这种情况就叫做饥饿问题。在保证数据安全的前提下，让线程能够按照某种特定的顺序访问临界资源，从而有效避免饥饿问

题，叫做同步。为了人为构造这种同步关系，我们需要用到条件变量函数，下面我们将对其进行一一介绍。

4.2条件变量函数

初始化条件变量

int pthread_cond_init(pthread_cond_t *restrict cond,const pthread_condattr_t *restrict
attr);

参数：
cond：指向要初始化的条件变量的指针。条件变量的类型是pthread_cond_t，通常定义为全局变量或动态分配的内存。
attr：指向pthread_condattr_t类型的指针，表示条件变量的属性。通常情况下，可以将其设置为NULL，使用默认属性。

等待条件变量的满足

int pthread_cond_wait(pthread_cond_t *restrict cond, pthread_mutex_t *restrict mutex);
pthread_cond_wait函数接受两个参数：

cond：指向要等待的条件变量的指针，类型为pthread_cond_t
mutex：指向与条件变量关联的互斥锁的指针，类型为pthread_mutex_t

pthread_cond_wait函数的执行过程如下：

1.线程调用pthread_cond_wait函数，进入等待状态。
2.在进入等待状态之前，线程必须先获取互斥锁，即调用pthread_mutex_lock函数。获取互斥锁是为了保证线程在等待条件变量期间的原子性操作。
3.一旦线程获取了互斥锁，它会释放该互斥锁，并进入等待状态。此时，其他线程可以获取互斥锁并进入临界区执行操作。
4.在等待状态中，线程会等待条件变量的满足。条件变量的满足通常由其他线程发出信号或广播来实现。
5.当条件变量被满足时，线程会被唤醒并重新获取互斥锁。
6.一旦线程重新获取了互斥锁，它将退出pthread_cond_wait函数，并继续执行后续的代码。

唤醒等待

int pthread_cond_broadcast(pthread_cond_t *cond);
```

``pthread_cond_broadcast`函数用于向等待特定条件变量的所有线程发送信号。它会同时唤醒所有等待该条件变量的线程，使它们从等待状态返回，并尝试重新获取相关的互斥锁。这样，所有等待线程都有机会继续执行。

- 当调用`pthread_cond_broadcast`函数时，所有等待该条件变量的线程都会被唤醒，不管它们当前处于什么状态。
- 具体哪个线程会获得互斥锁并继续执行取决于调度策略和操作系统。
- 如果没有线程在等待条件变量，调用`pthread_cond_broadcast`函数也不会产生任何影响。

int pthread_cond_signal(pthread_cond_t *cond);
```

``pthread_cond_signal`函数用于向等待特定条件变量的一个线程发送信号。它会唤醒等待该条件变量的某个线程（通常是最先等待的线程），使其从等待状态返回，并尝试重新获取相关的互斥锁。

- 当调用`pthread_cond_signal`函数时，只有一个等待该条件变量的线程会被唤醒。
- 具体哪个线程会被唤醒取决于调度策略和操作系统。
- 如果没有线程在等待条件变量，调用`pthread_cond_signal`函数也不会产生任何影响。

销毁条件变量对象，并释放相关的资源。

int pthread_cond_destroy(pthread_cond_t *cond);
cond：指向要销毁的条件变量的指针，类型为pthread_cond_t
    
    pthread_cond_destroy函数的执行过程如下：

1.在调用pthread_cond_destroy函数之前，确保条件变量不再使用，并且没有线程在等待或被唤醒。

2.调用pthread_cond_destroy函数将销毁条件变量对象，并释放与之相关的资源。

3.销毁条件变量后，它将不再可用，不应再对其进行任何操作。

简单实例的应用

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <pthread.h>
pthread_cond_t cond;
pthread_mutex_t mutex;
void* r1(void* arg)
{
	while (1) {
		pthread_cond_wait(&cond, &mutex); 
		printf("活动\n");
	}
}
void* r2(void* arg)
{
	while (1) {
		pthread_cond_signal(&cond);
		sleep(1);
	}
}
int main(void)
{
	pthread_t t1, t2;
	pthread_cond_init(&cond, NULL);
	pthread_mutex_init(&mutex, NULL);
	pthread_create(&t1, NULL, r1, NULL); 
	pthread_create(&t2, NULL, r2, NULL);
	pthread_join(t1, NULL);
	pthread_join(t2, NULL);
	pthread_mutex_destroy(&mutex);
	pthread_cond_destroy(&cond);
}
[root@localhost linux]# ./a.out
    活动
    活动
    活动

4.生产者消费者模型

4.1基本原理

生产者消费者模式就是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。这个阻塞队列就是用来给生产者和消费者解耦的。

在这里插入图片描述

4.2基于BlockingQueue的生产者消费者模型

在这里插入图片描述

这里偷一张别人的图来解释一下这个模型，其基本思想就是模拟一个队列然后生产者往队列里塞入元素，消费者消费元素，当元素耗尽停止消费，当队列塞满停止生产

4.3POSIX信号量

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

初始化信号量

int sem_init(sem_t *sem, int pshared, unsigned int value);
sem：指向要初始化的信号量的指针，类型为sem_t。
pshared：指定信号量的共享类型。如果为0，表示信号量只能由当前进程的线程使用；如果为非零值，表示信号量可以在多个进程之间共享。在多线程编程中，通常将该参数设置为0。
value：指定信号量的初始值。该值确定了在无阻塞情况下可以执行的线程或进程数量。

销毁信号量

int sem_destroy(sem_t *sem);
sem：指向要销毁的信号量的指针，类型为sem_t。

等待信号量

int sem_wait(sem_t *sem);
sem：指向要等待的信号量的指针，类型为sem_t。
    
1.如果信号量的值大于0，表示资源可用，sem_wait函数将减少信号量的值并立即返回。

2.如果信号量的值为0，表示资源不可用，sem_wait函数将阻塞当前线程或进程，直到信号量的值大于0或等待超时。

3.如果等待超时，sem_wait函数将返回一个非零值，表示等待超时。

4.如果成功获取了信号量的资源，sem_wait函数将返回0。

发布信号量

int sem_post(sem_t *sem);
sem：指向要发布的信号量的指针，类型为sem_t。

1.sem_post函数将增加信号量的值。

2.如果有其他线程或进程正在等待该信号量的资源，它们将被唤醒，并有机会获取资源。

3.如果没有其他线程或进程在等待该信号量的资源，操作仅仅是增加信号量的值，并不会有其他影响。

4.4基于环形队列的生产消费模型

在这里插入图片描述

其实实际上就是把刚才队列的数据结构换成了一个用数组模拟的环形队列，设计合理即可，下面有个演示案例。

#include <iostream>
#include <vector>
#include <stdlib.h>
#include <semaphore.h>
#include <pthread.h>
#define NUM 16
class RingQueue {
private:
	std::vector<int> q;
	int cap;
	sem_t data_sem;
	sem_t space_sem;
	int consume_step;
	int product_step;
public:
	RingQueue(int _cap = NUM) :q(_cap), cap(_cap)
	{
		sem_init(&data_sem, 0, 0);
		sem_init(&space_sem, 0, cap);
		consume_step = 0;
		product_step = 0;
	}
	void PutData(const int& data)
	{
		sem_wait(&space_sem); // P
		q[consume_step] = data;
		consume_step++;
		consume_step %= cap;
		sem_post(&data_sem); //V
	}
	void GetData(int& data)
	{
		sem_wait(&data_sem);
		data = q[product_step];
		product_step++;
		product_step %= cap;
		sem_post(&space_sem);
	}
	~RingQueue()
	{
		sem_destroy(&data_sem);
		sem_destroy(&space_sem);
	}
};
void* consumer(void* arg)
{
	RingQueue* rqp = (RingQueue*)arg;
	int data;
	for (; ; ) {
		rqp->GetData(data);
		std::cout << "Consume data done : " << data << std::endl;
		sleep(1);
	}
}
//more faster
void* producter(void* arg)
{
	RingQueue* rqp = (RingQueue*)arg;
		srand((unsigned long)time(NULL));
	for (; ; ) {
		int data = rand() % 1024;
		rqp->PutData(data);
		std::cout << "Prodoct data done: " << data << std::endl;
		// sleep(1);
	}
}
int main()
{
	RingQueue rq;
	pthread_t c, p;
	pthread_create(&c, NULL, consumer, (void*)&rq);
	pthread_create(&p, NULL, producter, (void*)&rq);
	pthread_join(c, NULL);
	pthread_join(p, NULL);
}

5.线程池

5.1基本概念

线程池是一种用于管理和复用线程的机制，它可以提高线程的利用率和效率。在使用线程池的情况下，可以避免频繁地创建和销毁线程，从而减少线程创建和上下文切换的开销。

在使用线程池时需要根据具体的应用场景和需求进行评估和权衡，合理地配置线程池的大小、监控线程池的状态、正确处理任务和资源管理等问题，可以最大程度地减轻其有的缺点，并提高线程池的性能和稳定性。

5.2经典线程池代码案例

threadPool.hpp

#pragma once

#include <iostream>
#include <vector>
#include <string>
#include <queue>
#include <unistd.h>
#include "thread.hpp"
#include "lockGuard.hpp"
#include "log.hpp"

const int g_thread_num = 3;
// 本质是: 生产消费模型
template <class T>
class ThreadPool
{
public:
    pthread_mutex_t *getMutex()
    {
        return &lock;
    }
    bool isEmpty()
    {
        return task_queue_.empty();
    }
    void waitCond()
    {
        pthread_cond_wait(&cond, &lock);
    }
    T getTask()
    {
        T t = task_queue_.front();
        task_queue_.pop();
        return t;
    }

private:
    ThreadPool(int thread_num = g_thread_num) : num_(thread_num)
    {
        pthread_mutex_init(&lock, nullptr);
        pthread_cond_init(&cond, nullptr);
        for (int i = 1; i <= num_; i++)
        {
            threads_.push_back(new Thread(i, routine, this));
        }
    }
    ThreadPool(const ThreadPool<T> &other) = delete;
    const ThreadPool<T> &operator=(const ThreadPool<T> &other) = delete;

public:
    // 考虑一下多线程使用单例的过程
    static ThreadPool<T> *getThreadPool(int num = g_thread_num)
    {
        // 可以有效减少未来必定要进行加锁检测的问题
        // 拦截大量的在已经创建好单例的时候，剩余线程请求单例的而直接访问锁的行为
        if (nullptr == thread_ptr) 
        {
            lockGuard lockguard(&mutex);
            // 但是，未来任何一个线程想获取单例，都必须调用getThreadPool接口
            // 但是，一定会存在大量的申请和释放锁的行为，这个是无用且浪费资源的
            // pthread_mutex_lock(&mutex);
            if (nullptr == thread_ptr)
            {
                thread_ptr = new ThreadPool<T>(num);
            }
            // pthread_mutex_unlock(&mutex);
        }
        return thread_ptr;
    }
    // 1. run()
    void run()
    {
        for (auto &iter : threads_)
        {
            iter->start();
            // std::cout << iter->name() << " 启动成功" << std::endl;
            logMessage(NORMAL, "%s %s", iter->name().c_str(), "启动成功");
        }
    }
    // 线程池本质也是一个生产消费模型
    // void *routine(void *args)
    // 消费过程
    static void *routine(void *args)
    {
        ThreadData *td = (ThreadData *)args;
        ThreadPool<T> *tp = (ThreadPool<T> *)td->args_;
        while (true)
        {
            T task;
            {
                lockGuard lockguard(tp->getMutex());
                while (tp->isEmpty())
                    tp->waitCond();
                // 读取任务
                task = tp->getTask(); // 任务队列是共享的-> 将任务从共享，拿到自己的私有空间
            }
            task(td->name_);
            // lock
            // while(task_queue_.empty()) wait();
            // 获取任务
            // unlock

            // 处理任务
        }
    }
    // 2. pushTask()
    void pushTask(const T &task)
    {
        lockGuard lockguard(&lock);
        task_queue_.push(task);
        pthread_cond_signal(&cond);
    }
    // test func
    // void joins()
    // {
    //     for (auto &iter : threads_)
    //     {
    //         iter->join();
    //     }
    // }
    ~ThreadPool()
    {
        for (auto &iter : threads_)
        {
            iter->join();
            delete iter;
        }
        pthread_mutex_destroy(&lock);
        pthread_cond_destroy(&cond);
    }

private:
    std::vector<Thread *> threads_;
    int num_;
    std::queue<T> task_queue_;

    static ThreadPool<T> *thread_ptr;
    static pthread_mutex_t mutex;

    // 方案2:
    //  queue1,queue2
    //  std::queue<T> *p_queue, *c_queue
    //  p_queue->queue1
    //  c_queue->queue2
    //  p_queue -> 生产一批任务之后，swap(p_queue,c_queue),唤醒所有线程/一个线程
    //  当消费者处理完毕的时候，你也可以进行swap(p_queue,c_queue)
    //  因为我们生产和消费用的是不同的队列，未来我们要进行资源的处理的时候，仅仅是指针

    pthread_mutex_t lock;
    pthread_cond_t cond;
};
template <typename T>
ThreadPool<T> *ThreadPool<T>::thread_ptr = nullptr;

template <typename T>
pthread_mutex_t ThreadPool<T>::mutex = PTHREAD_MUTEX_INITIALIZER;

thread.hpp


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

// typedef std::function<void* (void*)> fun_t;
typedef void *(*fun_t)(void *);

class ThreadData
{
public:
    void *args_;
    std::string name_;
};

class Thread
{
public:
    Thread(int num, fun_t callback, void *args) : func_(callback)
    {
        char nameBuffer[64];
        snprintf(nameBuffer, sizeof nameBuffer, "Thread-%d", num);
        name_ = nameBuffer;

        tdata_.args_ = args;
        tdata_.name_ = name_;
    }
    void start()
    {
        pthread_create(&tid_, nullptr, func_, (void*)&tdata_);
    }
    void join()
    {
        pthread_join(tid_, nullptr);
    }
    std::string name()
    {
        return name_;
    }
    ~Thread()
    {
    }

private:
    std::string name_;
    fun_t func_;
    ThreadData tdata_;
    pthread_t tid_;
};

lockGuard.hpp
    
#pragma once

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

class Mutex
{
public:
    Mutex(pthread_mutex_t *mtx):pmtx_(mtx)
    {}
    void lock() 
    {
        // std::cout << "要进行加锁" << std::endl;
        pthread_mutex_lock(pmtx_);
    }
    void unlock()
    {
        // std::cout << "要进行解锁" << std::endl;
        pthread_mutex_unlock(pmtx_);
    }
    ~Mutex()
    {}
private:
    pthread_mutex_t *pmtx_;
};

// RAII风格的加锁方式
class lockGuard
{
public:
    lockGuard(pthread_mutex_t *mtx):mtx_(mtx)
    {
        mtx_.lock();
    }
    ~lockGuard()
    {
        mtx_.unlock();
    }
private:
    Mutex mtx_;
};

log.hpp
    
#pragma once

#include <iostream>
#include <cstdio>
#include <cstdarg>
#include <ctime>
#include <string>

// 日志是有日志级别的
#define DEBUG   0
#define NORMAL  1
#define WARNING 2
#define ERROR   3
#define FATAL   4

const char *gLevelMap[] = {
    "DEBUG",
    "NORMAL",
    "WARNING",
    "ERROR",
    "FATAL"
};

#define LOGFILE "./threadpool.log"

// 完整的日志功能，至少: 日志等级 时间 支持用户自定义(日志内容, 文件行，文件名)
void logMessage(int level, const char *format, ...)
{
#ifndef DEBUG_SHOW
    if(level== DEBUG) return;
#endif
    // va_list ap;
    // va_start(ap, format);
    // while()
    // int x = va_arg(ap, int);
    // va_end(ap); //ap=nullptr
    char stdBuffer[1024]; //标准部分
    time_t timestamp = time(nullptr);
    // struct tm *localtime = localtime(&timestamp);
    snprintf(stdBuffer, sizeof stdBuffer, "[%s] [%ld] ", gLevelMap[level], timestamp);

    char logBuffer[1024]; //自定义部分
    va_list args;
    va_start(args, format);
    // vprintf(format, args);
    vsnprintf(logBuffer, sizeof logBuffer, format, args);
    va_end(args);

    FILE *fp = fopen(LOGFILE, "a");
    // printf("%s%s\n", stdBuffer, logBuffer);
    fprintf(fp, "%s%s\n", stdBuffer, logBuffer);
    fclose(fp);
}

Task.hpp
    
#pragma once

#include <iostream>
#include <string>
#include <functional>
#include "log.hpp"

typedef std::function<int(int, int)> func_t;

class Task
{
public:
    Task(){}
    Task(int x, int y, func_t func):x_(x), y_(y), func_(func)
    {}
    void operator ()(const std::string &name)
    {
        // std::cout << "线程 " << name << " 处理完成, 结果是: " << x_ << "+" << y_ << "=" << func_(x_, y_) << std::endl;
        logMessage(WARNING, "%s处理完成: %d+%d=%d | %s | %d",
            name.c_str(), x_, y_, func_(x_, y_), __FILE__, __LINE__);
    }
public:
    int x_;
    int y_;
    // int type;
    func_t func_;
};

testMain.cc

#include "threadPool.hpp"
#include "Task.hpp"
#include <ctime>
#include <cstdlib>
#include <iostream>
#include <unistd.h>

// void *run(void *args)
// {
//     while(true)
//     {
//         ThreadPool<Task>::getThreadPool();
//     }
// }

int main()
{
    // logMessage(NORMAL, "%s %d %c %f \n", "这是一条日志信息", 1234, 'c', 3.14);

    srand((unsigned long)time(nullptr) ^ getpid());
    // ThreadPool<Task> *tp = new ThreadPool<Task>();
    // ThreadPool<Task> *tp = ThreadPool<Task>::getThreadPool();
    // 那么，如果单例本身也在被多线程申请使用呢？？
    ThreadPool<Task>::getThreadPool()->run();
    //thread1,2,3,4

    while(true)
    {
        //生产的过程，制作任务的时候，要花时间
        int x = rand()%100 + 1;
        usleep(7721);
        int y = rand()%30 + 1;
        Task t(x, y, [](int x, int y)->int{
            return x + y;
        });

        // std::cout << "制作任务完成: " << x << "+" << y << "=?" << std::endl;
        logMessage(DEBUG, "制作任务完成: %d+%d=?", x, y);
        logMessage(DEBUG, "制作任务完成: %d+%d=?", x, y);
        logMessage(DEBUG, "制作任务完成: %d+%d=?", x, y);
        logMessage(DEBUG, "制作任务完成: %d+%d=?", x, y);

        // 推送任务到线程池中
        ThreadPool<Task>::getThreadPool()->pushTask(t);

        sleep(1);
    }
    return 0;
}