🌈前言
🌸1、Linux线程同步
- 🍨1.1、同步概念与竞态条件
- 🍧1.2、条件变量
🌺2、条件变量相关API
- 🍨2.1、初始化和销毁条件变量
- 🍧2.2、阻塞等待条件满足
- 🎃2.3、唤醒阻塞等待的条件变量
- 🎂2.4、为什么 pthread_cond_wait 需要互斥锁?⭐⭐⭐
🍀3、生产者消费者模型
- 🍨3.1、概念
- 🍧3.2、基于BlockingQueue的生产者消费者模型
- 🎃3.3、阻塞队列的实现

🌈前言

这篇文章给大家带来线程同步与互斥的学习！！！

🌸1、Linux线程同步

🍨1.1、同步概念与竞态条件

首先抛出一个问题：线程互斥，它是对的，但是它合理（任何场景）吗？？？答：不一定合理

举个例子⭐⭐⭐

我们去食堂打饭，食堂打饭的规则是竞争式的抢饭（不用排队）

力气大的人会优先去抢到饭（男生 --优先级高的线程），力气小的就会一直抢不到饭（女生 – 优先级小的线程）

这种规则没有错，确实食堂阿姨一次只能给一个人打饭，但是不合理，会造成弱小的人的饥饿问题（迟迟没有吃到饭）！！！

在多线程竞争锁来看，优先级高的线程会一直优先申请到锁资源，而优先级低的线程会长时间得不到对应的资源，会造成多执行流下的饥饿问题！！！

互斥下的饥饿问题：多线程下的某个执行流，长时间得不到某种资源

同步概念⭐⭐⭐

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

竞态条件：因为时序(CPU调度)问题，而导致程序异常，我们称之为竞态条件。在线程场景下，这种问题也不难理解

同步与互斥是互帮互助的，互斥是解决线程安全问题，而同步是解决合理性问题

🍧1.2、条件变量

我们已经直到同步是什么了，那么如何实现同步与互斥呢？答：条件变量

概念：⭐⭐⭐

当一个线程互斥地访问临界资源时，需要另一个线程对临界资源的状态做改变，就需要条件变量了

例如：一个线程访问队列时，发现队列为空，它只能等待，直到其它线程将一个节点添加到队列中。这种情况就需要用到条件变量

当条件(由程序员设置)满足时，条件变量就会阻塞等待执行该函数的线程，并且释放锁，随后下一个线程就会申请到锁继续判断

🌺2、条件变量相关API

🍨2.1、初始化和销毁条件变量

初始化条件变量有二种方法

第一种方法：静态分配

#include <pthread.h>
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;

代码解析

pthread_cond_t是条件变量，它是一个联合体，里面有一个结构体描述条件变量的属性

PTHREAD_COND_INITIALIZER：它是一个宏，用于初始化条件变量

注意：静态分配不用释放条件变量

第二个方法：动态分配

#include <pthread.h>
int pthread_cond_init(pthread_cond_t *restrict cond, 
							const pthread_condattr_t *restrictattr);

函数解析

cond：要初始化的条件变量（pthread_mutex_t变量的地址）

restrictattr：设置条件变量的属性，一般为NULL/nullptr

返回值：初始化成功返回0，失败返回一个错误码errno

注意：动态分配需要释放条件变量

销毁条件变量：

#include <pthread.h>
int pthread_cond_destroy(pthread_cond_t *cond)

函数解析

cond：要销毁的条件变量（pthread_cond_t变量的地址）

返回值：初始化成功返回0，失败返回一个错误码errno

销毁条件变量需要注意

使用 PTHREAD_ COND_ INITIALIZER 初始化的条件变量不需要销毁

使用 pthread_cond_init初始化的条件变量，需要进行销毁

🍧2.2、阻塞等待条件满足

#include <pthread.h>
int pthread_cond_wait(pthread_cond_t *restrict cond, 
						 pthread_mutex_t *restrict mutex);

函数解析

cond：阻塞要在这个条件变量上等待的线程（pthread_cond_t变量的地址）

mutex：互斥锁，同步需要与互斥锁绑定使用，因为阻塞等待时，会释放该线程的锁，后面被唤醒时，会重新获取锁，后面代码感受

返回值：成功完成后，返回零值；否则，返回错误编号(errno)以指示错误

🎃2.3、唤醒阻塞等待的条件变量

#include <pthread.h>
int pthread_cond_broadcast(pthread_cond_t *cond);
int pthread_cond_signal(pthread_cond_t *cond);

函数解析

cond：唤醒在条件变量上等待的线程（pthread_cond_t变量的地址）

pthread_cond_broadcast：唤醒被阻塞等待的全部线程

pthread_cond_signal：唤醒被阻塞等待的一个线程，按顺序唤醒

返回值：如果成功，pthread_cond_broadcast()和pthread-cond_signal()函数返回零；否则应返回一个错误编号(errno)以指示错误

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

// 定义条件变量
pthread_cond_t pc;
// pthread_cond_t pc = PTHREAD_COND_INITIALIZER;   // 静态初始化
// 定义互斥锁
pthread_mutex_t pm;
// pthread_mutex_t pm = PTHREAD_MUTEX_INITIALIZER; // 静态初始化

// 定义全局退出变量 -- volatile保持内存可见性
volatile bool quit = false;

// 定义一个方法集合
typedef void (*Func)();
vector<Func> handler;

void Print()
{
    cout << "hello Print" << endl;
}

// 线程执行函数
void *CallThread(void *args)
{
    // 线程分离，线程退出后自动释放资源，主线程不用等待
    pthread_detach(pthread_self());
    string name = static_cast<const char *>(args);
    while (!quit)
    {
        // 执行这个函数，说明特定的条件变量没有就绪，执行该函数的线程会被阻塞挂起等待
        pthread_cond_wait(&pc, &pm);
        cout << name << ": " << pthread_self() << ", run..." << endl;
        // 执行方法集合的全部方法
        for (auto &f : handler)
        {
            f();
        }
    }
    return nullptr;     
}

int main()
{
    // 加载函数方法 -- 第二个是lambda表达式（底层是仿函数operator()）
    handler.push_back(Print);
    handler.push_back([]()
                      { cout << "hello lambda" << endl; }
                      );
    // 初始化条件变量和互斥锁
    pthread_mutex_init(&pm, nullptr);
    pthread_cond_init(&pc, nullptr);
    // 线程创建
    pthread_t t1, t2, t3;
    pthread_create(&t1, nullptr, CallThread, (void *)"Thread1");
    pthread_create(&t2, nullptr, CallThread, (void *)"Thread2");
    pthread_create(&t3, nullptr, CallThread, (void *)"Thread3");

    while (true)
    {
        char ch = '\0';
        cout << "请输入一个字符y/n: ";
        cin >> ch;
        if (ch == 'y')
        {
            //  该函数可以让条件变量就绪，唤醒单个被阻塞挂起线程
            pthread_cond_signal(&pc);
            sleep(1);
        }
        else
        {
            quit = true;
            // 唤醒所有被阻塞挂的线程
            pthread_cond_broadcast(&pc);
            break;
        }
    }
    // 释放条件变量和互斥锁
    pthread_cond_destroy(&pc);
    pthread_mutex_destroy(&pm);
    return 0;
}

从运行结果图可以看出线程是按顺序轮询执行的…

在这里插入图片描述

为什么输入n后，唤醒全部线程后，最后不是打印三次方法集合的信息呢？

因为全部线程被唤醒后，又会重新去竞争锁(条件变量需要重新申请锁)，只有一个线程可以竞争成功，没有竞争成功的线，会重新判断!quit，!quit为false，退出执行函数
这个线程执行完方法集合发现!quit为false不能进入循环，就退出了，届时，全部线程就已经退出了，只有一个线程执行了方法集合！！！

🎂2.4、为什么 pthread_cond_wait 需要互斥锁?⭐⭐⭐

条件与条件变量

条件：对应的共享资源的状态，(比如抢票，票数小于0，就不能抢了)，通过判断的方式，来判断对应的资源是否符合要求

条件变量：在条件满足或不满足的前提下，进行wait(等待) 或 signal(唤醒)的一种方式

结论

条件等待是线程间同步的一种手段，如果只有一个线程，条件不满足，一直等下去都不会满足

所以必须要有一个线程通过某些操作，改变共享变量，使原先不满足的条件变得满足，并且友好的通知(唤醒)等待在条件变量上的线程

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

在这里插入图片描述

按照上面的说法，我们设计出如下的代码：先上锁，发现条件不满足，解锁，然后等待在条件变量上不就行了，如下代码

// 错误的设计
pthread_mutex_lock(&mutex);
while (condition_is_false) 
{
	pthread_mutex_unlock(&mutex);
	// 解锁之后，等待之前，条件可能已经满足，信号已经发出，但是该信号可能被错过 -- 发生线程切换
	pthread_cond_wait(&cond);
	pthread_mutex_lock(&mutex);
} 
pthread_mutex_unlock(&mutex);

结论

由于解锁和等待不是原子操作。调用解锁之后， pthread_cond_wait 之前，如果已经有其他线程获取到互斥量，摒弃条件满足，发送了信号，那么 pthread_cond_wait 将错过这个信号，可能会导致线程永远阻塞在这个 pthread_cond_wait 。所以解锁和等待必须是一个原子操作

int pthread_cond_wait(pthread_cond_ t *cond,pthread_mutex_ t * mutex); 进入该函数后，会去看条件量等于0不？等于，就把互斥量变成1，直到cond_ wait返回，把条件量改成1，把互斥量恢复成原样 – 原子操作

条件变量使用规范

等待条件代码

pthread_mutex_lock(&mutex);
while (条件为假）
pthread_cond_wait(cond, mutex);
修改条件
pthread_mutex_unlock(&mutex);

给条件发送信号代码

pthread_mutex_lock(&mutex);
设置条件为真
pthread_cond_signal(cond);
pthread_mutex_unlock(&mutex);

🍀3、生产者消费者模型

🍨3.1、概念

概念

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

在这里插入图片描述

如果还是不理解的话，请看下面的例子

消费者线程有多个，消费者之间是什么关系呢？

它们之间是竞争的关系（商品少的时候，就要竞争） – 互斥

供应商线程有多个，供应商之间是什么关系呢？

它们之间是竞争的关系（供应商要竞争超市的架子，摆上自己的商品） – 互斥

消费者和供应商之间又是什么关系呢？

它们之间是同步与互斥的关系，因为消费者去购买商品时，可能出现缺货的情况，这时候就要通知供应商供给商品了，反之，生产者供给商品后，也要通知消费者来买…

总结：⭐⭐⭐⭐⭐

321原则：3种关系（消费者与消费者之间的关系，生产者和生产者之间的关系，消费者和生产者之间的关系），2种角色（消费者和生产者），一个交易场所（超市）

生产者消费者模型优点：解耦、支持并发、支持忙闲不均

🍧3.2、基于BlockingQueue的生产者消费者模型

BlockingQueue：

在多线程编程中阻塞队列(Blocking Queue)是一种常用于实现生产者和消费者模型的数据结构
其与普通的队列区别在于，当队列为空时，从队列获取元素的操作将会被阻塞，直到队列中被放入了元素（这里要设置条件判空，设置条件变量）
当队列满时，往队列里存放元素的操作也会被阻塞，直到有元素被从队列中取出(以上的操作都是基于不同的线程来说的，线程在对阻塞队列进程操作时会被阻塞)

在这里插入图片描述

🎃3.3、阻塞队列的实现

实现方式：

主要问题：消费者和生产者的三种关系要控制好
唤醒各自线程的时机，不要乱加锁，可能造成死锁或一直等待的问题

blockqueue.hpp

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

const int DfCap = 5; // 默认容量大小

namespace Mybq
{
    template <typename T>
    class BlockQueue
    {
    public:
        BlockQueue(uint32_t cap = DfCap)
            : _cap(cap), _bq()
        {
            pthread_mutex_init(&mutex, nullptr);
            pthread_cond_init(&conCond, nullptr);
            pthread_cond_init(&proCond, nullptr);
        }

        // 生产者生产
        void push(const T &val)
        {
            // 加锁 -- 保护临界资源(队列)
            lockQueue();
            // 循环判断阻塞队列是否为满 -- 防止伪唤醒 -> proPendwait调用失败、多线程竞争锁或系统原因等等...
            while (isFull())
            {
                // 等待的时候，自动释放mutex锁
                proPendwait(); // 阻塞等待，等待被唤醒
                // 等待完后，是在临界区醒的，重新申请mutex锁
            }
            // 加载资源 -> 解锁
            _bq.push(val);
            unlockQueue();
            // 生产者唤醒消费者，因为生产者已经把资源放到队列里面了(条件变量就绪)
            weakupJCon();
        }
        
        // 消费者消费
        T pop()
        {
            lockQueue();
            // 防止伪唤醒
            while (isEmpty())
            {
                // 阻塞等待，等待被唤醒
                conPendWait();
            }
            // 删除资源->解锁
            T val = _bq.front();
            _bq.pop();
            unlockQueue();
            // 消费者唤醒生产者，因为队列的资源没有了(条件变量就绪)
            weakupPro();
            return val;
        }

        ~BlockQueue()
        {
            pthread_mutex_destroy(&mutex);
            pthread_cond_destroy(&conCond);
            pthread_cond_destroy(&proCond);
        }

    private:
        // 判断阻塞队列是否为空，是否为满
        bool isFull()
        {
            return _bq.size() == _cap;
        }

        bool isEmpty()
        {
            return _bq.empty();
        }
        //-------------------------------------------

        // 互斥锁加锁解锁
        void lockQueue()
        {
            pthread_mutex_lock(&mutex);
        }

        void unlockQueue()
        {
            pthread_mutex_unlock(&mutex);
        }

        //--------------------------------------------

        // 在条件变量下阻塞等待，等待被唤醒
        void proPendwait()
        {
            // 1. 条件变量在阻塞等待线程的时候，会自动释放mutex互斥锁！！！
            // 释放锁：因为阻塞等待的线程占用着锁，其他线程不能申请 -- 没有锁了意味着不能访问临界资源
            pthread_cond_wait(&proCond, &mutex);
            // 2. 当阻塞结束(唤醒)，返回时，pthread_cond_wait，会自动帮你重新获取mutex锁，最后才返回
        }

        void conPendWait()
        {
            pthread_cond_wait(&conCond, &mutex);
        }

        //--------------------------------------------

        // 条件变量就绪，被唤醒阻塞等待的程序
        void weakupPro()
        {
            // 消费者唤醒生产者，因为队列的资源没有了(消费者调用)
            pthread_cond_signal(&proCond);
        }

        void weakupJCon()
        {
            // 生产者唤醒消费者，因为生产者已经把资源放到队列里面了(生产者调用)
            pthread_cond_signal(&conCond);
        }

    private:
        uint32_t _cap; // 阻塞队列容量大小
        queue<T> _bq;
        pthread_mutex_t mutex;  // 阻塞队列互斥锁
        pthread_cond_t conCond; // 消费者条件变量
        pthread_cond_t proCond; // 生产者条件变量
    };
}

test.cpp

#include "blockqueue.hpp"
#include <ctime>
#include <cstdlib>

// 生产者线程执行函数
void *CallProducer(void *args)
{
    Mybq::BlockQueue<int>* bqp = static_cast< Mybq::BlockQueue<int>*>(args);
    while (true)
    {
    // 生产数据 -- [0, 100]
    int data = rand() % 100 + 1;
    bqp->push(data);
    cout << "Producer data sucess, data: " << data << endl;
    sleep(2);
    }
    return nullptr;
}

// 消费者线程执行函数
void *Callconsumer(void *args)
{
    Mybq::BlockQueue<int>* bqp = static_cast< Mybq::BlockQueue<int>*>(args);
    while (true)
    {
    // 消费数据
    int data = bqp->pop();
    cout << "Consumer data sucess, data: " << data << endl;
    }
    return nullptr;
}

int main()
{
    srand((unsigned int)time(nullptr)); // 随机数种子
    Mybq::BlockQueue<int> bq;
    // 多线程测试
    pthread_t producer; // 生产者线程
    pthread_t consumer; // 消费者线程
                        // &bq将阻塞队列的地址传给线程执行函数的参数
    pthread_create(&producer, nullptr, CallProducer, &bq);
    pthread_create(&consumer, nullptr, Callconsumer, &bq);
    // 等待线程
    pthread_join(producer, nullptr);
    pthread_join(consumer, nullptr);
    return 0;
}