【Linux系统】多线程

本篇博客继上一篇《线程与线程控制》，又整理了多线程相关的线程安全问题、互斥与锁、同步与条件变量、生产消费模型、线程池等内容，旨在让读者更加深刻地理解线程和初步掌握多线程编程。（欲知线程的相关概念、线程控制的相关接口等，请见：【Linux系统】线程与线程控制-CSDN博客）

一、线程互斥

1.线程安全

.1）引例：抢票

.2）引例详解

.3）临界、原子性、互斥量

2.互斥量的相关接口

.1）初始化与销毁

.2）加锁与解锁

.3）在上文引例中实现线程互斥

3.互斥量的原理

4.死锁

补.特殊锁

.1）自旋锁

.2）读写锁

二、线程同步

1.竞态条件、条件变量

2.条件变量的相关接口

.1）初始化与销毁

.2）等待与唤醒

.3）在上文引例中实现线程同步

三、生产消费模型

1.三种关系、两种角色、一种容器

2.基于阻塞队列的生产消费模型

.1）单生产单消费模型

.2）基于任务的多生产多消费模型

.补）RAII 风格的互斥锁

3.POSIX 信号量

.1）基本原理

.2）相关接口

.3）在上文引例中引入二元信号量

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

.1）单生产单消费模型

.2）多生产多消费模型

四、线程池

1.池化技术

2.线程池的实现

五、线程安全的单例模式

1.饿汉和懒汉

2.基于懒汉模式的单例线程池

一、线程互斥

1.线程安全

由于一个进程地址空间是可以被多个线程共享的，因此位于代码区的全局变量能被多个线程同时访问和修改。

如果一个进程中存在多个线程，且调度器会频繁地发生线程调度与切换，使多个线程交叉执行 —— 一个线程还没有执行完就轮到下一个执行了，每个线程都执行一点，一个线程在时间片到期、等待更高优先级线程到来的时候，会发生线程切换 —— 此时，又涉及了多个线程访问同一个全局变量，就会引发线程安全问题。

为了更方便地理解线程安全问题，下面引入一个生活案例和相关代码。

.1）引例：抢票

在春运的时候，抢火车票是十分常见的事，除了自己在线上平台、线下售票处抢票外，还可以拜托票贩子替自己抢票。

此处引入以下代码，模拟票贩子抢票的过程，并假设现在有5个票贩子分抢100张票。

mythread.cc：

#include <iostream>
#include <pthread.h>
#include <vector>
#include <cstdio>
#include <unistd.h>


using namespace std;

#define NUM 5       //假设有5人（5个线程）抢票

int tickets = 100;   //假设有100张票

//线程的基本信息
class ThreadInfo
{
public:
    ThreadInfo(const string &threadname)
    :threadname_(threadname)
    {}

public:
    string threadname_;//线程的名字
};
//线程的例程，负责抢票
void *GrabTickets(void *args)
{
    ThreadInfo *ti = static_cast<ThreadInfo*>(args);
    string name(ti->threadname_);
    while(1)
    {
        if(tickets > 0)
        {   //在每次抢票前，休眠1000毫秒
            usleep(1000);
            //抢票时，显示正在抢票的线程名和票的剩余数量
            printf("%s get a ticket: %d\n", name.c_str(), tickets);
            tickets--;//每抢到一次，票数应减少1
        }
        else break;
    }
    //票抢完时，提示一个线程已退出
    printf("%s quit...\n", name.c_str());
    return nullptr;
}

int main()
{
    //用vector管理抢票的线程和线程的信息
    vector<pthread_t> tids;
    vector<ThreadInfo*> tis;
    //创建抢票的线程
    for(int i = 1; i <= NUM; i++)
    {
        pthread_t tid;
        ThreadInfo *ti = new ThreadInfo("Thread-"+to_string(i));
        pthread_create(&tid, nullptr, GrabTickets, ti);
        tids.push_back(tid);
        tis.push_back(ti);
    }

    // 主线程等待回收所有线程
    for(auto tid : tids)
    {
        pthread_join(tid, nullptr);
    }

    // 释放new的资源
    for(auto ti : tis)
    {
        delete ti;
    }
    // 抢票程序结束
    return 0;
}

Makefile：

mythread:mythread.cc
	g++ -o $@ $^ -std=c++11 -lpthread
.PHONY:clean
clean:
	rm -f mythread

由演示图，在抢票进程中的 5 个线程分别执行抢票的动作，但它们不仅屡次抢到了同一张票，有的甚至还抢到了不存在的票（票的剩余数量为负数）。

票应该是每一张仅由一人持有，每个线程不应该抢到同一张票，或不存在的票。

这就是多个线程交叉执行，访问同一个全局变量所引发的线程安全问题。

.2）引例详解

线程抢到了不存在的票（票的剩余数量为负数），跟线程例程中的 “if ( tickets > 0 ) ” 、“usleep(1000);”、 “tickets --;”有很大关系，因为当票的剩余数量 0 时，例程中的“tickets --;”经过 “if ( tickets > 0 ) ”的判断，按理来说是不会执行的。但为什么 “tickets --;” 还是执行了，最终导致票的剩余数量从 0 减为了负数，难道是 if 语句的判断失误了？

下面就对 “if ( tickets > 0 ) ” 、“usleep(1000);”、 “tickets --;” 这三句代码的执行，进行更加详细的解释。

在上文引例的代码中，主线程创建好 5 个子线程以后，它们便开始执行各自的例程。

if ( tickets > 0 )

假设现在票已经只剩一张了，即全局变量 tickets = 1。

在线程 Thread-1 执行到 if 判断时，CPU会从内存中将 tickets 变量的数据加载到了 CPU 的寄存器 ebx 中。此时 tickets = 1，符合大于 0 的条件，线程 Thread-1 会继续执行后续代码。

usleep(1000);

if 语句后，紧接着的就是 “usleep(1000);”。

线程 Thread-1 在执行到延时 “usleep(1000);” 一句的时候，就会被放进等待队列里，切换下一个线程 Thread-2 去执行它的例程。

线程 Thread-1 被切走时，它的上下文数据也会被切走，因此，ebx寄存器中的数据也被保存在线程 Thread-1 的 PCB 中，且随着线程 Thread-1 的切走也一起被切走。

但线程 Thread-2 仍会重复线程 Thread-2 的过程，在执行到 “usleep(1000);”被切走，再换上线程 Thread-3 ......以此类推，直到线程 Thread-5 被切走，终于又轮到线程 Thread-1 执行了（因为它在等待队列的队头）。

而此时，5 个线程都保存了变量 tickets = 1 的数据，都符合 if 的判断条件，都能执行后续代码。

tickets --;

线程 Thread-1 保存了变量 tickets = 1 的数据和切走前的上下文，会接着它被切走的位置，继续执行 if 之后的代码，先执行完打印，就执行到 “tickets --;” 了。

-- 操作在汇编中会被转换成至少 3 条语句，因此，-- 操作的完成也基本分为三步：

（1）将内存的数据加载到寄存器ebx；

（2）对数据进行减 1 的修改操作；

（3）将修改后的数据写回内存。

在先前，线程 Thread-1 已经完成了将 tickets 的数据从内存加载到寄存器 ebx 的工作，接下来会在寄存器 ebx 中执行完对 tickets 的 -- 操作，并将 -- 后的数据写回内存中 tickets 的地址。完成这一系列工作后，线程 Thread-1 再循环执行到 if 语句，认为 tickets 已经是 0 了，于是跳出 while 循环并退出了。

和线程 Thread-1 一样，线程 Thread-2 也会接着它被切走的位置继续执行，对 tickets 完成 -- 操作，并将 -- 后的数据写回内存中 tickets 的地址。完成这一系列工作后，线程 Thread-2 再循环执行到 if 语句，认为 tickets 已经是 -1 了，于是跳出 while 循环并退出了。

以此类推，直到线程 Thread-5 完成-- 操作并退出。

但此时，数据的管理已经完全乱套了，票的剩余数量也由此变成了负数。

这种现象就叫线程安全问题，更具体地说，又叫数据不一致问题。导致数据不一致问题的原因一般是，共享资源没有被保护，多个线程对共享资源进行了交叉访问；而解决数据不一致问题的办法，就是对共享资源加锁。

.3）临界、原子性、互斥量

要懂得如何对共享资源加锁，首先要理解与线程互斥有关的一些概念。

【Tips】线程互斥的相关概念

临界资源：多个执行流进行安全访问的共享资源。上文中，全局变量 tickets 存在多线程交叉访问所导致的数据不一致问题，显然就不是一个临界资源。
临界区：多个执行流中，访问临界资源的代码。上文中，如果 tickets 是一个临界资源，在每个线程的例程中，涉及 tickets 的访问和修改的代码就属于临界区（但 tickets 不是一个临界资源，因此上文中线程例程的代码不存在临界区）。
原子性：简单来说就是，访问一个资源的时候，没有中间态，要么完成访问，要么就不访问。C/C++的 -- 和 ++ 操作，在访问资源时存在中间态，因此不是原子性的。
线程互斥：多个线程访问共享资源不再是交叉访问，而是串行访问，即任何时候有且仅有一个执行流在访问共享资源。上文中，如果能让 tickets 从一个共享资源变成一个临界资源，就能使多个线程串行访问 tickets，从而避免数据不一致问题，此时则称实现了线程互斥。

【Tips】互斥量 mutex

大多时候，线程使用的数据都是局部变量，局部变量的地址存在于线程独立的栈空间中，使其他线程无法获得这个局部变量。
不过，有时还会存在一些变量在线程之间共享，以完成线程之间的交互。但多线程是并发执行的，多个线程并发地访问共享变量，可能会导致数据不一致问题。
为了解决数据不一致问题，就需要对共享资源加锁，实现线程互斥；而要对共享资源加锁，就基本要做到以下三点：

代码必须有互斥行为，即一个线程进入一个临界区执行时，不允许其他线程进入这个临界区。
如果多个线程同时需要进入临界区中执行代码，且当下没有线程在这个临界区中，那么仅允许一个线程进入这个临界区。
如果一个线程不在一个临界区中执行，那么这个线程就不能阻止其他线程进入这个临界区。

在 Linux 中，提供了一把互斥锁，叫互斥量 mutex。

2.互斥量的相关接口

每一个线程在进入临界区之前都必须先申请互斥量（锁），只有申请到互斥量的线程才可以进入临界区并对临界资源进行访问；当线程离开临界区的时，需要释放锁，这样才能让其余要进入临界区的线程继续竞争锁。

【ps】使用锁的注意事项

大多情况下，加锁对性能都有一定的不可避免的损耗，因为加锁使多执行流由并行执行变成串行执行了。
在合适的位置进行加锁和解锁，可以尽可能减少加锁带来的性能损耗。
进行临界资源的保护，是所有执行流都应该遵守的准则，也是程序员在编码时需要注意的。

.1）初始化与销毁

初始化一个互斥量

#include <pthread.h>
int pthread_mutex_init(pthread_mutex_t *restrict mutex,
       const pthread_mutexattr_t *restrict attr);
功能：以动态分配的方式初始化一个锁
参数：1.mutex：锁的指针，指向待初始化的锁。
     2.attr：锁的属性，不关心则置为NULL即可。
返回值：初始化成功则返回0，失败则返回错误码。
ps：以静态分配的方式初始化一个锁：
    pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

销毁一个互斥量

#include <pthread.h>
int pthread_mutex_destroy(pthread_mutex_t *mutex);
功能：销毁一个动态分配的互斥锁
参数：锁的指针，指向待销毁的锁。
返回值：成功返回0，失败返回错误码。
ps：1.以静态分配的方式来初始化的锁无需调用 pthread_mutex_destroy() 销毁。
    2.不要尝试销毁一个已加锁的锁。
    3.不要对已销毁的锁尝试加锁。

.2）加锁与解锁

对一个互斥量进行加锁

#include <pthread.h>
int pthread_mutex_lock(pthread_mutex_t *mutex);
功能：对一个未加锁的锁进行加锁
参数：锁的指针，指向待加锁的锁。
返回值：加锁成功则返回0，失败则返回错误码。
ps：1.若互斥量处于未锁状态，pthread_mutex_lock() 会将互斥量锁定，同时返回0。
    2.pthread_mutex_lock() 被一个线程调用时，存在其他线程已经锁定互斥量，或其他线
      程同时也在申请互斥量，但没有竞争到互斥量，那么pthread_mutex_lock() 会陷入阻
      塞（执行流被挂起），直到互斥量解锁。

【补】加锁操作的更多说明

        不同线程对锁的竞争能力可能会不同。一个线程刚把锁释放，紧接着就立即去申请锁，那么该线程申请到锁的几率是比其它进程要大的，因为其它线程正处于被挂起的状态，要等待锁被释放。在锁被释放的时候，操作系统要先唤醒这些被挂起的进程，然后才去申请锁，这个过程与先前一直在运行的线程相比一定是更慢的。在纯互斥环境中，如果锁分配不够合理，容易导致其他线程的饥饿问题（一个线程长时间申请不到互斥量），因此，需要让刚释放锁的线程不能再立即申请到锁，必须让它排在等待队列的最后。

        可能同时存在多个线程在等待一把锁资源。当一个锁被释放的时候，操作系统如果把所有等待的线程全部唤醒，这也是不合理的，因为最终只会有一个线程拿到锁资源。于是，系统会让所有的线程按照一定的顺序去获取锁，这种按照一定的顺序性获取资源的过程就叫同步。

        所有线程在执行临界区代码访问临界资源之前，都需要先申请锁，因此，锁其实是一种共享资源，这也决定了锁的申请和释放一定要被设计成原子性的。例如，一个线程在执行临界区的代码时，是可以被切换的，在被切出去的时候，是以持有锁的状态被切出去的。因此，在该线程释放锁资源之前，其它线程无法进入临界区访问临界资源。

对一个互斥量进行解锁

#include <pthread.h>
int pthread_mutex_unlock(pthread_mutex_t *mutex);
功能：对一个已加锁的锁进行解锁
参数：锁的指针，指向待解锁的锁。
返回值：解锁成功则返回0，失败则返回错误码。

.3）在上文引例中实现线程互斥

mythread.cc：

#include <iostream>
#include <pthread.h>
#include <vector>
#include <cstdio>
#include <unistd.h>
using namespace std;

#define NUM 5       //假设有5人（5个线程）抢票

int tickets = 100; // 定义1000张票

//线程的基本信息
class ThreadInfo
{
public:
    ThreadInfo(const string &threadname, pthread_mutex_t *lock)
    :threadname_(threadname)
    ,lock_(lock)
    {}

public:
    string threadname_;    //线程的名字
    pthread_mutex_t *lock_;//线程申请的锁
};

//抢票
void *GrabTickets(void *args)
{
    ThreadInfo *ti = static_cast<ThreadInfo*>(args);
    string name(ti->threadname_);
    while(true)
    {   
        // 加锁
        pthread_mutex_lock(ti->lock_); 
        if(tickets > 0)
        {
            usleep(10000);
            printf("%s get a ticket: %d\n", name.c_str(), tickets);
            tickets--;
            // 每抢一次票，解锁一次
            pthread_mutex_unlock(ti->lock_); 
        }
        else 
        {
            //抢完票后解锁
            pthread_mutex_unlock(ti->lock_); 
            break;
        }
        // 用休眠来模拟抢到票的后续动作
        usleep(13);

        // pthread_mutex_unlock(ti->lock_); 
        // 不能在这里解锁，若 tickets == 0 时跳出循环，导致锁未释放，
        // 其它线程就会阻塞住，进而导致程序卡死
    }

    printf("%s quit...\n", name.c_str());

    return NULL;
}

int main()
{
    //定义一个互斥量，并以动态分配的方式对其进行初始化
    pthread_mutex_t lock;               
    pthread_mutex_init(&lock, nullptr); 
    //创建子线程
    vector<pthread_t> tids;
    vector<ThreadInfo*> tis;
    for(int i = 1; i <= NUM; i++)
    {
        pthread_t tid;
        ThreadInfo *ti = new ThreadInfo("Thread-"+to_string(i), &lock);
        pthread_create(&tid, nullptr, GrabTickets, ti);
        tids.push_back(tid);
        tis.push_back(ti);
    }

    // 等待回收所有线程
    for(auto tid : tids)
    {
        pthread_join(tid, nullptr);
    }

    // 释放new的资源
    for(auto ti : tis)
    {
        delete ti;
    }

    // 释放动态分配的互斥量
    pthread_mutex_destroy(&lock); 

    return 0;
}

Makefile

mythread:mythread.cc
	g++ -o $@ $^ -std=c++11 -lpthread
.PHONY:clean
clean:
	rm -f mythread

由演示图，实现线程互斥后，每个抢票的线程不再会抢到相同的票，票的剩余数量也不再出现负数了。

3.互斥量的原理

互斥量也是一种资源，且所有线程都可以对其进行申请和释放，因此互斥量本身就是一个共享资源，而它的安全性是通过加锁和解锁操作本身所具有的原子性来保证的。

虽然操作系统内部并不存在锁的概念，调度器在调度轻量级进程时，也不会考虑是否有锁存在，但站在其他线程的角度，在一个线程持有锁的过程中只有“申请锁之前”和“申请锁之后”两种状态，因此站在其他线程的角度，一个线程持有锁的过程是具有原子性的。

而为了能让线程持有锁的过程是具有原子性的，大多数体系结构都提供了 swap 或 xchange 汇编指令，通过一条汇编指令来保证加锁的原子性。

//加锁解锁的汇编伪代码
lock:
	movb %al, $0
	xchange %al, mutex   //加锁过程中，xchange是原子的，可以保证锁的安全
	if(al寄存器的内容 > 0)
	{
		return 0;
	}
	else
	{
		挂起等待;	
	}
	goto lock;
 
unlock:
	movb mutex, $1
	唤醒等待mutex的线程;
	return 0;

xchange 汇编只有一条指令，这使得，只要一个线程通过xchage申请到了锁，就算在持有锁的过程中被切走，也是带走了锁了，其他线程是无法拿到锁的，只有等它将锁释放。

4.死锁

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

例如在多线程中，一个线程已经申请锁之后，另一个线程再一次申请锁，但锁已经被申请走了，另一个线程就只能在等待队列中等待，直到锁的资源被释放才被唤醒。
又例如，存在线程A、线程B、线程C、线程D，在不同的时间结点，分别占用资源和申请资源，起先，A 使用着 C 资源，B 使用着 D 资源，等到了某一个时间点， A 需要使用 D 资源完成任务，B 需要使用 C 资源完成任务，但 A 正用着C资源，B 也正用着 D 资源，一来二去，产生死锁了。

单线程也是可能产生死锁的。如果一个线程连续申请了两次锁，那么这个线程就会被挂起。这个线程第一次申请锁的时候，是能够成功的，但第二次申请时，由于锁已经被申请过了，于是导致申请失败，进而导致被挂起，直到锁被释放时才会被唤醒，然而，这个锁本来就在自己手上，自己又处于被挂起的状态，根本没有机会释放锁，所以这个线程永远不会被唤醒，也就产生死锁了。

【Tips】死锁的四个必要条件

当一个线程满足了以下四个条件，就可能产生死锁：

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

【Tips】如何避免死锁

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

补.特殊锁

.1）自旋锁

上文中的锁，是一种阻塞等待类型的锁，它的特点是，锁资源已被一个线程占用时，其他申请锁的线程会进入挂起状态，等到锁可用时再被唤醒去竞争锁资源。

但阻塞锁更适用于临界区的执行时间较长的情况，如果临界区的执行时间较短，来回的挂起和唤醒就会附带一定的性能损耗。
自旋锁与阻塞锁的不同在于，它可以让线程不用进入挂起等待状态，而是一直竞争直到持有锁资源为止，是一种非阻塞类型的锁。

//实现方式：
while(pthread_mutex_trylock(&mutex)){}
//1.如果pthread_mutex_trylock()返回0值，
//  表示竞争锁资源成功，循环条件不成立，线程将持有锁并执行后续代码。
//2.如果pthread_mutex_trylock()返回非0值，
//  表示竞争锁资源失败，循环条件成立，线程将重新竞争锁。


//补：自旋锁的相关接口:
int pthread_mutex_trylock(pthread_mutex_t *mutex);
//功能:申请自旋锁，线程竞争锁资源成功时会返回0，竞争锁资源失败时返回非0值。
int pthread_spin_init(pthread_spinlock_t *lock, int pshared);
//功能：对锁进行初始化操作。
//参数 pshared 为 PTHREAD_PROCESS_PRIVATE，表示锁只能在当前进程内使用，
//             为 PTHREAD_PROCESS_SHARED，表示锁能够被多个进程共享。
int pthread_spin_destroy(pthread_spinlock_t *lock);
//功能：对锁进行销毁操作。
int pthread_spin_lock(pthread_spinlock_t *lock);
//功能：竞争持有锁。
int pthread_spin_unlock(pthread_spinlock_t *lock);
//功能：解锁。

.2）读写锁

读写锁适用于访问读端多、写端少的资源的情况。对于读端多、写端少的资源，当写端想要对资源做修改时，就可能会因为竞争锁资源的能力相较于读端更弱，而无法持有锁，并且为了数据的安全，在写端对资源做修改时，读端是不能访问资源的。

由生产消费模型（详见下文），写端其实是生产者，读端其实是消费者，它们之间具有以下关系：

写端与写端之间是互斥的。
读端与写端之间既是同步的，也是互斥的。
读者与读者之间是共享的（由于读者是一种特殊的消费者，不会取走数据而只读取数据，因此读者之间并不存在互斥）。

//补：读写锁的相关接口：

int pthread_rwlock_init(pthread_rwlock_t *restrict rwlock,
       const pthread_rwlockattr_t *restrict attr);	
//功能：初始化锁
int pthread_rwlock_destroy(pthread_rwlock_t *rwlock);
//功能：释放锁
int pthread_rwlock_rdlock(pthread_rwlock_t *rwlock);
//功能：对读端加锁
int pthread_rwlock_wrlock(pthread_rwlock_t *rwlock);
//功能：对写端加锁
int pthread_rwlock_unlock(pthread_rwlock_t *rwlock);
//功能：对读端或写端解锁
int pthread_rwlockattr_setkind_np(pthread_rwlockattr_t *attr, int pref);
//功能：设置锁的优先级
//参数 pref 为 PTHREAD_RWLOCK_PREFER_READER_NP (默认设置) 读端优先，但可能会导致写端饥饿情况
//         为 PTHREAD_RWLOCK_PREFER_WRITER_NP 写端优先，但目前有BUG 会导致表现行为和 PTHREAD_RWLOCK_PREFER_READER_NP 一致
//         为 PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP 写端优先，但写端不能递归加锁

二、线程同步

同步可以更好地实现和完善互斥。

单纯的互斥，可能会导致线程饥饿问题，而同步可以让线程按照一定的顺序访问资源，使每个线程能够充分利用资源，提高程序执行效率。

【Tips】同步

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

1.竞态条件、条件变量

【Tips】竞态条件

因时序问题而导致程序异常的情况。

单纯的加锁是存在某些隐患的，如果个别线程的竞争力突出，使其每次都能够申请到锁，而申请到锁之后又偷懒不做任务，那么，这个线程其实相当于在一直不停地申请锁和释放锁，就会导致其他线程长时间竞争不到锁，引起线程饥饿问题。
尽管单纯的加锁本身是没有错的，能够保证在一段时间内有且仅有一个线程进入临界区，但它的问题在于，无法高效地让每一个线程使用这份临界资源。
于此，现在不妨增加一个规则：当一个线程释放锁后，这个线程不能立即再次申请锁，而要排到锁的资源等待队列的队尾进行等待，使下一个申请锁的线程一定是排在等待队列的队头，由此，就能够让多个线程按照某种次序进行临界资源的访问。这就是线程同步。
具体的例子如，假设当下有两个线程访问一块临界区，一个线程要往临界区写入数据，另一个线程要从临界区读取数据，但写端线程的竞争力更强，每次都更容易竞争到锁。在引入同步前，写端线程由于自身竞争力更强，可能会一直在执行写入操作，一直到临界区被写满后，写端线程就可能在一直不停地申请锁和释放锁。而读端线程由于竞争力较弱，每次都更难申请到锁，可能无法进行数据的读取，引起线程饥饿问题。在引入同步后，写端线程每申请一次锁、每执行完一次数据的写入操作、每释放一次锁，就会被加入到等待队列的队尾，使处于队头的读端线程可以正常地申请锁、读取数据、释放锁，从而避免了线程饥饿问题，也使得每个线程能够充分利用资源，提高了程序执行效率。

那，同步具体是如何实现的呢？

【Tips】条件变量

条件变量是一种针对某种资源是否就绪的数据化描述，是一种利用线程之间共享全局变量以实现同步的机制，通常配合互斥量一起使用。它主要包括以下两个动作：

一个线程因等待条件变量的条件成立而被挂起。
另一个线程使条件成立后唤醒等待的线程。

2.条件变量的相关接口

.1）初始化与销毁

初始化一个条件变量

#include<pthread.h>
int pthread_cond_init(pthread_cond_t *restrict cond,
        const pthread_condattr_t *restrict attr);
功能：以动态分配的方式初始化一个条件变量
参数：1.cond:一个指针，指向待初始化的条件变量。
     2.attr:条件变量的属性，不关心置为 NULL 即可。 
返回值：初始化成功返回0，失败返回错误码。
ps：以静态分配的方式初始化一个条件变量：pthread_cond_t con = PTHREAD_COND_INITIALIZER;
    此种方式不必用 pthread_cond_destroy() 销毁

销毁一个条件变量

#include<pthread.h>
int pthread_cond_destroy(pthread_cond_t *cond);
功能：销毁一个动态分配的条件变量
参数：一个指针，指向待销毁的条件变量。
返回值：初始化成功返回0，失败返回错误码。

.2）等待与唤醒

等待条件变量的条件成立

#include<pthread.h>
int pthread_cond_wait(pthread_cond_t *restrict cond, 
                pthread_mutex_t *restrict mutex);
功能：使一个线程等待条件变量的条件成立
参数：1.cond：一个指针，指向需等待的条件变量。
     2.mutex：当前线程所在临界区的，相应的互斥锁。
返回值：初始化成功返回0，失败返回错误码。

【ps】互斥锁是 pthread_cond_wait() 的参数之一的原因
条件等待是实现线程同步的一种手段。如果在一个条件变量下，只存在一个线程处于等待，那么就算线程能一直等下去，条件也始终不会满足，因此必须存在另一个线程通过某些操作改变共享变量，使原先不满足的条件变得满足，并通知在这个条件变量下等待中的线程。也就是说，条件等待这种同步手段适用于多线程之间。
而条件并不会无缘无故地突然满足，一定会与共享数据的修改有关，因此就需要互斥锁来保护数据的安全。
再者，如果在调用 pthread_cond_wait() 时，无需用到相关的互斥锁，那么，当线程进入临界区时，先加锁并判断内部资源的情况——不满足当前线程的执行条件——于是线程就在该条件变量下进行等待；但线程在被挂起的同时是拿着锁的，导致锁不再被释放了，进而导致死锁问题。因此，在调用 pthread_cond_wait() 时，还需要用到相关的互斥锁，让线程因条件不满足而进行等待时，释放它持有的互斥锁；直到线程被唤醒时，再拿回原有的互斥锁，继续执行临界区的代码。

【ps】pthread_cond_wait() 的调用，最好发生在加锁和解锁之间

为了 pthread_cond_wait() 能够将临界资源不就绪的相关线程挂起，首先需判断临界资源是否就绪，而判断临界资源是否就绪，涉及临界资源的访问，因此 pthread_cond_wait() 的调用最好发生在加锁和解锁之间。

【ps】pthread_cond_wait() 的使用规范
//...
pthread_mutex_lock(&mutex);
while (条件为假)
	pthread_cond_wait(&cond, &mutex);
修改条件
pthread_mutex_unlock(&mutex);
//...

唤醒等待中的线程

#include<pthread.h>
int pthread_cond_signal(pthread_cond_t *cond);
功能：唤醒一个处于等待队列队头的线程
参数：一个指针，指向当前线程所等待的条件变量（以此唤醒在cond条件变量下等待的一个线程）
返回值：初始化成功返回0，失败返回错误码。

#include<pthread.h>
int pthread_cond_broadcast(pthread_cond_t *cond);
功能：唤醒等待队列中的所有线程
参数：一个指针，指向当前线程所等待的条件变量（以此唤醒在cond条件变量下等待的所有线程）
返回值：初始化成功返回0，失败返回错误码。

.3）在上文引例中实现线程同步

mythread.cc：

#include <iostream>
#include <pthread.h>
#include <vector>
#include <cstdio>
#include <unistd.h>
#include <cassert>
using namespace std;

#define NUM 5       //假设有5人（5个线程）抢票

int tickets = 100;  // 定义1000张票

pthread_cond_t cond = PTHREAD_COND_INITIALIZER; // 以静态分配的方式初始化一个条件变量

//线程的基本信息
class ThreadInfo
{
public:
    ThreadInfo(const string &threadname, pthread_mutex_t *lock)
    :threadname_(threadname)
    ,lock_(lock)
    {}

public:
    string threadname_;    //线程的名字
    pthread_mutex_t *lock_;//线程申请的锁
};

//抢票
void *GrabTickets(void *args)
{
    ThreadInfo *ti = static_cast<ThreadInfo*>(args);
    string name(ti->threadname_);
    while(true)
    {   
        // 加锁
        int n = pthread_mutex_lock(ti->lock_); 
        assert(n == 0);
        // 进入等待队列
        pthread_cond_wait(&cond, ti->lock_); 
        if(tickets > 0)
        {
            usleep(10000);
            printf("%s get a ticket: %d\n", name.c_str(), tickets);
            tickets--;
            // 每抢一次票，解锁一次
            n = pthread_mutex_unlock(ti->lock_); 
            assert(n == 0);
        }
        else 
        {
            //抢完票后解锁
            n  = pthread_mutex_unlock(ti->lock_); 
            assert(n == 0);
            break;
        }
        // 用休眠来模拟抢到票的后续动作
        usleep(13);

    }

    printf("%s quit...\n", name.c_str());

    return NULL;
}

int main()
{
    //定义一个互斥量，并以动态分配的方式对其进行初始化
    pthread_mutex_t lock;               
    pthread_mutex_init(&lock, nullptr); 
    //创建子线程
    vector<pthread_t> tids;
    vector<ThreadInfo*> tis;
    for(int i = 1; i <= NUM; i++)
    {
        pthread_t tid;
        ThreadInfo *ti = new ThreadInfo("Thread-"+to_string(i), &lock);
        pthread_create(&tid, nullptr, GrabTickets, ti);
        tids.push_back(tid);
        tis.push_back(ti);
    }

    // 主线挨个唤醒等待中的子线程
    while(true)
    {
        sleep(1);
        pthread_cond_signal(&cond); 
        cout << "main thread wakeup a new thread" << endl;
    }

    // 等待回收所有线程
    for(auto tid : tids)
    {
        pthread_join(tid, nullptr);
    }

    // 释放new的资源
    for(auto ti : tis)
    {
        delete ti;
    }

    // 释放动态分配的互斥量
    pthread_mutex_destroy(&lock); 

    return 0;
}

Makefile：

mythread:mythread.cc
	g++ -o $@ $^ -std=c++11 -lpthread
.PHONY:clean
clean:
	rm -f mythread

由演示图，实现线程同步后，可见抢票的子线程在有序地、轮流地执行抢票操作。

三、生产消费模型

生产消费者模型（consumer producter ）是多线程多进程下同步互斥的一种场景，通过一个容器来解决生产者和消费者的强耦合问题。

1.三种关系、两种角色、一种容器

以生活为鉴，顾客、超市、供货商就是一个典型的生产消费模型。

超市就相当于是一个大型容器，可以存放一定量的商品。
顾客是消费者，会从超市里购买商品；供货商是生产者，会将商品放到超市里。
可能存在多个顾客要购买商品，而商品的数量有限，因此顾客和顾客之间存在竞争关系，换句话说，顾客和顾客之间是互斥关系。
可能存在多个供货商要向超市供货，而超市的容量有限，因此供货商和供货商之间也是互斥关系。
顾客与供货商之间没有直接的联系，而是经由超市这个中间媒介而存在间接的联系。
超市里原本是没有商品的，消费者需要等供货商将商品放到超市里，才能从超市里购买商品；而当超市里堆满商品的时候，供货商就不能继续将商品放到超市里了，需要等消费者购买了商品，将超市腾出一些空间，才能继续将商品放到超市里——为了平衡供需，顾客和供货商之间既要依照一定顺序去超市进行购买和供货，且在一方去超市进行购买或供货的时候，另一方不能去超市供进行货或购买，因此顾客和供货商之间既是互斥关系，又是同步关系。

回到线程，读取数据的线程叫做消费者线程，产生数据的线程叫做生产者线程，而它们之间共享的特定数据结构就叫做缓冲区。

【Tips】生产消费模型的特点

三种关系：生产者和生产者之间是互斥关系，消费者和消费者之间爷是互斥关系、生产者和消费者之间既是互斥关系又是同步关系。
两种角色： 生产者和消费者，通常由进程或线程承担。
一种容器：通常指内存中的一段缓冲区。
生产者和生产者、消费者和消费者、生产者和消费者，它们之间存在互斥关系，是因为，缓冲区是一种临界资源，可能会被多个执行流同时访问，需要互斥锁的保护；所有的生产者和消费者都会竞争式地申请锁。
生产者和消费者之间存在同步关系，是因为，如果让生产者一直生产数据，一旦缓冲区被塞满，生产者再生产的数据就无法被保存，进而丢失；反之，让消费者一直消费，一旦缓冲区被耗空，消费者就无法再消费了，继续消费可能导致非法访问；让生产者和消费者按一定顺序访问缓冲区，就可以有效避免上述问题，同时提高程序执行的效率。
互斥关系保证了数据的安全性，而同步关系保证了多线程之间的协同性。

【Tips】生产消费模型的优点

解耦。
支持并发。
支持忙闲不均。

【Tips】生产消费模型的解藕特性
以下面的代码为例：
int add(int x,int y)
{
    return x + y;
}
int main()
{
    int x,y;
    int z = add(x,y);
    return 0;
}
在 main 函数中调用 add 函数完成，因为是单执行流，所以main 函数只能等待（这种函数调用或称紧耦合）。

假如采用生产消费模型，让 main 函数是一个线程且充当生产者， add 函数是另一个线程且充当消费者，此时 main 函数这个线程在 add 线程进行计算的时候就不需要等待了，可以继续生产数据往超市里面放，add 线程现在也不用等 main 线程生产出一组数据再计算一组数据，而是直接去超市里面取数据进行计算。对 main 函数和 add 函数来说，此时就解藕了（这种函数调用或称松耦合）。

2.基于阻塞队列的生产消费模型

在多线程编程中，阻塞队列（Blocking Queue）可以作为生产消费模型中的“一种容器”，是一种常用于实现生产消费模型的数据结构。

它与普通队列的区别主要在于，当阻塞队列为空时，从队列获取元素的操作会被阻塞，直到元素被放入队列中；当队列存满时，往队列里存放元素的操作也会被阻塞，直到有元素从队列中被取出。

【Tips】阻塞队列的特点

作为“一种容器”，阻塞队列也是被多线程竞争的共享资源，因此需要互斥锁来实现生产者和消费者、生产者和生产者、消费者和消费者之间的互斥关系，以保证数据安全。
读取阻塞队列数据的操作（出队）只能由消费者来完成，且在消费者读取期间，生产者会因进入相应的等待队列而不能进行数据的写入。
改写阻塞队列数据的操作（入队）只能由生产者来完成，只有阻塞队列为空或未满时，生产者才能写入数据，且在生产者写入期间，消费者会因进入相应的等待队列而不能进行数据的读取。

基于阻塞队列的生产消费模型，根据竞争资源的生产者线程和消费者线程的数量，可以分为单生产单消费模型、多生产多消费模型。

.1）单生产单消费模型

单生产单消费模型中只有一个生产者线程和一个消费者线程，因此，相应的阻塞队列只需维护生产者与消费者之间的同步与互斥，而无需实现生产者与生产者之间、消费者与消费者之间的互斥。

BlockQueue.hpp：

//阻塞队列的实现
#pragma once
#include <iostream>
#include <pthread.h>
#include <queue>

template <class T>
class BlockQueue
{
    static const int defaultmaximum = 20;//阻塞队列的默认容量

public:
    //初始化
    BlockQueue(int maximum = defaultmaximum)
        : maximum_(maximum)
    {
        pthread_mutex_init(&mutex_, nullptr);
        pthread_cond_init(&c_cond_, nullptr);
        pthread_cond_init(&p_cond_, nullptr);
        low_water_ = maximum_ / 3;    // 低水位线是队列最大容量的 1/3
        high_water_ = (maximum_*2)/3; // 高水位线是队列最大容量的 2/3
    }
    
    //资源出队，由消费者负责
    T pop()
    {
        //加锁
        pthread_mutex_lock(&mutex_);
        // 1.消费条件不满足，就让消费者挂起等待，直到条件满足
        while (q_.size() == 0)
        { 
            pthread_cond_wait(&c_cond_, &mutex_);
        }
        // 2.消费条件满足
        //取队头
        T out = q_.front();
        q_.pop();
        // 若消费者将资源消费到队列的低水位线，就唤醒生产者进行生产
        if (q_.size() <= low_water_)
        {
            pthread_cond_signal(&p_cond_); 
            //生产者被唤醒的时候，消费者应挂起等待
            pthread_cond_wait(&c_cond_, &mutex_); 
            std::cout << "c is sleep..." << std::endl;
        }
        //解锁
        pthread_mutex_unlock(&mutex_);
        return out;
    }

    //资源入队，由生产者负责
    void push(const T& data)
    {
        //加锁
        pthread_mutex_lock(&mutex_);
        // 1.生产条件不满足，就让生产者挂起等待，直到条件满足
        while (q_.size() == maximum_) 
        {
            pthread_cond_wait(&p_cond_, &mutex_);
        }
        // 2.生产条件满足
        //将资源入队
        q_.push(data); 
        //若生产者将资源生产至队列的高水位线，就唤醒消费者进行消费
        if (q_.size() >= high_water_)
        {
            pthread_cond_signal(&c_cond_); 
            //消费者被唤醒的时候，生产者应挂起等待
            pthread_cond_wait(&p_cond_, &mutex_);
            std::cout << "p is sleep.." << std::endl;
        }
        //解锁
        pthread_mutex_unlock(&mutex_);
        
    }

    //销毁
    ~BlockQueue()
    {
        pthread_mutex_destroy(&mutex_);
        pthread_cond_destroy(&c_cond_);
        pthread_cond_destroy(&p_cond_);
    }

private:
    std::queue<T> q_;       // 生产者和消费者共享的阻塞队列
    int maximum_;           // 队列的最大容量
    pthread_mutex_t mutex_; // 定义一个互斥锁
    pthread_cond_t p_cond_; // 定义一个生产者条件变量，让生产者在这个条件变量下进行等待
    pthread_cond_t c_cond_; // 定义一个消费者条件变量，让消费者在这个条件变量下进行等待
    int low_water_;         // 队列的低水位线（队列最大容量的 1/3），控制生产
    int high_water_;        // 队列的高水位线（队列最大容量的 2/3），控制消费
};

【补】BlockQueue 的实现细节

要为生产者提供入队的接口（生产），为消费者提供出队的接口（消费）。
阻塞队列是共享资源，入队和出队涉及对共享资源的修改，故入队和出队的操作应发生在加锁和解锁之间。
生产者生产（入队）时，消费者不能消费（出队）；消费者消费（出队）时，生产者不能生产（入队）。
在加锁之后、实际进行生产（入队）或消费（出队）之前，应先判断是否满足生产条件（队列未满）或消费条件（队列不为空），若不满足条件，应通过 while 循环将线程挂起，直到条件满足再进行相应的生产或消费，避免两个线程在条件未满足时竞争锁而引起死锁。
当生产者生产了一定量的数据，就唤醒消费者去消费；当消费者消费了一定量的数据，就唤醒生产者去生产，这样可以实现生产和消费的同步。
程序运行期间，仅允许一个线程持有互斥锁，因此，不光条件未满足时要将相应的线程挂起，在唤醒另一个线程时，负责唤醒的线程也应被挂起。

ProdCon.cc

//单生产单消费模型的程序主体
#include "BlockQueue.hpp"
#include <unistd.h>
//消费者的例程
void *Consumer(void *args)
{
    BlockQueue<int> *bq = static_cast<BlockQueue<int>*>(args);

    while(true)
    {
        // 消费，即从队列里面拿数据
        int data = bq->pop();
        std::cout << "消费了一个数据: " << data << std::endl;
        usleep(1000000);
    }
}
//生产者的例程
void *Productor(void *args)
{
    BlockQueue<int> *bq = static_cast<BlockQueue<int>*>(args);
    int data = 1;
    while(true)
    {
        // 生产，即往队列里面放数据
        bq->push(data);
        std::cout << "生产了一个数据：" << data << std::endl;
        data++;
        usleep(100000);
    }
}

int main()
{
    //生成随机值
    srand((unsigned int)time(nullptr));
    //堆上申请队列
    BlockQueue<int> *bq = new BlockQueue<int>();

    pthread_t c, p;
    //创建消费者线程
    pthread_create(&c, nullptr, Consumer, bq);
    //创建生产者线程
    pthread_create(&p, nullptr, Productor, bq);

    //主线程回收消费者和生产者
    pthread_join(c, nullptr);
    pthread_join(p, nullptr);
    //释放new申请的队列
    delete bq; 
    return 0;
}

Makefile：

ProdCon:ProdCon.cc
	g++ -o $@ $^ -std=c++11 -lpthread
.PHONY:clean
clean:
	rm -f ProdCon

.2）基于任务的多生产多消费模型

多生产多消费模型中有多个生产者线程和多个消费者线程，因此，相应的阻塞队列既需维护生产者与生产者之间、消费者与消费者之间的互斥，又需维护生产者与消费者之间的同步与互斥。

多生产多消费是更加接近实际应用情景的，为更好地演示生产消费模型的作用，此处引入一个模拟实现的简易计算器，作为生产者和消费者共同的任务，并让生产者负责生产需计算的问题，让消费者负责解开问题的答案。

Task.hpp：

//模拟实现的简易计算器
#include <iostream>
#include <string>

//定义线程退出码
enum
{
    DIVERROR = 1, //除错误
    MODERROR,     //模错误
    UNKNOWERRROR  //未知错误
};

//简易计算器
class Task
{
public:
    //初始化成员变量
    Task(int a, int b, char op)
        :data1_(a), data2_(b), op_(op), result_(0), exitcode_(0)
    {}

    //完成实际的运算
    void run()
    {
        switch(op_)
        {
            case '+':
                result_ = data1_ + data2_;
                break;
            case '-':
                result_ = data1_ - data2_;
                break;
            case '*':
                result_ = data1_ * data2_;
                break;
            case '/':
                if(data2_ == 0) exitcode_ = DIVERROR;
                else result_ = data1_ / data2_;
                break;
            case '%':
                if(data2_ == 0) exitcode_ = MODERROR;
                else result_ = data1_ % data2_;
                break;
            default:
                exitcode_ = UNKNOWERRROR;
                break;
        }
    }

    //打印运算问题：计算数1 + 运算符 + 计算数2 
    std::string get_task()
    {
        std::string ret = std::to_string(data1_);
        ret += ' ';
        ret += op_;
        ret += ' ';
        ret += std::to_string(data2_);
        ret += ' ';
        ret += '=';
        ret += ' ';
        ret += '?';
        return ret;
    }   

    //打印运算结果：计算数1 + 运算符 + 计算数2 + 运算结果 + 退出码
    std::string result_to_string()
    {
        std::string ret = std::to_string(data1_);
        ret += ' ';
        ret += op_;
        ret += ' ';
        ret += std::to_string(data2_);
        ret += ' ';
        ret += '=';
        ret += ' ';
        ret += std::to_string(result_);
        ret += "[exitcode: ";
        ret += std::to_string(exitcode_);
        ret += ']';

        return ret;
    }

 
private:
    int data1_;   //计算数1
    int data2_;   //计算数2
    char op_;     //运算符
    int result_;  //运算结果
    int exitcode_;//线程退出码
};

Cal.cc ：

//多生产多消费模型的程序主体
#include "BlockQueue.hpp"
#include <unistd.h>
#include "Task.hpp"

//定义运算符
const std::string opers = "+-*/%";

// 生产者负责生产计算问题
void *Productor(void *args)
{
    int len = opers.size();
    BlockQueue<Task> *bq = static_cast<BlockQueue<Task> *>(args);
    int data = 1;
    while (true)
    {
        // 模拟生产数据的过程
        //生成计算数1、计算数2、运算符
        int data1 = rand() % 10 + 1; // [1, 10]
        usleep(10);
        int data2 = rand() % 13;     // [0, 13]
        usleep(10);
        char op = opers[rand() % len];
        //创建Task类的对象
        Task task(data1, data2, op);

        // 生产，即往队列里面放数据
        bq->push(task);
        std::cout << pthread_self() << "@ 生产了一个任务: " << task.get_task().c_str() << std::endl;
        usleep(1000000);
    }
}

// 消费者负责计算问题的结果
void *Consumer(void *args)
{
    BlockQueue<Task> *bq = static_cast<BlockQueue<Task> *>(args);

    while (true)
    {
        // 消费，即从队列里面拿数据
        Task task = bq->pop();

        // 模拟数据处理的过程
        task.run();
        std::cout << pthread_self() << "# 处理任务: " << task.get_task().c_str() << ", 运算结果是: " << task.result_to_string().c_str() << std::endl;
        usleep(1000000);
    }
}


int main()
{
    //生成随机数
    srand((unsigned int)time(nullptr));
    //在堆上申请队列
    BlockQueue<Task> *bq = new BlockQueue<Task>();

    pthread_t c[3], p[5];
    //创建消费者线程
    for (int i = 0; i < 3; i++)
    {
        pthread_create(c + i, nullptr, Consumer, bq);
    }
    //创建生产者线程
    for (int i = 0; i < 5; i++)
    {
        pthread_create(p+i, nullptr, Productor, bq);
    }

    // 主线程回收消费者和生产者
    for(int i = 0; i < 3; i++)
    {
        pthread_join(c[i], nullptr);
    }
    for(int i = 0; i < 5; i++)
    {
        pthread_join(p[i], nullptr);
    }
    // 释放new申请的队列
    delete bq; 

    return 0;
}

BlockQueue.hpp：

//阻塞队列的实现
#pragma once
#include <iostream>
#include <pthread.h>
#include <queue>

template <class T>
class BlockQueue
{
    static const int defaultmaximum = 20;//阻塞队列的默认容量

public:
    //初始化
    BlockQueue(int maximum = defaultmaximum)
        : maximum_(maximum)
    {
        pthread_mutex_init(&mutex_, nullptr);
        pthread_cond_init(&c_cond_, nullptr);
        pthread_cond_init(&p_cond_, nullptr);
        low_water_ = maximum_ / 3;    // 低水位线是队列最大容量的 1/3
        high_water_ = (maximum_*2)/3; // 高水位线是队列最大容量的 2/3
    }
    
    //资源出队，由消费者负责
    T pop()
    {
        //加锁
        pthread_mutex_lock(&mutex_);
        // 1.消费条件不满足，就让消费者挂起等待，直到条件满足
        while (q_.size() == 0)
        { 
            pthread_cond_wait(&c_cond_, &mutex_);
        }
        // 2.消费条件满足
        //取队头
        T out = q_.front();
        q_.pop();
        // 若消费者将资源消费到队列的低水位线，就唤醒生产者进行生产
        if (q_.size() <= low_water_)
        {
            pthread_cond_signal(&p_cond_); 
            //生产者被唤醒的时候，消费者应挂起等待
            pthread_cond_wait(&c_cond_, &mutex_); 
            std::cout << "c is sleep..." << std::endl;
        }
        //解锁
        pthread_mutex_unlock(&mutex_);
        return out;
    }

    //资源入队，由生产者负责
    void push(const T& data)
    {
        //加锁
        pthread_mutex_lock(&mutex_);
        // 1.生产条件不满足，就让生产者挂起等待，直到条件满足
        while (q_.size() == maximum_) 
        {
            pthread_cond_wait(&p_cond_, &mutex_);
        }
        // 2.生产条件满足
        //将资源入队
        q_.push(data); 
        //若生产者将资源生产至队列的高水位线，就唤醒消费者进行消费
        if (q_.size() >= high_water_)
        {
            pthread_cond_signal(&c_cond_); 
            //消费者被唤醒的时候，生产者应挂起等待
            pthread_cond_wait(&p_cond_, &mutex_);
            std::cout << "p is sleep.." << std::endl;
        }
        //解锁
        pthread_mutex_unlock(&mutex_);
        
    }

    //销毁
    ~BlockQueue()
    {
        pthread_mutex_destroy(&mutex_);
        pthread_cond_destroy(&c_cond_);
        pthread_cond_destroy(&p_cond_);
    }

private:
    std::queue<T> q_;       // 生产者和消费者共享的阻塞队列
    int maximum_;           // 队列的最大容量
    pthread_mutex_t mutex_; // 定义一个互斥锁
    pthread_cond_t p_cond_; // 定义一个生产者条件变量，让生产者在这个条件变量下进行等待
    pthread_cond_t c_cond_; // 定义一个消费者条件变量，让消费者在这个条件变量下进行等待
    int low_water_;         // 队列的低水位线（队列最大容量的 1/3），控制生产
    int high_water_;        // 队列的高水位线（队列最大容量的 2/3），控制消费
};

Makefile：

Cal:Cal.cc
	g++ -o $@ $^ -std=c++11 -lpthread
.PHONY:clean
clean:
	rm -f Cal

.补）RAII 风格的互斥锁

lockGuard.hpp：

#pragma once
#include <iostream>
#include <pthread.h>
 
class Mutex
{
public:
    Mutex(pthread_mutex_t* mtx) 
        :_pmtx(mtx)
    {}
    void lock()
    {
        pthread_mutex_lock(_pmtx);
        std::cout << "加锁成功" << std::endl;
    }
    void unlock()
    {
        pthread_mutex_unlock(_pmtx);
        std::cout << "解锁成功" << std::endl;
    }
    ~Mutex()
    {}
protected:
    pthread_mutex_t* _pmtx;
};
 
class lockGuard 
{
public:
    lockGuard(pthread_mutex_t* mtx) 
        :_mtx(mtx)
    {
        _mtx.lock();
    }
    ~lockGuard()
    {
        _mtx.unlock();
    }
protected:
    Mutex _mtx;
};

BlockQueue.hpp：

//将 RAII 风格的互斥锁应用到阻塞队列中
#pragma once
#include <iostream>
#include <pthread.h>
#include <queue>
#include "lockGuard.hpp"

template <class T>
class BlockQueue
{
    static const int defaultmaximum = 20;//阻塞队列的默认容量

public:
    //初始化
    BlockQueue(int maximum = defaultmaximum)
        : maximum_(maximum)
    {
        pthread_mutex_init(&mutex_, nullptr);
        pthread_cond_init(&c_cond_, nullptr);
        pthread_cond_init(&p_cond_, nullptr);
        low_water_ = maximum_ / 3;   
        high_water_ = (maximum_*2)/3; 
    }
    
    T pop()
    {
        // lockgrard 会自动调用构造函数初始化，同时完成加锁
        lockGuard lockgrard(&mutex_);

        // 1.消费条件不满足，就让消费者挂起等待，直到条件满足
        while (q_.size() == 0)
        { 
            pthread_cond_wait(&c_cond_, &mutex_);
        }
        // 2.消费条件满足
        //取队头
        T out = q_.front();
        q_.pop();
        // 若消费者将资源消费到队列的低水位线，就唤醒生产者进行生产
        if (q_.size() <= low_water_)
        {
            pthread_cond_signal(&p_cond_); 
            pthread_cond_wait(&c_cond_, &mutex_); 
            std::cout << "c is sleep..." << std::endl;
        }

        //pthread_mutex_unlock(&mutex_);

        return out;

        //出作用域后，lockgrard 自动调用析构函数销毁，同时完成解锁
    }

    void push(const T& data)
    {
        // lockgrard 会自动调用构造函数初始化，同时完成加锁
        lockGuard lockgrard(&mutex_);

        // 1.生产条件不满足，就让生产者挂起等待，直到条件满足
        while (q_.size() == maximum_) 
        {
            pthread_cond_wait(&p_cond_, &mutex_);
        }
        // 2.生产条件满足
        //将资源入队
        q_.push(data); 
        //若生产者将资源生产至队列的高水位线，就唤醒消费者进行消费
        if (q_.size() >= high_water_)
        {
            pthread_cond_signal(&c_cond_); 
            pthread_cond_wait(&p_cond_, &mutex_);
            std::cout << "p is sleep.." << std::endl;
        }

        //pthread_mutex_unlock(&mutex_);

        //出作用域后，lockgrard 自动调用析构函数销毁，同时完成解锁
        
    }

    //销毁
    ~BlockQueue()
    {
        pthread_mutex_destroy(&mutex_);
        pthread_cond_destroy(&c_cond_);
        pthread_cond_destroy(&p_cond_);
    }

private:
    std::queue<T> q_;       // 生产者和消费者共享的阻塞队列
    int maximum_;           // 队列的最大容量
    pthread_mutex_t mutex_; // 定义一个互斥锁
    pthread_cond_t p_cond_; // 定义一个生产者条件变量，让生产者在这个条件变量下进行等待
    pthread_cond_t c_cond_; // 定义一个消费者条件变量，让消费者在这个条件变量下进行等待
    int low_water_;         // 队列的低水位线（队列最大容量的 1/3），控制生产
    int high_water_;        // 队列的高水位线（队列最大容量的 2/3），控制消费
};

3.POSIX 信号量

.1）基本原理

POSIX 信号量和 System V 信号量的原理基本相同，都可以实现无冲突地访问临界资源，但 POSIX 信号量主要服务于线程同步，System V 信号量主要服务于进程或线程间的互斥。

System V 信号量是 system V IPC 所提供的一种通信方式，用于保证进程间的同步与互斥。在【Linux系统】进程间通信-CSDN博客一篇中，已对 System V 信号量的原理作了详细的阐述，在此恕不赘述。

临界资源也可以被划分为多份，只要规定好线程的临界区，就可以让多个线程并发访问临界资源。

POSIX 信号量本质也是一把计数器，用于描述可用临界资源的数量。在申请 POSIX 信号量时，已经涉及了临界资源的访问操作，间接判断了临界资源是否就绪，因此，只要成功申请到 POSIX 信号量，相关的临界资源就一定是就绪的。

POSIX 信号量不让多余的线程访问临界资源，如果临界资源只有十份，POSIX 信号量就不会允许同时有十一个线程对其访问。但如果真的出现一个临界资源同时被两个线程访问了，大概率跟代码中的资源分配操作有关，属于编码 Bug。

.2）相关接口

#include <semaphore.h>
int sem_init(sem_t *sem, int pshared, unsigned int value);
功能：初始化信号量
参数：1.sem：要初始化的信号量
     2.pshared：0表示线程间共享，非0表示进程间共享。
     3.value：信号量初始值

int sem_destroy(sem_t *sem);
功能：销毁信号量

int sem_wait(sem_t *sem); 
功能：等待/申请信号量，会对信号量做减减操作（简称P操作）

int sem_post(sem_t *sem);
功能：发布/释放信号量，会对信号量做加加操作（简称V操作）

以上接口，返回值均为：调用成功返回0，失败返回-1，并且设置合适的错误码。

.3）在上文引例中引入二元信号量

若信号量的初始值为 1，则说明信号量所描述的临界资源只有一份，而这种值为 1 的信号量被称为二元信号量，其作用基本等同于互斥锁。

mythread.cc：

#include <iostream>
#include <pthread.h>
#include <vector>
#include <cstdio>
#include <unistd.h>
#include <semaphore.h>
using namespace std;

class Sem{
public:
	Sem(int num)
	{
		sem_init(&_sem, 0, num);
	}
	~Sem()
	{
		sem_destroy(&_sem);
	}
	void P()
	{
		sem_wait(&_sem);
	}
	void V()
	{
		sem_post(&_sem);
	}
private:
	sem_t _sem;
};

Sem sem(1);         //二元信号量

#define NUM 5       //假设有5人（5个线程）抢票

int tickets = 100;  // 定义1000张票

pthread_cond_t cond = PTHREAD_COND_INITIALIZER; // 以静态分配的方式初始化一个条件变量


//线程的基本信息
class ThreadInfo
{
public:
    ThreadInfo(const string &threadname, pthread_mutex_t *lock)
    :threadname_(threadname)
    ,lock_(lock)
    {}

public:
    string threadname_;    //线程的名字
    pthread_mutex_t *lock_;//线程申请的锁
};

//抢票
void *GrabTickets(void *args)
{
    ThreadInfo *ti = static_cast<ThreadInfo*>(args);
    string name(ti->threadname_);
    while(true)
    {   
        // 加锁
        sem.P();
        // 进入等待队列
        pthread_cond_wait(&cond, ti->lock_); 
        if(tickets > 0)
        {
            usleep(10000);
            printf("%s get a ticket: %d\n", name.c_str(), tickets);
            tickets--;
            // 每抢一次票，解锁一次
            sem.V();
        }
        else 
        {
            //抢完票后解锁
            sem.V();
            break;
        }
        // 用休眠来模拟抢到票的后续动作
        usleep(13);

    }

    printf("%s quit...\n", name.c_str());

    return NULL;
}

int main()
{
    //定义一个互斥量，并以动态分配的方式对其进行初始化
    pthread_mutex_t lock;               
    pthread_mutex_init(&lock, nullptr); 
    //创建子线程
    vector<pthread_t> tids;
    vector<ThreadInfo*> tis;
    for(int i = 1; i <= NUM; i++)
    {
        pthread_t tid;
        ThreadInfo *ti = new ThreadInfo("Thread-"+to_string(i), &lock);
        pthread_create(&tid, nullptr, GrabTickets, ti);
        tids.push_back(tid);
        tis.push_back(ti);
    }

    // 主线挨个唤醒等待中的子线程
    while(true)
    {
        sleep(1);
        pthread_cond_signal(&cond); 
        cout << "main thread wakeup a new thread" << endl;
    }

    // 等待回收所有线程
    for(auto tid : tids)
    {
        pthread_join(tid, nullptr);
    }

    // 释放new的资源
    for(auto ti : tis)
    {
        delete ti;
    }

    // 释放动态分配的互斥量
    pthread_mutex_destroy(&lock); 

    return 0;
}

Makefile：

mythread:mythread.cc
	g++ -o $@ $^ -std=c++11 -lpthread
.PHONY:clean
clean:
	rm -f mythread

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

环形队列也可以作为生产消费模型中的“一种容器”，相较于阻塞队列，它更容易控制生产者和消费者之间的同步和互斥。

.1）单生产单消费模型

临界资源包括可用的空间资源、可访问的数据资源等，可以被划分为多份来管理，在基于环形队列的生产消费模型中，生产者关注的是这多份的空间资源，而消费者关注的是这多份的数据资源。

通过对空间资源的申请和释放、对数据资源的申请和释放，可以实现生产者和消费者之间的同步和互斥，其中，对多份的空间资源和多份的数据资源的描述工作，由信号量来负责；而多份的空间资源和多份的数据资源的存储工作，由环形队列来负责；至于生产者和消费者之间的同步和互斥，则由信号量和环形队列协作完成。

【Tips】如何管理空间资源和数据资源

基于环形队列的生产消费模型，是由环形队列和信号量协作实现的。根据环形队列和信号量的特点，大块的临界资源可被划分为多份，临界资源的类型可分为空间资源和数据资源，其中，环形队列主要负责资源的存储，信号量主要负责资源的描述。

（1）环形队列的存储和访问

        生产者和消费者关注的资源类型自然有所不同，其中，生产者关注的是环形队列中是否有空间，只要环形队列尚有空间，生产者就可以进行生产；而消费者关注的是环形队列中是否有数据，只要环形队列尚有数据，消费者就可以进行消费。

（2）信号量的描述

临界资源的类型被分为空间资源和数据资源两种，因此描述资源的信号量也相应有两种。

        由于初始时，环形队列为空，其中的空间均可用，因此描述空间资源的信号量（下称 pspace_sem）初始值应为环形队列的容量；且由于初始时，环形队列为空，其中没有数据可用，因此描述数据资源的信号量（下称 cdata_sem）初始值应为 0 。

每当一份空间资源被申请或被释放，pspace_sem 的值要相应地 - 1 或 + 1；每当一份数据资源被申请或被释放，cdata_sem 的值要相应地 + 1 或 - 1。

（3）资源的申请和释放

        生产者申请空间资源，而释放数据资源。

        在进行一次生产前，生产者要先申请 pspace_sem 信号量，若申请时 pspace_sem 的值非 0 （说明队列未满），则申请成功，同时对 pspace_sem 做减减操作（P操作），接下来可以进行生产；若申请时 pspace_sem 的值为 0（说明队列已满），则申请失败，生产者会去 pspace_sem 的等待队列下挂起，直到有可用的空间资源后再被唤醒。

在完成一次生产后，生产者要将生产数据入队的同时，还要释放 cdata_sem，对 cdata_sem 做加加操作（V操作），使队列中原本由空间资源占用的位置，现在变成数据资源在占用。

        消费者申请数据资源，而释放空间资源。

        在进行一次消费前，消费者要先申请 cdata_sem 信号量，若申请时 cdata_sem 的值非 0（说明队列中有数据），则申请成功，同时对 cdata_sem 做加加操作（V操作），接下来可以进行消费；若申请时 pspace_sem 的值为 0（说明队列为空），则申请失败，消费者会去 cdata_sem 的等待队列下挂起，直到有可用的数据资源后再被唤醒。

        在完成一次消费后，消费者要将自己消费的数据出队，同时还要释放 pspace_sem ，对 pspace_sem 做减减操作（P操作），使队列中原本由数据资源占用的位置，现在变成空间资源在占用。

【Tips】生产者和消费者如何访问资源

（1）不同时期，生产者和消费者在环形队列中所处的位置

环形队列恰好为空（有空间资源，无数据资源）或恰好为满（无空间资源，有数据资源）时，生产者和消费者处于同一位置；环形队列未满时（既有空间资源，又有数据资源），生产者和消费者处于不同位置。

（2）生产和消费都在进行时，生产者和消费者不能同时访问环形队列中的同一个位置。

环形队列中的任意一个位置都有双重含义，当这个位置没有元素的时候，意味着这是空间资源，当这个位置有元素存在的时候，意味着这是数据资源。

如果生产者和消费者同时访问了队列中的同一个位置，就意味着它们对同一份临界资源进行了访问操作，可能造成数据不一致等问题。

因此，生产和消费都在进行时，同一时刻下，生产者和消费者必须访问的是环形队列中的不同位置，此时生产者和消费者是可以同时进行生产和消费的，既实现了线程的同步和互斥，也避免了数据不一致等问题。

（3）生产者的生产一定先于消费者的消费

消费者消费的数据是由生产者生产的，没有生产就没有消费，因此生产者的生产一定先于消费者的消费，生产者和消费者在环形队列中动态的相对位置，是消费者不断追及生产者的过程，消费者可以紧跟着生产者（至少差一个位置），但不能追上生产者（位置重叠），甚至超过生产者。

（4）在消费者不断追及生产者的过程中，生产者不能甩开消费者一个环形队列的容量

如果消费者的消费速度整体慢于生产者的生产速度，就势必会导致环形队列被填满，以及生产者和消费者的位置发生重叠，此时两个线程访问了同一份临界资源，可能造成数据不一致等问题。

且此时如果生产者继续生产，其位置就会越过消费者，甩开消费者一圈以上的距离，开始覆盖原先生产的数据，造成数据的丢失。

因此，在消费者不断追及生产者的过程中，生产者也要照顾消费者的消费速度，不能生产得过快，以至于甩开消费者整整一圈（一个环形队列的容量）。

【Tips】生产者的伪代码：

pspace_sem = 环形队列的容量;
 
P(pspace_sem);//申请空间资源
//申请成功，继续向下运行。
//申请失败，阻塞在申请处。
 
.......//从事生产活动，将数据放入队列中
 
V(pspace_sem);//归还数据资源

【Tips】消费者的伪代码：

cdata_sem = 0;
 
P(cdata_sem);//申请数据资源
//申请成功，继续向下运行。
//申请失败，阻塞在申请处。
 
.......//从事消费活动，从队列中取数据
 
V(cdata_sem);//归还空间资源

以下为代码实现：

RingQueue.hpp：

//环形队列的具体实现
#pragma once
#include <pthread.h>
#include <vector>
#include <semaphore.h>

//环形队列
template<class T>
class RingQueue 
{
private:
    static const int defaultcap = 5;
    // 申请一个信号量
    void P(sem_t *sem) 
    {
        sem_wait(sem); 
    }
    // 释放一个信号量
    void V(sem_t *sem) 
    {
        sem_post(sem);
    }
public:
    RingQueue(int cap = defaultcap)
    :ringqueue_(cap), cap_(cap), c_step(0), p_step(0)
    {
        sem_init(&cdata_sem, 0, 0);
        sem_init(&pspace_sem, 0, cap_);
    }
    
    //生产（入队）
    void Push(const T &data) 
    {
        //申请空间资源
        P(&pspace_sem);
        //将生产的数据入队
        ringqueue_[p_step] = data;
        //释放数据资源
        V(&cdata_sem);
        //调整生产者下一个要生产的位置
        p_step++;
        p_step %= cap_;//防越界
    }

    //消费（出队）
    void Pop(T *out) 
    {
        //申请数据资源
        P(&cdata_sem);
        //将消费的数据从队列中取出
        *out = ringqueue_[c_step];
        //释放空间资源
        V(&pspace_sem);
        //调整消费者下一个要消费的位置
        c_step++;
        c_step %= cap_;//防越界
    }

    ~RingQueue()
    {
        sem_destroy(&cdata_sem);
        sem_destroy(&pspace_sem);
    }
private:
    std::vector<T> ringqueue_; // 用一个 vector 模拟环形队列
    int cap_;    // 环形队列的容量
    int c_step;  // 消费者下一个要消费的位置
    int p_step;  // 生产者下一个要生产的位置
    sem_t pspace_sem; // 空间资源信号量
    sem_t cdata_sem;  // 数据资源信号量

};

Main.cc：

//单生产单消费模型的程序主体
#include "RingQueue.hpp"
#include <iostream>
#include <unistd.h>
using namespace std;

//生产者例程
void *Producer(void *args)
{
    RingQueue<int> *rq = static_cast<RingQueue<int>*>(args);
    while(true)
    {
        usleep(10000); 
        int data = rand() % 10;
        rq->Push(data);
        cout << "Producer is running... produce a data: " << data << endl;
    }
}

//消费者例程
void *Consumer(void *args)
{
    RingQueue<int> *rq = static_cast<RingQueue<int>*>(args);
    while(true)
    {
        int data = 0;
        rq->Pop(&data);
        cout << "Consumer is running... get a data: " << data << endl;
        usleep(1000000);
    }
}

int main()
{
    //生成随机数
    srand((unsigned int)time(nullptr));
    //在堆上申请一个环形队列
    RingQueue<int> *rq = new RingQueue<int>();
    //创建消费者线程和生产者线程
    pthread_t c, p;
    pthread_create(&c, nullptr, Consumer, rq);
    pthread_create(&p, nullptr, Producer, rq);
    //主线程回收消费者和生产者
    pthread_join(c, nullptr);
    pthread_join(p, nullptr);
    //释放new申请的环形队列
    delete rq;

    return 0;
}

Makefile：

Main:Main.cc
	g++ -o $@ $^ -std=c++11 -lpthread
.PHONY:clean
clean:
	rm -f Main

.2）多生产多消费模型

多生产多消费模型不仅涉及生产者和消费者之间的同步和互斥，还涉及生产者与生产者之间、消费者与消费者之间的互斥。

为了维护生产者与生产者之间、消费者与消费者之间的互斥关系，且保护环形队列中任意位置上的临界资源，就需要在单生产单消费模型的基础上，用到互斥锁。

环形队列中，为了维护生产者和消费者之间的同步和互斥，已经涉及了信号量的申请，而互斥锁的申请应该发生在信号量的申请之后。

这是因为，如果先加锁再申请信号量的话，那么申请信号量的代码就位于临界区中，使得申请互斥锁和申请信号量的动作是串行的，始终就只有一个生产者线程或消费者线程能持有锁，同时也就只有这一个线程能去申请信号量，而其他线程只能挂起等待锁被释放。

如果先申请信号量的话，虽然一段时间内也只有一个线程能够持有锁，但是其他线程还可以先去申请信号量。信号量的申请本身具有原子性，无需加锁保护，只要信号量能够被申请，就说明还有临界资源可用。而申请到信号量的线程就挂起等待锁被释放，拿到锁之后去可以直接去执行临界区的代码。

【Tips】在多生产多消费模型中，要先申请信号量，再申请互斥锁。

RingQueue.hpp：

//环形队列的具体实现
#pragma once
#include <vector>
#include <string>
#include <pthread.h>
#include <semaphore.h>


//环形队列
template<class T>
class RingQueue 
{
private:
    static const int defaultcap = 5;

    // 申请一个信号量
    void P(sem_t *sem) 
    {
        sem_wait(sem); 
    }
    // 释放一个信号量
    void V(sem_t *sem) 
    {
        sem_post(sem);
    }
    // 加锁
    void Lock(pthread_mutex_t *mutex)
    {
        pthread_mutex_lock(mutex);
    }
    // 解锁
    void Unlock(pthread_mutex_t *mutex)
    {
        pthread_mutex_unlock(mutex);
    }

public:
    RingQueue(int cap = defaultcap)
        :ringqueue_(cap), cap_(cap), c_step(0), p_step(0)
    {
        sem_init(&cdata_sem, 0, 0);
        sem_init(&pspace_sem, 0, cap_);
    }
    
    //生产（入队）
    void Push(const T &data) 
    {
        //申请空间资源
        P(&pspace_sem);
        //加锁
        Lock(&p_mutex);
        //将生产的数据入队
        ringqueue_[p_step] = data;
        //调整生产者下一个要生产的位置
        p_step++;
        p_step %= cap_;//防越界
        //解锁
        Unlock(&p_mutex);
        //释放数据资源
        V(&cdata_sem);
    }

    //消费（出队）
    void Pop(T *out) 
    {
        //申请数据资源
        P(&cdata_sem);
        //加锁
        Lock(&c_mutex);
        //将消费的数据从队列中取出
        *out = ringqueue_[c_step];
        //调整消费者下一个要消费的位置
        c_step++;
        c_step %= cap_;//防越界
        //解锁
        Unlock(&c_mutex);
        //释放空间资源
        V(&pspace_sem);

    }

    ~RingQueue()
    {
        sem_destroy(&cdata_sem);
        sem_destroy(&pspace_sem);
        pthread_mutex_destroy(&c_mutex);
        pthread_mutex_destroy(&p_mutex);
    }
private:
    std::vector<T> ringqueue_; // 用一个 vector 模拟环形队列
    int cap_;    // 环形队列的容量
    int c_step;  // 消费者下一个要消费的位置
    int p_step;  // 生产者下一个要生产的位置
    sem_t pspace_sem; // 空间资源信号量
    sem_t cdata_sem;  // 数据资源信号量
    pthread_mutex_t c_mutex;  // 保护消费位置的互斥锁
    pthread_mutex_t p_mutex;  // 保护生产位置的互斥锁
};


//这里定义一个Message类，方便演示代码的运行
template <class T>
class Message
{
public:
    Message(std::string thread_name, RingQueue<T> *ringqueue)
        :thread_name_(thread_name), ringqueue_(ringqueue)
    {}

    std::string &get_thread_name()
    {
        return thread_name_;
    }

    RingQueue<T> *get_ringqueue()
    {
        return ringqueue_;
    }
private:
    std::string thread_name_;//线程名
    RingQueue<T> *ringqueue_;//环形队列
};

Main.cc：

//多生产多消费模型的程序主体
#include "RingQueue.hpp"
#include <iostream>
#include <unistd.h>
#include <vector>
using namespace std;

//消费者例程
void *Consumer(void *args)
{
    Message<int> *message = static_cast<Message<int> *>(args);
    RingQueue<int> *rq = message->get_ringqueue();
    string name = message->get_thread_name();
    while (true)
    {
        int data = 0;
        rq->Pop(&data);
        printf("%s is running... get a data: %d\n", name.c_str(), data);
    }
}

//生产者例程
void *Producer(void *args)
{
    Message<int> *message = static_cast<Message<int> *>(args);
    RingQueue<int> *rq = message->get_ringqueue();
    string name = message->get_thread_name();
    while (true)
    {
        int data = rand() % 10;
        rq->Push(data);
        printf("%s is running... produce a data: %d\n", name.c_str(), data);
        usleep(1000000);
    }
}

int main()
{
    //生成随机数
    srand((unsigned int)time(nullptr));
    //在堆上申请环形队列
    RingQueue<int> *rq = new RingQueue<int>(); 
    //集中管理 Message 对象
    vector<Message<int>*> messages; 

    pthread_t c[3], p[5];
    //先创建生产者
    for (int i = 0; i < 5; i++)
    {
        Message<int> *message = new Message<int>("Producer Thread "+to_string(i), rq);
        pthread_create(p + i, nullptr, Producer, message);
        messages.push_back(message);
    }
    //再创建消费者
    for (int i = 0; i < 3; i++)
    {
        Message<int> *message = new Message<int>("Consumer Thread "+to_string(i), rq);
        pthread_create(c + i, nullptr, Consumer, message);
        messages.push_back(message);
    }

    //主线程回收消费者和生产者
    for (int i = 0; i < 3; i++)
    {
        pthread_join(c[i], nullptr);
    }
    for (int i = 0; i < 5; i++)
    {
        pthread_join(p[i], nullptr);
    }
    //释放new的资源
    for (auto message : messages)
    {
        delete message;
    }
    delete rq;

    return 0;
}

Makefile：

Main:Main.cc
	g++ -o $@ $^ -std=c++11 -lpthread
.PHONY:clean
clean:
	rm -f Main

四、线程池

1.池化技术

在一台计算机中，磁盘是存储数据的主力，内存是加载数据的主力，但磁盘处理数据的效率远不如内存，因此为了提高计算机的运行效率，内存中被设计了一块类似于“池”的空间，磁盘中会被读取的数据将按需暂存在这块“池”空间中，这样就使得数据的处理操作大部分都发生在内存中。

这就好比，从前有座山，山顶有座庙，庙里有许多和尚要喝水，而水在山下的湖里，每次下山去湖里打水再返回庙里，来来回回十分麻烦，于是为了更方便地取水和用水，和尚们在半山腰建了一个池子来储水，这样一来，定时把湖水定量地送往半山腰，等每次庙里缺水了就只需要到半山腰的池子打水即可，节省了大量上山下山的时间和人力。

而这就是池化技术，所谓池化就是将原本要跑很远、跑多次才能拿到的东西，按需屯在往返中途的"池"中，从此以后往"池"中存、从"池"中取。

【Tips】池化技术的优点

减少内存碎片化：内存池化技术通过预先分配一定大小的内存块，并在程序运行过程中重复使用这些内存块，避免了频繁地进行内存分配和释放操作。这样可以减少内存碎片化的问题，提高内存的利用率。
降低内存管理开销：频繁的内存分配和释放操作会带来较大的开销，包括时间开销和空间开销。而通过内存池化技术，可以避免多次的内存分配和释放，从而大大降低了内存管理的开销，提高了程序的运行效率。
提升程序性能：通过减少内存碎片化、降低内存管理开销，内存池化技术可以提升程序的整体性能。它能够减少因频繁内存操作而导致的性能下降，使程序更高效地利用内存资源，加快数据的访问速度，提高程序的响应能力和执行效率。

池化技术的应用有进程池、线程池等。

在之前的博客中（【Linux系统】进程间通信-CSDN博客），匿名管道一节谈及过进程池。进程的创建会伴随着系统资源的消耗，如果频繁的申请和释放进程资源，就会对计算机运行的性能造成一定损耗，而如果一次性申请一批资源，就可以避免频繁的申请，从而保障了运行的高效性。

2.线程池的实现

线程池是一种线程使用模式，也是池化技术的一种体现。

多线程的创建会伴随着系统资源的开销，多线程的调度也会伴随着 CPU 调度的开销，线程一旦过多就会影响缓存局部和整体性能，而这个问题可以交由线程池来解决。

线程池可以维护多个线程，等待着监督管理者分配可并发执行的任务，避免了在处理短时间任务时创建与销毁线程的代价，不仅能够保证内核充分利用，还能防止过分调度。

【ps】线程池中线程数量的说明

线程池中可用线程的数量应该取决于可用的并发处理器、处理器内核、内存、网络sockets等的数量。

【Tips】线程池的应用场景

需要大量的线程来完成任务，且完成任务的时间较短。
适用于对性能有苛刻要求的应用，例如要求服务器迅速响应客户请求。
适用于接受突发性的大量请求的、但不至于使服务器因此产生大量线程的应用。

下面实现一个简单的线程池，线程池中存在一个任务队列和多个线程。，并模拟上文中并发的计算任务。

Task.hpp

//模拟实现的简易计算器（与上文一致）
#include <iostream>
#include <string>

//定义线程退出码
enum
{
    DIVERROR = 1, //除错误
    MODERROR,     //模错误
    UNKNOWERRROR  //未知错误
};

//简易计算器
class Task
{
public:
    //初始化成员变量
    Task(int a, int b, char op)
        :data1_(a), data2_(b), op_(op), result_(0), exitcode_(0)
    {}

    //完成实际的运算
    void run()
    {
        switch(op_)
        {
            case '+':
                result_ = data1_ + data2_;
                break;
            case '-':
                result_ = data1_ - data2_;
                break;
            case '*':
                result_ = data1_ * data2_;
                break;
            case '/':
                if(data2_ == 0) exitcode_ = DIVERROR;
                else result_ = data1_ / data2_;
                break;
            case '%':
                if(data2_ == 0) exitcode_ = MODERROR;
                else result_ = data1_ % data2_;
                break;
            default:
                exitcode_ = UNKNOWERRROR;
                break;
        }
    }

    //打印运算问题：计算数1 + 运算符 + 计算数2 
    std::string get_task()
    {
        std::string ret = std::to_string(data1_);
        ret += ' ';
        ret += op_;
        ret += ' ';
        ret += std::to_string(data2_);
        ret += ' ';
        ret += '=';
        ret += ' ';
        ret += '?';
        return ret;
    }   

    //打印运算结果：计算数1 + 运算符 + 计算数2 + 运算结果 + 退出码
    std::string result_to_string()
    {
        std::string ret = std::to_string(data1_);
        ret += ' ';
        ret += op_;
        ret += ' ';
        ret += std::to_string(data2_);
        ret += ' ';
        ret += '=';
        ret += ' ';
        ret += std::to_string(result_);
        ret += "[exitcode: ";
        ret += std::to_string(exitcode_);
        ret += ']';

        return ret;
    }

 
private:
    int data1_;   //计算数1
    int data2_;   //计算数2
    char op_;     //运算符
    int result_;  //运算结果
    int exitcode_;//线程退出码
};

ThreadPool.hpp

//线程池的实现
#pragma once
#include <pthread.h>
#include <vector>
#include <string>
#include <queue>
#include <unistd.h>
#include <unordered_map>

//定义线程的相关信息
struct ThreadInfo
{
    pthread_t tid_;    // 线程的TID
    std::string name_; // 线程名
};

//线程池
template <class T>
class ThreadPool
{
    static const int defaultnum = 5; //默认线程池中的线程数量
public:
    //加锁
    void Lock()
    {
        pthread_mutex_lock(&mutex_);
    }
    //解锁
    void Unlock()
    {
        pthread_mutex_unlock(&mutex_);
    }
    //唤醒
    void Weakup()
    {
        pthread_cond_signal(&cond_);
    }
    //挂起（休眠）
    void Sleep()
    {
        pthread_cond_wait(&cond_, &mutex_);
    }
    //对任务队列验空
    bool IsTaskQueueEmpty()
    {
        return tasks_.empty();
    }
    //获取任务（出队）
    T PopTasks()
    {
        T task = tasks_.front();
        tasks_.pop();
        return task;
    }
    //获取一个线程的线程名
    const std::string &GetThreadName(pthread_t tid)
    {
        return um_[tid];
    }

public:
    //构造初始化成员
    ThreadPool(int thread_num = defaultnum)
        :threads_(thread_num), thread_num_(thread_num)
    {
        pthread_mutex_init(&mutex_, nullptr);
        pthread_cond_init(&cond_, nullptr);    
    }

    //线程例程
    //pthread_create() 要求 Routine() 的返回类型必须是 void*，参数类型也必须是 void*。
    //由于非静态成员函数的第一个参数是隐藏的 this 指针，因此 Routine() 前不加 static，参数就不匹配
    //加上 static，由于静态成员函数中无法访问到非静态成员，因此还需将 this 指针原本所指的当前线程
    //作为 Routine() 的参数传递， 让 Routine() 可以去调用非静态的成员。
    static void *Routine(void *args) 
    {
        ThreadPool *tp = static_cast<ThreadPool*>(args);
        std::string name = tp->GetThreadName(pthread_self());
        while(true)
        {
            //加锁（任务队列是共享资源）
            tp->Lock();
            //任务队列非空才获取任务
            while(tp->IsTaskQueueEmpty())
            {
                tp->Sleep();
            }
            T task = tp->PopTasks();
            //解锁
            tp->Unlock();
            //处理任务
            task.run();
            //打印任务的处理结果
            printf("%s is running----%s\n", name.c_str(), task.result_to_string().c_str());
        }
    }

    //在线程池中批量创建线程
    void start()
    {
        for(int i = 0; i < thread_num_; i++)
        {
            threads_[i].name_ = "Thread-" + std::to_string(i);
            pthread_create(&(threads_[i].tid_), nullptr, Routine, this);//参数传入this指针
            um_[threads_[i].tid_] = threads_[i].name_;
        }
    }

    //将任务入队
    void push(const T& task)
    {
        Lock();
        tasks_.push(task);
        Weakup();
        Unlock();
    }

    //析构销毁互斥锁和条件变量
    ~ThreadPool()
    {
        pthread_mutex_destroy(&mutex_);
        pthread_cond_destroy(&cond_);
    }

private:
    std::vector<ThreadInfo> threads_; // 用一个vector来管理池中的多线程
    int thread_num_;                  // 线程池中的线程数量
    std::queue<T> tasks_;             // 线程间共享的任务队列
    pthread_mutex_t mutex_;           // 定义一把让所有线程保持互斥的锁
    pthread_cond_t cond_;             // 定义一个让所有线程保持同步的条件变量
    std::unordered_map<pthread_t, std::string> um_; // 用一个 unordered_map 快速检索一个线程的线程名
};

Cal.cc

// 程序主体
#include "ThreadPool.hpp"
#include "Task.hpp"
#include <unistd.h>
#include <iostream>

using namespace std;

const std::string opers = "+-*/%";

int main()
{
    srand((unsigned int)time(nullptr));
    ThreadPool<Task> *tp = new ThreadPool<Task>(5);
    tp->start();//在线程池中创建线程
    int len = opers.size();
    while(true)
    {
        //1.创建任务对象
        int data1 = rand() % 10 + 1; // 取值范围：[1, 10]
        usleep(10);

        int data2 = rand() % 13;     // 取值范围：[0, 13]
        usleep(10);

        char op = opers[rand() % len];

        Task task(data1, data2, op); //正式创建任务对象

        //2.将任务对象交给线程池处理
        printf("main thread push a task: %s\n", task.get_task().c_str());
        tp->push(task);

        usleep(100000);
    }

    return 0;
}

Makefile

Cal:Cal.cc
	g++ -o $@ $^ -std=c++11 -lpthread
.PHONY:clean
clean:
	rm -f Cal

五、线程安全的单例模式

单例模式是一种设计模式。设计模式是指，一套被反复使用、大多数人知晓的、经过分类整理的、代码设计的经验总结，可以提高代码可重用性，让代码更容易被他人理解，保证代码可靠性。设计模式可以使代码编写真正工程化，是软件工程的经络。

单例模式的作用是，可以保证系统中该类只有一个实例，并提供一个访问它的全局访问点，使该实例被所有程序模块共享。例如在某个服务器程序中，服务器的配置数据存放在一个文件，而这些配置数据由一个单例对象统一读取，服务进程中的其他对象可以通过这个单例对象来获取这些配置信息，这样就简化了在复杂环境下的配置管理。

【Tips】单例模式的实现要点：

因为全局只能有一个对象，所以需要将构造函数私有化；
用一个static静态指针（类的成员变量之一，在类外初始化）管理实例化的单例对象，并且提供一个静态成员函数，以获取这个static静态指针；
禁止拷贝，保证全局只有一个单例对象；
可以使用互斥锁来保证数据读取时的线程安全。

它具体又有两种实现方式——饿汉模式和懒汉模式。

1.饿汉和懒汉

饿汉模式是指，在程序启动时（即 main() 开始前）就实例化出单例对象，也可以形象地理解为，吃完饭立刻洗碗，保证下一顿饭可以直接拿碗开吃。

全局变量和静态变量，在 main() 开始前就已经被创建好了，而局部对象是在 main() 运行中创建的。由此，饿汉模式的实现大致为：在单例类 Singleton 中定义一个 T 类型的静态成员变量，并在类中提供获取这个静态成员变量的静态成员函数。无论创建多少个 Singleton 对象，最终都会只有一个 T 类型的静态成员变量有且仅有一个，且在 main() 开始前就已经被创建好了，后续可以直接使用。

//饿汉模式实现样例
template <typename T>
class Singleton 
{
	static T data; //定义一个 T 类型的静态成员变量
public:
	static T* GetInstance() //提供一个获取静态成员变量的静态成员函数
    {
		return &data;
	}
};

懒汉模式是指，单例对象在第一次被需要使用时才实例化，也可以形象地理解为，吃完饭先不洗碗，如果下一顿饭要用到这个碗就等下一顿饭再洗。

如果单例对象的构造十分耗时，或者会占用很多资源（例如加载插件、初始化网络连接、读取文件等），为了不影响程序的正常启动，可以使用懒汉模式（或称延迟加载）。

饿汉模式的实现大致为：在单例类 Singleton 中定义一个 T 类型的静态指针，并在类中提供能够创建 T 类型单例对象的静态成员函数。在 main() 开始前，并不会立即就创建出一个 T 类型的静态变量，而是等需要时，再调用 GetInstance() 去创建。

由于第一次调用 Getlnstance() 创建 T 类型的静态变量时，可能存在多个线程同时调用而可能会创建出多份实例，因此创建过程需加锁保护，且要加静态的锁。

template <typename T>
class Singleton 
{
	static T* inst; //定义一个 T 类型的静态指针
public:
    static T* GetInstance() //提供一个能够创建 T 类型静态变量的静态成员函数
    {
        pthread_mutex_lock(&mutex);
        if (inst == NULL) 
        {
        	inst = new T();
        }
        pthread_mutex_unlock(&mutex);

    	return inst;
    }

protected:
    static pthread_mutex_t mutex;//静态的互斥锁
};

2.基于懒汉模式的单例线程池

Task.hpp

//模拟实现的简易计算器（与上文一致）
#include <iostream>
#include <string>

//定义线程退出码
enum
{
    DIVERROR = 1, //除错误
    MODERROR,     //模错误
    UNKNOWERRROR  //未知错误
};

//简易计算器
class Task
{
public:
    //初始化成员变量
    Task(int a, int b, char op)
        :data1_(a), data2_(b), op_(op), result_(0), exitcode_(0)
    {}

    //完成实际的运算
    void run()
    {
        switch(op_)
        {
            case '+':
                result_ = data1_ + data2_;
                break;
            case '-':
                result_ = data1_ - data2_;
                break;
            case '*':
                result_ = data1_ * data2_;
                break;
            case '/':
                if(data2_ == 0) exitcode_ = DIVERROR;
                else result_ = data1_ / data2_;
                break;
            case '%':
                if(data2_ == 0) exitcode_ = MODERROR;
                else result_ = data1_ % data2_;
                break;
            default:
                exitcode_ = UNKNOWERRROR;
                break;
        }
    }

    //打印运算问题：计算数1 + 运算符 + 计算数2 
    std::string get_task()
    {
        std::string ret = std::to_string(data1_);
        ret += ' ';
        ret += op_;
        ret += ' ';
        ret += std::to_string(data2_);
        ret += ' ';
        ret += '=';
        ret += ' ';
        ret += '?';
        return ret;
    }   

    //打印运算结果：计算数1 + 运算符 + 计算数2 + 运算结果 + 退出码
    std::string result_to_string()
    {
        std::string ret = std::to_string(data1_);
        ret += ' ';
        ret += op_;
        ret += ' ';
        ret += std::to_string(data2_);
        ret += ' ';
        ret += '=';
        ret += ' ';
        ret += std::to_string(result_);
        ret += "[exitcode: ";
        ret += std::to_string(exitcode_);
        ret += ']';

        return ret;
    }

 
private:
    int data1_;   //计算数1
    int data2_;   //计算数2
    char op_;     //运算符
    int result_;  //运算结果
    int exitcode_;//线程退出码
};

ThreadPool.hpp

#pragma once
#include <pthread.h>
#include <vector>
#include <string>
#include <queue>
#include <unistd.h>
#include <unordered_map>

struct ThreadInfo
{
    pthread_t tid_;    
    std::string name_; 
};

template <class T>
class ThreadPool
{
    static const int defaultnum = 5;

public:
    void Lock()
    {
        pthread_mutex_lock(&mutex_);
    }

    void Unlock()
    {
        pthread_mutex_unlock(&mutex_);
    }

    void Weakup()
    {
        pthread_cond_signal(&cond_);
    }

    void Sleep()
    {
        pthread_cond_wait(&cond_, &mutex_);
    }

    bool IsTaskQueueEmpty()
    {
        return tasks_.empty();
    }

    T PopTasks()
    {
        T task = tasks_.front();
        tasks_.pop();
        return task;
    }

    const std::string &GetThreadName(pthread_t tid)
    {
        return um_[tid];
    }

public:
    static void *Routine(void *args)
    {
        ThreadPool *tp = static_cast<ThreadPool *>(args);
        std::string name = tp->GetThreadName(pthread_self());
        while (true)
        {
            tp->Lock();

            while (tp->IsTaskQueueEmpty())
            {
                tp->Sleep();
            }
            T task = tp->PopTasks();

            tp->Unlock();

            task.run(); 
            printf("%s is running----%s\n", name.c_str(), task.result_to_string().c_str());
        }
    }

    void start()
    {
        for (int i = 0; i < thread_num_; i++)
        {
            threads_[i].name_ = "Thread-" + std::to_string(i);
            pthread_create(&(threads_[i].tid_), nullptr, Routine, this);
            um_[threads_[i].tid_] = threads_[i].name_;
        }
    }

    void push(const T &task)
    {
        Lock();
        tasks_.push(task);
        Weakup();
        Unlock();
    }

    // 提供一个创建单例对象的静态接口
    static ThreadPool<T> *GetInstance() 
    {
        if (ptp_ == nullptr)
        {
            pthread_mutex_lock(&smutex_);
            if (ptp_ == nullptr)
            {
                printf("log: singleton creat done first!\n");
                ptp_ = new ThreadPool<T>();
            }
            pthread_mutex_unlock(&smutex_);
        }
        return ptp_;
    }

private:
    ThreadPool(int thread_num = defaultnum)
        : threads_(thread_num), thread_num_(thread_num)
    {
        pthread_mutex_init(&mutex_, nullptr);
        pthread_cond_init(&cond_, nullptr);
    }
    ~ThreadPool()
    {
        pthread_mutex_destroy(&mutex_);
        pthread_cond_destroy(&cond_);
    }
    ThreadPool(const ThreadPool<T> &tp) = delete;
    const ThreadPool<T> &operator=(const ThreadPool<T> &tp) = delete;

private:
    std::vector<ThreadInfo> threads_; // 用一个vector来管理池中的多线程
    int thread_num_;                  // 线程池中的线程数量
    std::queue<T> tasks_;             // 线程间共享的任务队列
    pthread_mutex_t mutex_;           // 定义一把让所有线程保持互斥的锁
    pthread_cond_t cond_;             // 定义一个让所有线程保持同步的条件变量
    std::unordered_map<pthread_t, std::string> um_; // 用一个 unordered_map 快速检索一个线程的线程名
    
    static ThreadPool<T> *ptp_;     // 静态指针
    static pthread_mutex_t smutex_; // 静态的互斥锁
};

//初始化静态指针和静态锁
template <class T>
ThreadPool<T> *ThreadPool<T>::ptp_ = nullptr;
template <class T>
pthread_mutex_t ThreadPool<T>::smutex_ = PTHREAD_MUTEX_INITIALIZER;

Cal.cc

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

using namespace std;

const std::string opers = "+-*/%";

int main()
{
    printf("main thread is start!...\n");
    sleep(3);
    srand((unsigned int)time(nullptr));
    // 获取一个单例对象，并创建一批线程
    ThreadPool<Task>::GetInstance()->start(); 
    int len = opers.size();
    while(true)
    {

        int data1 = rand() % 10 + 1; // [1, 10]
        usleep(10);
        int data2 = rand() % 13; // [0, 13]
        usleep(10);
        char op = opers[rand() % len];
        Task task(data1, data2, op);

        printf("main thread push a task: %s\n", task.get_task().c_str());
        ThreadPool<Task>::GetInstance()->push(task);
        usleep(1000000);
    }
    return 0;
}

Makefile

Cal:Cal.cc
	g++ -o $@ $^ -std=c++11 -lpthread
.PHONY:clean
clean:
	rm -f Cal