【Linux】线程同步与互斥 (生产者消费者模型)

🌈 个人主页：Zfox_
🔥 系列专栏：Linux

一：🔥 线程互斥

🦋 1-1 进程线程间的互斥相关背景概念

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

🦋 1-2 互斥量mutex

• ⼤部分情况，线程使⽤的数据都是局部变量，变量的地址空间在线程栈空间内，这种情况，变量归属单个线程，其他线程⽆法获得这种变量。
• 但有时候，很多变量都需要在线程间共享，这样的变量称为共享变量，可以通过数据的共享，完成线程之间的交互。
• 多个线程并发的操作共享变量，会带来⼀些问题。

// 操作共享变量会有问题的售票系统代码
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <pthread.h>

int ticket = 100;

void *route(void *arg)
{
	char *id = (char*)arg;
	while ( 1 ) {
		if ( ticket > 0 ) {
			usleep(1000);
			printf("%s sells ticket:%d\n", id, ticket);
			ticket--;
			} else {
			break;
		}
	}
} 

int main( void )
{
	pthread_t t1, t2, t3, t4;
	
	pthread_create(&t1, NULL, route, "thread 1");
	pthread_create(&t2, NULL, route, "thread 2");
	pthread_create(&t3, NULL, route, "thread 3");
	pthread_create(&t4, NULL, route, "thread 4");
	
	pthread_join(t1, NULL);
	pthread_join(t2, NULL);
	pthread_join(t3, NULL);
	pthread_join(t4, NULL);
} 

🥗 ⼀次执⾏结果：

thread 4 sells ticket:100
...
thread 4 sells ticket:1
thread 2 sells ticket:0
thread 1 sells ticket:-1
thread 3 sells ticket:-2

为什么可能⽆法获得争取结果？

• if 语句判断条件为真以后，代码可以并发的切换到其他线程
• usleep 这个模拟漫⻓业务的过程，在这个漫⻓的业务过程中，可能有很多个线程会进⼊该代码段
• –ticket 操作本⾝就不是⼀个原⼦操作

操作并不是原⼦操作，⽽是对应三条汇编指令：

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

要解决以上问题，需要做到三点：

• 代码必须要有互斥⾏为：当代码进⼊临界区执⾏时，不允许其他线程进⼊该临界区。
• 如果多个线程同时要求执⾏临界区的代码，并且临界区没有线程在执⾏，那么只能允许⼀个线程进⼊该临界区。
• 如果线程不在临界区中执⾏，那么该线程不能阻⽌其他线程进⼊临界区。要做到这三点，本质上就是需要⼀把锁。Linux上提供的这把锁叫互斥量。

🦋 互斥量的接⼝

🍡 初始化互斥量
初始化互斥量有两种⽅法：
⽅法1，静态分配:

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER

⽅法2，动态分配:

int pthread_mutex_init(pthread_mutex_t *restrict mutex, const
						pthread_mutexattr_t *restrict attr);

参数：
	mutex：要初始化的互斥量
	attr：NULL

🍡 销毁互斥量
销毁互斥量需要注意：

• 使⽤ PTHREAD_ MUTEX_ INITIALIZER 初始化的互斥量不需要销毁
• 不要销毁⼀个已经加锁的互斥量
• 已经销毁的互斥量，要确保后⾯不会有线程再尝试加锁

int pthread_mutex_destroy(pthread_mutex_t *mutex)；

🍡 互斥量加锁和解锁

int pthread_mutex_lock(pthread_mutex_t *mutex);
int pthread_mutex_unlock(pthread_mutex_t *mutex);
返回值:成功返回0,失败返回错误号

调⽤ pthread_ lock 时，可能会遇到以下情况:

• 互斥量处于未锁状态，该函数会将互斥量锁定，同时返回成功
• 发起函数调⽤时，其他线程已经锁定互斥量，或者存在其他线程同时申请互斥量，但没有竞争到互斥量，那么 pthread_ lock 调⽤会陷⼊阻塞(执⾏流被挂起)，等待互斥量解锁。

🍧 改进上⾯的售票系统:

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

int ticket = 100;
pthread_mutex_t mutex;

void *route(void *arg)
{
    char *id = (char *)arg;
    while (1)
    {
        pthread_mutex_lock(&mutex);
        if (ticket > 0)
        {
            usleep(1000);
            printf("%s sells ticket:%d\n", id, ticket);
            ticket--;
            pthread_mutex_unlock(&mutex);
            // sched_yield(); 放弃CPU
        }
        else
        {
            pthread_mutex_unlock(&mutex);
            break;
        }
    }
}

int main()
{
    pthread_t t1, t2, t3, t4;
    pthread_mutex_init(&mutex, NULL);
    pthread_create(&t1, NULL, route, (void*)"thread 1");
    pthread_create(&t2, NULL, route, (void*)"thread 2");
    pthread_create(&t3, NULL, route, (void*)"thread 3");
    pthread_create(&t4, NULL, route, (void*)"thread 4");

    pthread_join(t1, NULL);
    pthread_join(t2, NULL);
    pthread_join(t3, NULL);
    pthread_join(t4, NULL);

    pthread_mutex_destroy(&mutex);
    return 0;
}

🦋 1-3 互斥量实现原理探究

• 经过上⾯的例⼦，⼤家已经意识到单纯的 i++ 或者 ++i 都不是原⼦的，有可能会有数据⼀致性问题
• 为了实现互斥锁操作, ⼤多数体系结构都提供了 swap 或 exchange 指令, 该指令的作⽤是把寄存器和内存单元的数据相交换, 由于只有⼀条指令, 保证了原⼦性, 即使是多处理器平台, 访问内存的总线周期也有先后, ⼀个处理器上的交换指令执⾏时另⼀个处理器的交换指令只能等待总线周期。 现在我们把 lock 和 unlock 的伪代码改⼀下 。
  

🦋 1-4 互斥量的封装

Mutex.hpp

#pragma once

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

namespace MutexModule
{
    class Mutex
    {
    public:
        Mutex(const Mutex&) = delete;
        const Mutex& operator = (const Mutex&) = delete;
        Mutex()
        {
            int n = ::pthread_mutex_init(&_lock, nullptr);
            (void)n;
        }

        void Lock()
        {
            int n = ::pthread_mutex_lock(&_lock);
            (void)n;
        }

        void Unlock()
        {
            int n = ::pthread_mutex_unlock(&_lock);
            (void)n;
        }

        ~Mutex()
        {
            int n = ::pthread_mutex_destroy(&_lock);
            (void)n;
        }
    private:
        pthread_mutex_t _lock;
    };

    class LockGuard
    {
    public:
        LockGuard(Mutex &mtx)
            :_mtx(mtx)
        {
            _mtx.Lock();
        }
        ~LockGuard()
        {
            _mtx.Unlock();
        }
    private:
        Mutex &_mtx;
    };
}

🍧 抢票的代码就可以更新成为

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <pthread.h>
#include "Mutex.hpp"

using namespace MutexModule;
int ticket = 1000;

Mutex mutex;
void *route(void *arg)
{
    char *id = (char *)arg;
    while (1)
    {
        LockGuard lockguard(mutex); // 使⽤RAII⻛格的锁
        if (ticket > 0)
        {
            usleep(1000);
            printf("%s sells ticket:%d\n", id, ticket);
            ticket--;
        }
        else
        {
            break;
        }
    }
    return nullptr;
}

int main()
{
    pthread_t t1, t2, t3, t4;

    pthread_create(&t1, NULL, route, (void *)"thread 1");
    pthread_create(&t2, NULL, route, (void *)"thread 2");
    pthread_create(&t3, NULL, route, (void *)"thread 3");
    pthread_create(&t4, NULL, route, (void *)"thread 4");

    pthread_join(t1, NULL);
    pthread_join(t2, NULL);
    pthread_join(t3, NULL);
    pthread_join(t4, NULL);

    return 0;
}

RAII⻛格的互斥锁， C++11也有，⽐如： 

std::mutex mtx;
std::lock_guard<std::mutex> guard(mtx);

此处我们仅做封装，⽅便后续使⽤，详情⻅C++博客 

二：🔥 线程同步

🦋 2-1 条件变量

• 🍥 当⼀个线程互斥地访问某个变量时，它可能发现在其它线程改变状态之前，它什么也做不了。
• 🍥 例如⼀个线程访问队列时，发现队列为空，它只能等待，只到其它线程将⼀个节点添加到队列中。这种情况就需要⽤到条件变量。

🦋 2-2 同步概念与竞态条件

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

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

🦋 2-3 条件变量函数

🌯 初始化

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

参数：
	cond：要初始化的条件变量
	attr：NULL

🌯 销毁

int pthread_cond_destroy(pthread_cond_t *cond)

🌯 等待条件满⾜

int pthread_cond_wait(pthread_cond_t *restrict cond,pthread_mutex_t *restrict
						mutex);

参数：
	cond：要在这个条件变量上等待
	mutex：互斥量，后⾯详细解释

🌯 唤醒等待

int pthread_cond_broadcast(pthread_cond_t *cond);
int pthread_cond_signal(pthread_cond_t *cond);

简单案例：

• 我们先使⽤ PTHREAD_COND / MUTEX_INITIALIZER 进⾏测试，对其他细节暂不追究
• 然后将接⼝更改成为使⽤ pthread_cond_init / pthread_cond_destroy 的⽅式，⽅便后续进⾏封装

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

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;


void *active(void *args)
{
    std::string name = static_cast<const char*>(args);

    while(true)
    {
        pthread_mutex_lock(&mutex);

        // 没有对资源是否就绪的判定
        pthread_cond_wait(&cond, &mutex);
        printf("%s is active!\n", name.c_str());

        pthread_mutex_unlock(&mutex);
    }

    return nullptr;
}

int main()
{
    pthread_t tid1, tid2, tid3;
    pthread_create(&tid1, nullptr, active, (void*)"thread-1");
    pthread_create(&tid1, nullptr, active, (void*)"thread-2");
    pthread_create(&tid1, nullptr, active, (void*)"thread-3");

    sleep(1);
    printf("Main thread ctrl begin...\n");

    while(true)
    {
        printf("main wakeup thread...\n");
        pthread_cond_signal(&cond);
        sleep(1);
    }



    pthread_join(tid1, nullptr);
    pthread_join(tid2, nullptr);
    pthread_join(tid3, nullptr);

    return 0;
}

🧋运行结果：

Main thread ctrl begin...
main wakeup thread...
thread-1 is active!
main wakeup thread...
thread-2 is active!
main wakeup thread...
thread-3 is active!

三：🔥 ⽣产者消费者模型

• 321原则(便于记忆) 三种关系 两个角色 一个消费场所（某种数据结构组织的连续的内存空间）

🥙 生产者-消费者模型（Producer-Consumer Model）是一种经典的多线程同步问题，它描述了两个线程（或进程）之间的协作：一个或多个生产者线程生成数据项，并将它们放入缓冲区中；一个或多个消费者线程从缓冲区中取出数据项，并进行处理。这个模型通常用于解决生产者和消费者在不同速度下工作时的同步和数据传输问题。

🦋 3-1 为何要使⽤⽣产者消费者模型

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

🦋 3-2 ⽣产者消费者模型优点

• 🌶️ 解耦
• 🌶️ 支持并发
• 🌶️ 支持忙闲不均
  

🦋 3-3 基于BlockingQueue的⽣产者消费者模型

🍱 3-3-1 BlockingQueue

在多线程编程中阻塞队列 (Blocking Queue) 是⼀种常⽤于实现⽣产者和消费者模型的数据结构。其与普通的队列区别在于，当队列为空时，从队列获取元素的操作将会被阻塞，直到队列中被放⼊了元素；当队列满时，往队列⾥存放元素的操作也会被阻塞，直到有元素被从队列中取出 (以上的操作都是基于不同的线程来说的，线程在对阻塞队列进程操作时会被阻塞)

🍱 3-3-2 C++ queue模拟阻塞队列的⽣产消费模型

代码：

• 为了便于理解，我们以单⽣产者，单消费者，来进⾏讲解。
• 刚开始写，我们采⽤原始接⼝。
• 我们先写单⽣产，单消费。然后改成多⽣产，多消费（这⾥代码其实不变，这里用到了更后面的cond的封装头文件）。

BlockQueue.hpp

#pragma

#include <iostream>
#include <queue>
#include <pthread.h>
#include "Mutex.hpp"
#include "Cond.hpp"

namespace BlockQueueModule
{
    using namespace LockModule;
    using namespace CondModule;

    // version2
    static const int gcap = 10;

    template<typename T>
    class BlockQueue
    {
    public:
        BlockQueue(int cap = gcap)
            :_cap(cap)
            ,_cwait_num(0)
            ,_pwait_num(0)
        {}

        bool IsFull() { return _q.size() == _cap; }

        bool IsEmpty() { return _q.empty(); }

        void Equeue(const T &in)   // 生产者
        {
            LockGuard lockguard(_mutex);

            // 生产数据有条件
            // 结论1：在临界区中等待是必然的（当前）
            while(IsFull())  // 为了防止伪唤醒 使用while判断
            {
                std::cout << "生产者进入等待..." << std::endl;

                // 2. 等待 释放锁
                _pwait_num++;
                _productor_cond.Wait(_mutex);     // wait的时候，必定是持有锁的
                _pwait_num--;
                // 3. 返回，线程被唤醒 重新申请并持有锁
                std::cout << "生产者被唤醒..." << std::endl;
            }
            // 4. isfull不满足 || 线程被唤醒 
            _q.push(in);    // 生产

            // 肯定有数据
            if(_cwait_num)
            {
                std::cout << "叫醒消费者" << std::endl;
                _consumer_cond.Notify();
            }
        }

        void Pop(T* out)  // 消费者
        {
            LockGuard lockguard(_mutex);

            while(IsEmpty())
            {
                std::cout << "消费者进入等待..." << std::endl;
                _cwait_num++;
                _consumer_cond.Wait(_mutex);   // 伪唤醒
                _cwait_num--;
                std::cout << "消费者被唤醒..." << std::endl;
            }
            // 4. 线程被唤醒
            *out = _q.front();
            _q.pop();

            // 一定不为满
            if(_pwait_num)
            {
                std::cout << "叫醒生产者" << std::endl;
                _productor_cond.Notify();
            }
        }

        ~BlockQueue()
        {}
    private:
        std::queue<T> _q;                   // 临界资源
        Mutex _mutex;             // 互斥
        Cond _productor_cond;     // 生产者条件变量
        Cond _consumer_cond;      // 消费者条件变量
        int _cap;                           // bq最大容量

        int _cwait_num;
        int _pwait_num;
    };
}

main.cc

#include <functional>
#include "BlockQueue.hpp"
#include "Task.hpp"
#include <unistd.h>

using namespace BlockQueueModule;
using namespace TaskModule;

using task_t = std::function<void()>;

void *Comsumer(void *args)
{
    BlockQueue<int> *bq = static_cast<BlockQueue<int> *>(args);

    int data = 10;
    while(true)
    {
        // 1. 从bq中拿到数据
        bq->Pop(&data);

        // 2. 做处理
        printf("Comsumer 消费了一个数据：%d\n", data);
        data++;
    }
}

void *Productor(void *args)
{
    BlockQueue<int> *bq = static_cast<BlockQueue<int> *>(args);

    // 1. 从外部获取数据
    int data = 10;
    while(true)
    {
        sleep(2);
        // 2. 生产到队列中
        printf("Productor 生产了一个数据：%d\n", data);
        bq->Equeue(data);

        data++;
    }
}

int main()
{
    BlockQueue<int> *bq = new BlockQueue<int>(5);     // 共享资源 -> 临界资源
    // 单生产 单消费
    pthread_t c1, c2, p1, p2, p3;
    pthread_create(&c1, nullptr, Comsumer, (void*)bq);
    pthread_create(&p3, nullptr, Productor, (void*)bq);


    pthread_join(c1, nullptr);
    pthread_join(p3, nullptr);

    delete bq;

    return 0;
}

main.cc

#include "BlockQueue.hpp"
#include <unistd.h>

using namespace BlockQueueModule;

void *Comsumer(void *args)
{
    BlockQueue<int> *bq = static_cast<BlockQueue<int> *>(args);

    int data = 10;
    while(true)
    {
        // 1. 从bq中拿到数据
        bq->Pop(&data);

        // 2. 做处理
        printf("Comsumer 消费了一个数据：%d\n", data);
        data++;
    }
}

void *Productor(void *args)
{
    BlockQueue<int> *bq = static_cast<BlockQueue<int> *>(args);

    // 1. 从外部获取数据
    int data = 10;
    while(true)
    {
        sleep(2);
        // 2. 生产到队列中
        printf("Productor 生产了一个数据：%d\n", data);
        bq->Equeue(data);

        data++;
    }
}

int main()
{
    BlockQueue<int> *bq = new BlockQueue<int>(5);     // 共享资源 -> 临界资源
    // 单生产 单消费
    pthread_t c1, c2, p1, p2, p3;
    pthread_create(&c1, nullptr, Comsumer, (void*)bq);
    pthread_create(&p3, nullptr, Productor, (void*)bq);


    pthread_join(c1, nullptr);
    pthread_join(p3, nullptr);

    delete bq;

    return 0;
}

输出结果：

root@hcss-ecs-a9ee:~/code/linux/112/lesson31/2.BlockQueue# ./bq 
消费者进入等待...
Productor 生产了一个数据：10
叫醒消费者
消费者被唤醒...
Comsumer 消费了一个数据：10
消费者进入等待...
Productor 生产了一个数据：11
叫醒消费者
消费者被唤醒...
Comsumer 消费了一个数据：11

四：🔥 为什么 pthread_cond_wait 需要互斥量?

• 条件等待是线程间同步的⼀种⼿段，如果只有⼀个线程，条件不满⾜，⼀直等下去都不会满⾜，所以必须要有⼀个线程通过某些操作，改变共享变量，使原先不满⾜的条件变得满⾜，并且友好的通知等待在条件变量上的线程。
• 条件不会⽆缘⽆故的突然变得满⾜了，必然会牵扯到共享数据的变化。所以⼀定要⽤互斥锁来保护。没有互斥锁就⽆法安全的获取和修改共享数据。
  
• 按照上⾯的说法，我们设计出如下的代码：先上锁，发现条件不满⾜，解锁，然后等待在条件变量上不就⾏了，如下代码: 错误的设计

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 。所以解锁和等待必须是⼀个原⼦操作。 （这就是为什么wait的时候需要把条件变量和锁一起传进去）
• int pthread_cond_wait(pthread_cond_ t *cond,pthread_mutex_ t *mutex); 进⼊该函数后，会去看条件量等于0不？等于，就把互斥量变成1，直到 cond_ wait 返回，把条件量改成1，把互斥量恢复成原样。

🦋 4-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);

🦋 4-2 条件变量的封装

基于上⾯的基本认识，我们已经知道条件变量如何使⽤，虽然细节需要后⾯再来进⾏解释，但这⾥可以做⼀下基本的封装，以备后⽤.

Cond.hpp

#pragma once

#include <iostream>
#include <pthread.h>
#include "Mutex.hpp"

namespace CondModule
{
    using namespace LockModule;

    class Cond
    {
    public:
        Cond()
        {
            int n = ::pthread_cond_init(&_cond, nullptr);
            (void)n;
        }
        
        void Wait(Mutex &mutex) // 让我们的线程释放曾经持有的锁！
        {
            int n = ::pthread_cond_wait(&_cond, mutex.LockPtr());
        }

        void Notify()
        {
            int n = ::pthread_cond_signal(&_cond);
            (void)n;
        }

        void NotifyAll()
        {
            int n = ::pthread_cond_broadcast(&_cond);
            (void)n;
        }

        ~Cond()
        {
            int n = ::pthread_cond_destroy(&_cond);
        }
    private:
        pthread_cond_t _cond;
    };
}

五：🔥 POSIX信号量

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

🍲 初始化信号量

#include <semaphore.h>
int sem_init(sem_t *sem, int pshared, unsigned int value);

参数：
	pshared:0表⽰线程间共享，⾮零表⽰进程间共享
	value：信号量初始值

🍲 销毁信号量

int sem_destroy(sem_t *sem);

🍲 等待信号量

功能：等待信号量，会将信号量的值减1   为0将进行阻塞等待
int sem_wait(sem_t *sem); //P()

🍲 发布信号量

功能：发布信号量，表⽰资源使⽤完毕，可以归还资源了。将信号量值加1。
int sem_post(sem_t *sem);//V()

上⼀节⽣产者-消费者的例⼦是基于queue的,其空间可以动态分配,现在基于固定⼤⼩的环形队列重写这个程序（POSIX信号量）:

🦋 基于环形队列的⽣产消费模型

• 环形队列采⽤数组模拟，⽤模运算来模拟环状特性
  

• 环形结构起始状态和结束状态都是⼀样的，不好判断为空或者为满，所以可以通过加计数器或者标记位来判断满或者空。另外也可以预留⼀个空的位置，作为满的状态
  

但是我们现在有信号量这个计数器，就很简单的进⾏多线程间的同步过程。
// 随⼿做⼀下封装
Sem.hpp

#pragma

#include <iostream>
#include <semaphore.h>

namespace SemModule
{
    const int defaultsemval = 1;
    class Sem
    {
    public:
        Sem(int value = defaultsemval)
            :_init_value(value)
        {
            int n = ::sem_init(&_sem, 0, value);
            (void)n;
        }

        void P()
        {
            int n = ::sem_wait(&_sem);
            (void)n;
        }

        void V()
        {
            int n = ::sem_post(&_sem);
            (void)n;
        }

        ~Sem()
        {
            int n = ::sem_destroy(&_sem);
            (void)n;
        }
    private:
        sem_t _sem;
        int _init_value;
    };
}

RingBuffer.hpp

#pragma

#include <iostream>
#include <vector>
#include <semaphore.h>
#include <pthread.h>
#include "Sem.hpp"
#include "Mutex.hpp"

namespace RingBufferModule
{
    using namespace SemModule;
    using namespace LockModule;

    template<typename T>
    class RingBuffer
    {
    public:
        RingBuffer(int cap)
            :_ring(cap)
            ,_cap(cap)
            ,_p_step(0)
            ,_c_step(0)
            ,_datasem(0)
            ,_spacesem(cap)
        {
        }

        void Equeue(const T& in)
        {
            // 生产者
            // pthread_mutex_lock(&_p_lock);

            _spacesem.P();

            LockGuard lockguard(_p_lock);    // 放这里更好 申请信号量和申请锁是并行执行了 
            _ring[_p_step] = in;     // 生产完毕
            _p_step++;
            _p_step %= _cap;

            _datasem.V();
        }

        void Pop(T* out)
        {
            // 消费者
            // pthread_mutex_lock(&_c_lock);

            _datasem.P();

            LockGuard lockguard(_c_lock);
            *out = _ring[_c_step];
            _c_step++;
            _c_step %= _cap;

            _spacesem.V();
        }

        ~RingBuffer() 
        {
        }
    private:
        std::vector<T> _ring;   // 环，临界资源
        int _cap;               // 总容量
        int _p_step;            // 生产者位置
        int _c_step;            // 消费位置

        Sem _datasem;           // 数据信号量
        Sem _spacesem;          // 空间信号量

        Mutex _p_lock;
        Mutex _c_lock;
    };
}

main.cc

#include <functional>
#include "RingBuffer.hpp"
#include <unistd.h>
#include <pthread.h>

using namespace RingBufferModule;

void *Comsumer(void *args)
{
    RingBuffer<int> *ring_buffer = static_cast<RingBuffer<int> *>(args);
    while(true)
    {
        sleep(1);
        // 消费数据
        int data;
        ring_buffer->Pop(&data);
        
        // 处理数据：花时间
        std::cout << "消费了一个数据: " << data << std::endl;
    }
    
}

void *Productor(void *args)
{
    RingBuffer<int> *ring_buffer = static_cast<RingBuffer<int> *>(args);
    int data = 0;
    while(true)
    {
        // 获取数据: 花时间

        // 生产数据
        ring_buffer->Equeue(data);
        data++;
        std::cout << "生产了一个数据: " << data << std::endl;
    }
    
}

int main()
{
    RingBuffer<int> *ring_buffer = new RingBuffer<int>(5);

    pthread_t c1, c2, c3, p1, p2;
    pthread_create(&c1, nullptr, Comsumer, ring_buffer);
    pthread_create(&c2, nullptr, Comsumer, ring_buffer);
    pthread_create(&c3, nullptr, Comsumer, ring_buffer);
    pthread_create(&p1, nullptr, Productor, ring_buffer);
    pthread_create(&p2, nullptr, Productor, ring_buffer);

    pthread_join(c1, nullptr);
    pthread_join(c2, nullptr);
    pthread_join(c3, nullptr);
    pthread_join(p1, nullptr);
    pthread_join(p2, nullptr);

    delete ring_buffer;
    return 0;
}

🧁 运行结果：

生产了一个数据: 生产了一个数据: 11
生产了一个数据: 2
生产了一个数据: 3
生产了一个数据: 4

消费了一个数据: 0
生产了一个数据: 5
生产了一个数据: 6
消费了一个数据: 1
消费了一个数据: 0
生产了一个数据: 2
消费了一个数据: 2生产了一个数据: 7

六：🔥 C++同步互斥代码练习

leetcode 1114. 按序打印
答案：

#include <semaphore.h>

class Foo {
public:
    sem_t firstsem;
    sem_t secondsem;

    Foo() {
        sem_init(&firstsem, 0, 0);
        sem_init(&secondsem, 0, 0);
    }

    void first(function<void()> printFirst) {
        
        // printFirst() outputs "first". Do not change or remove this line.
        printFirst();
        sem_post(&firstsem);
    }

    void second(function<void()> printSecond) {
        
        // printSecond() outputs "second". Do not change or remove this line.
        sem_wait(&firstsem);
        printSecond();
        sem_post(&secondsem);
    }

    void third(function<void()> printThird) {
        
        // printThird() outputs "third". Do not change or remove this line.
        sem_wait(&secondsem);
        printThird();
    }
};

leetcode 1117. H2O 生成

#include <condition_variable>
#include <mutex>

class H2O {
public:
    std::mutex mtx;
    std::condition_variable cond_hyd, cond_oxy;
    int hyd = 0;
    H2O() {
        
    }

    void hydrogen(function<void()> releaseHydrogen) {
        
        // releaseHydrogen() outputs "H". Do not change or remove this line.
        std::unique_lock<std::mutex> lock(mtx);
        cond_hyd.wait(lock, [this]{ return hyd < 2; });

        hyd++;
        releaseHydrogen();
        if(hyd == 2)
        {
            cond_oxy.notify_one();
        }
    }

    void oxygen(function<void()> releaseOxygen) {
        std::unique_lock<std::mutex> lock(mtx);
        // releaseOxygen() outputs "O". Do not change or remove this line.
        cond_oxy.wait(lock, [this]{ return hyd == 2; });
        releaseOxygen();
        hyd = 0;
        cond_hyd.notify_all();
    }
};

七：🔥 共勉

以上就是我对 【Linux】线程同步与互斥 的理解，觉得这篇博客对你有帮助的，可以点赞收藏关注支持一波~😉