Linux下对线程的认识+生产消费者模型+信号量

线程的概念

线程是进程内部中更加轻量化的一种执行流。线程是CPU调度的基本单位，而进程是承担系统资源的实体。就是说一个进程中可能会有多个线程，而在Linux内核中并没有真正重新的创建线程并重新进行资源分配，因为我们每个线程指向的资源都是一样的，都在进程的地址空间空间中，所以Linux内部的线程创建本质就是创建PCB结构体（称作TCB），并指向同一个进程地址空间。可以说线程的创建其实就是进行资源的分配。

线程的理解

int gav=100;
void *ThreadRoutine(void *arg)
{
    const char *threadname = (const char *)arg;
    while (true)
    {
        cout << "I am a new thread: " << threadname<<" gav = "<<gav<<" &gav = "<<&gav<<endl;
        gav--;
        sleep(1);
    }
}
int main()
{
    // 已经有进程了
    pthread_t tid;
    pthread_create(&tid, nullptr, ThreadRoutine, (void *)"thread 1");

    // 主线程
    while (true)
    {
        cout << "I am main thread"<<" gav = "<<gav<<" &gav = "<<&gav<<endl;
        sleep(1);
    }
    return 0;
}

我们多个线程的全局数据是共享的，所以尽管创建了新线程，也是可以访问全局数据的。

我们的主线程LWP和PID的值是相等的，而LWP就是（light weight process）轻量级进程的缩写，我们CPU进行调度时看的就是LWP。其实g++编译线程相关程序的过程其实是需要带上库名的： -l pthread，其实原因也可以解释：

线程库

test.exe:test.cpp
	g++ -o $@ $^ -std=c++11 -l pthread
.PHONY:clean
clean:
	rm -f test.exe
run:
	./test.exe

我们发现当我们编写makefile文件时，在g++编译线程的程序时必须要用到-l pthread，其实就是因为我们的Linux内核中并没有创建线程的接口，只有轻量级进程的概念。（其实就是通过PCB代替了线程TCB）所以在当我们pthread_create创建线程时，本制就是在用户层和操作系统层之间封装一层软件层（pthread原生线程库），其中对上提供了线程的相关控制接口，对下就是通过调用相关轻量级进程的控制函数。

竟然这是一个库那么就好理解了：动态库和静态库的理解 Linux-CSDN博客

我们使用库文件是需要标明头文件的路径（-I），库文件的路径（-L），库名（-l），如果我们的头文件拷贝到/usr/include目录下，库文件拷贝到/lib64目录下的话就不需要指定各自路径，只需要指定库名就行，因为程序在编译的过程时会默认到以上路径下去找头文件和库文件，找不到才需要带上各自路径。

线程切换效率高

我们的线程指向的是同一个地址空间，同一张页表，所以各个线程中的固定资源都是一样的。CPU中存在着一个Cache缓存，这里面存放的就是程序所要执行的代码与数据，当执行当前代码时，进程上下文就会在Cache中找后续代码并执行，但是此过程可能会发生函数跳转，Cache失效会重行加载代码，但是根据局部性原理，大概率上下代码是连续执行的。所以我们的线程切换是不用切换Cache的。而且我们知道进程间是独立的，数据都是各自私有，所以线程切换相对于进程而言所切换的寄存器内容更少。

线程出现问题，进程就终止

int gav=100;
void *ThreadRoutine(void *arg)
{
    const char *threadname = (const char *)arg;
    while (true)
    {
        
        cout << "I am a new thread: " << threadname<<" gav = "<<gav<<" &gav = "<<&gav<<endl;
        gav--;
        sleep(1);
        int a=3;
        a/=0;
    }
}
int main()
{
    // 已经有进程了
    pthread_t tid_1,tid_2,tid_3;
    pthread_create(&tid_1, nullptr, ThreadRoutine, (void *)"thread 1");
    pthread_create(&tid_2, nullptr, ThreadRoutine, (void *)"thread 2");
    pthread_create(&tid_3, nullptr, ThreadRoutine, (void *)"thread 3");

    // 主线程
    while (true)
    {
        cout << "I am main thread"<<" gav = "<<gav<<" &gav = "<<&gav<<endl;
        sleep(1);
    }
    return 0;
}

其实就是因为线程中对每种信号的处理方式都是共享的，也就是handler表共享，所以一个线程崩溃的话，其他线程就会执行相同的处理方法。

线程终止与返回值

1. 从线程函数return。这种方法对主线程不适用,从main函数return相当于调用exit。

pthread_exit函数
功能：线程终止
原型：void pthread_exit(void *value_ptr);
参数：value_ptr:value_ptr，是线程退出的返回值，不要指向一个局部变量。
返回值：无返回值，跟进程一样，线程结束的时候无法返回到它的调用者（自身）

pthread_cancel函数
功能：取消一个执行中的线程，取消后该线程的返回值设为-1（PTHREAD_CANCELED）
原型：int pthread_cancel(pthread_t thread);
参数：thread：线程ID
返回值：成功返回0；失败返回错误码

线程等待：

pthread_join函数
功能：等待线程结束
原型
int pthread_join(pthread_t thread, void **value_ptr);
参数
thread:线程ID
value_ptr:它指向一个指针，后者指向线程的返回值（输出型参数）
返回值：成功返回0；失败返回错误码

void *ThreadRoutine(void *arg)
{
    const char *threadname = (const char *)arg;
    while (true)
    {
        cout << "I am a new thread: " << threadname << endl;
        sleep(1);
        break;
    }
    static char tmp[60]; // 局部变量出栈就会销毁，设为静态
    snprintf(tmp, sizeof(tmp), "i have done:%s", threadname);
    pthread_exit((void *)tmp);
}
int main()
{
    // 已经有进程了
    pthread_t tid_1;
    pthread_create(&tid_1, nullptr, ThreadRoutine, (void *)"thread 1");

    // 线程等待
    char *tmp = nullptr;
    int ret = pthread_join(tid_1, (void **)&tmp);
    if (ret != 0)
    {
        cout << "等待失败" << endl;
        return 1;
    }
    cout << "线程返回值：" << tmp << endl;

    return 0;
}

线程分离

线程默认是joinable的，也就是需要被等待的（回收资源），主线程在等待成功之前会一直进行阻塞。如果我们不关心线程返回值的话就可以将线程设置为分离状态，这样在线程退出之后，标准库会自动将线程回收，而不需要主线程join等待。（如果主线程依旧join等待的话就会等待失败）

int pthread_detach(pthread_t thread);
//可以指定线程分离也可以自行分离

线程原理（线程tid）

void *ThreadRoutine(void *arg)
{
    const char *threadname = (const char *)arg;
    while (true)
    {
        cout << "I am a new thread: " << threadname<<" my thread_id = "<<pthread_self()<<endl;
        sleep(1);
    }
}
int main()
{
    // 已经有进程了
    pthread_t tid_1,tid_2,tid_3;
    pthread_create(&tid_1, nullptr, ThreadRoutine, (void *)"thread 1");
    pthread_create(&tid_2, nullptr, ThreadRoutine, (void *)"thread 2");
    pthread_create(&tid_3, nullptr, ThreadRoutine, (void *)"thread 3");

    //线程等待---不等待也会造成僵尸问题
    pthread_join(tid_1,nullptr);
    pthread_join(tid_2,nullptr);
    pthread_join(tid_3,nullptr);
    return 0;
}

很容易的就发现了，线程id和LWP的是是不同的，但是它们都是标识当前线程的。初步分析：线程LWP和进程PID相近，而线程tid反倒有点像地址，而且tid在pthread_create函数中是一个输出型参数，会再函数内部将tid带出来。

将tid转成16进制：

char* Hex(pthread_t tid)
{
    static char tmp[30];
    snprintf(tmp,sizeof(tmp),"0X%x",tid);
    return tmp;
}

我们知道Linux是底层的系统中并没有线程的实现，而是只有轻量级进程，所以当我们程序运行时需要将pthread库加载进内存，然后映射到地址空间的共享区中。而我们的pthread库是需要对我们的线程进行管理的，也就是需要创建线程的相关数据属性（LWP、栈、上下文数据这些都是线程独立的资源）的结构体。而tid其实就是线程属性集合在库中的地址。因为是在库中维护的所以称为用户级线程

#include <sched.h>

int clone(int (*fn)(void *), void *child_stack,
                 int flags, void *arg, ...
                 /* pid_t *ptid, struct user_desc *tls, pid_t *ctid */ );
//pthread_create底层调用的就是该系统调用函数（fork调用的也是该函数）
第一个参数：新执行流所需要执行的方法
第二个参数：新执行流的栈空间地址

所以说我们的库会维护好我们新线程的栈空间，而主线程的栈是在地址空间的栈区里的。其实实际上创建的新线程的栈空间是在堆上new出来的。所以基于以上的认识就可以理解下图。

线程的局部存储

__thread 修饰的全局变量可以使得成为 线程的局部存储

也就是原来共享的全局变量，现在是每个线程各自独有一份相同变量名的数据。

线程的互斥与同步

线程互斥

临界资源：多线程执行流共享的资源就叫做临界资源
临界区：每个线程内部，访问临界资源的代码，就叫做临界区
互斥：任何时刻，互斥保证有且只有一个执行流进入临界区，访问临界资源，通常对临界资源起保护作用
原子性：不会被任何调度机制打断的操作，该操作只有两态，要么完成，要么未完成，可理解为一般只有一条汇编指令的代码

因为我们在多线程访问同一个全局数据（临界资源）的时候可能会造成数据资源非法访问的情况，所以我们提出了线程间的互斥概念。

举例多线程访问同资源：

int g_val = 1000;
void *ThreadRoutine(void *args)
{
    string name = static_cast<const char *>(args);
    while (1)
    {
        if (g_val > 0)
        {
            usleep(1000); //
            cout << "i am a new thread:" << name << " g_val = " << g_val << endl;
            g_val--;
        }
        else
            break;
    }
}
int main()
{
    // 已经有进程了
    pthread_t tid_1, tid_2, tid_3;
    pthread_create(&tid_1, nullptr, ThreadRoutine, (void *)"thread_1");
    pthread_create(&tid_2, nullptr, ThreadRoutine, (void *)"thread_2");
    pthread_create(&tid_3, nullptr, ThreadRoutine, (void *)"thread_3");

    // 线程等待

    pthread_join(tid_1, nullptr);
    pthread_join(tid_2, nullptr);
    pthread_join(tid_3, nullptr);

    return 0;
}

现象解释：

首先我们的g_val全局变量属于共享资源，所以我们在让多线程访问的时候必须要将其保护起来，也就是任何时刻只允许一个线程进行访问共享资源，否则会发生数据不一致问题。其实这和我们的CPU调度有关，线程是CPU调度的基本单位，而每个线程的时间片到了后就会将各自上下文存到PCB中，然后CPU调度下一个线程，调度下一个线程时会将下一个线程的进程上下文拷贝到CPU的寄存器当中。但是我们上一个被切换走的线程是可以在执行到任何代码段的时候被切换走的。

解释的话就是因为我们的g_vall--代码转成汇编指令并不是只有一条汇编代码而是三条，也就是说g_val--这不是原子性的：所以在CPU在执行g_vall--的过程中（假设当前g_val等于1），会先mov，将g_val的内容拷贝到eax寄存器中，然后将寄存器eax内的数据dec（也就是--操作），最后一步才将eax寄存器里的内容拷贝到全局变了g_val里。这其中只有在第三步才是真正做到g_val--的操作，所以在此之前只要CPU将线程切换的话g_val的值依旧是等于1，所以就会有多个线程同时在if语句内部进行--的操作就会导致数据访问异常。

实现加锁

在解决以上问题一般采用对临界区加锁，也就是任何时刻只允许一个执行流访问公共资源，也就是一个执行流加锁以后，其他执行流都会在加锁的函数处等待，只有等该执行流访问结束了以后，其他资源才能继续加锁访问临界资源。

初始化锁

 #include <pthread.h>

int pthread_mutex_destroy(pthread_mutex_t *mutex);//销毁锁
int pthread_mutex_init(pthread_mutex_t *restrict mutex,//局部锁
              const pthread_mutexattr_t *restrict attr);
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;//申请全局的锁

加锁解锁

#include <pthread.h>

int pthread_mutex_lock(pthread_mutex_t *mutex);//加锁
int pthread_mutex_trylock(pthread_mutex_t *mutex);//
int pthread_mutex_unlock(pthread_mutex_t *mutex);//解锁

使用锁

int g_val = 1000;
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
void *ThreadRoutine(void *args)
{
    string name = static_cast<const char *>(args);
    while (1)
    {
        pthread_mutex_lock(&mutex);//对临界区加锁
        if (g_val > 0)
        {
            usleep(1000); //
            cout << "i am a new thread:" << name << " g_val = " << g_val << endl;
            g_val--;
            pthread_mutex_unlock(&mutex);
        }
        else
        {
            pthread_mutex_unlock(&mutex);
            break;
        }
    }
}

申请锁是原子性

申请所的过程必须要是原子性的才能对临界区的资源进行保护，所以我们来对申请锁的汇编进行学习：

我们要知道al是我们CPU中的一个通用寄存器，xchgb是交换命令，mutex是锁结构中的一个数据。所以当我们的多线程竞争锁时，竞争成功的线程首先进来将数字0 move到al寄存器中，然后将al寄存器内的数据和mutex进行交换（而mutext的默认值都是1），所以此时al寄存器的内容就是1，而mutex数据的值就等于0，则进入if语句线程上锁成功，其他线程进来后也会从头执行相应的汇编代码，但是此时mutex的值还是0所以最终就会在else中挂起等待。虽然上锁的汇编代码不止一条，但是上锁的核心就是xchgb指令，所以无论线程在什么时候进行CPU的切换后，都不会有何影响，也就是说，哪个线程首次执行到xchgb汇编，哪个线程上锁成功。

对于解锁的过程就相当于是恢复mutex的默认值，也就是表明了各个线程可以重行竞争锁。

线程安全和可重入

线程安全：多个线程并发同一段代码时，不会出现不同的结果。常见对全局变量或者静态变量进行操作，并且没有锁保护的情况下，会出现该问题。线程安全描述的是线程。
重入：同一个函数被不同的执行流调用，当前一个流程还没有执行完，就有其他的执行流再次进入，我们称之为重入。一个函数在重入的情况下，运行结果不会出现任何不同或者任何问题，则该函数被称为可重入函数，否则，是不可重入函数。可重入描述的是函数。

死锁情况

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

线程同步

同步就是在保证数据安全的前提下，让线程能够按照某种特定的顺序访问临界资源，从而有效避免一个公共资源始终被同一个线程访问。

条件变量：

当一个线程互斥地访问某个变量时，它可能发现在其它线程改变状态之前，它什么也做不了（例如当内存为空时且生产者没生成数据时，消费者去消费也没用）这种情况就需要用到条件变量。也就是当生产者有数据时，向消费者发信号并唤醒消费者来消费。

初始化条件变量：
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;//生成全局条件变量
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);//依次唤醒每一个一个线程

使用条件变量


int ticket=1000;
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
void *my_stream(void *args)
{
    string name=static_cast<const char*>(args);
    while(1)
    {
        pthread_mutex_lock(&mutex);
        pthread_cond_wait(&cond,&mutex);//排队等待,均衡的让每个线程都能被分配到资源

        if(ticket>0)
            cout<<"i am a new thread, my name is "<<name<<",get a ticket:"<<ticket--<<endl;
        else 
            break;  
        pthread_mutex_unlock(&mutex);

    }
}
int main()
{

    pthread_t tid_1, tid_2, tid_3;
    pthread_create(&tid_1, nullptr, my_stream, (void*)"thread_1");
    pthread_create(&tid_2, nullptr, my_stream, (void*)"thread_2");
    pthread_create(&tid_3, nullptr, my_stream, (void*)"thread_3");

    while(ticket>0)
    {
        usleep(1000);
        // pthread_cond_signal(&cond);//唤醒等待队列的头一个线程
        pthread_cond_broadcast(&cond);//同时唤醒所有等待的线程，但是同一个线程不会被连续唤醒
    }     

    return 0;
}

多个线程同时在pthread_cond_wait(&cond,&mutex)下等待的时候会有以下情况：

线程在等待的过程中会释放锁资源(所以其他的线程也会执行到pthread_cond_wait函数处进行等待)其实是因为条件变量提供同步机制，也就是资源访问合理化，所以为了防止资源被一个线程访问完，所以线程等待时会释放锁，所以其他线程此时会排队的等在pthread_cond_wait处。而且释放锁是为生产者线程可以访问公共资源并向消费者发信号，避免死锁。
线程被唤醒后，由于是在临界区中，所以某个竞争成功的线程在执行pthread_cond_wait返回时会立即重新申请并持有锁(目的是防止共享资源被其他线程进入并破坏)

生产消费者模型

生产消费者模型就是：生产者（线程）向缓冲区中不断地生成数据。直到缓冲区满了。而消费者就是从缓冲区中取出数据并进行处理的线程或进程。消费者会不断从缓冲区中取出数据，缓冲区空了。

而生产者之间的关系是竞争（互斥），消费者之间也是竞争（互斥），生产者和消费者之间是（互斥与同步）
生产者和消费者不止一个，可能会是多个，所以也就是意味着多个线程或进程的访问
生产者和消费者间都是一个交易场所，也就是内存。

实现

实现生产消费者模型中的内存部分是通过实现一个阻塞队列实现的，也就是先传参固定阻塞队列可存储数据的个数，然后通过线程来充当生产者与消费者进行向阻塞队列中生产消费数据。并且需要保证阻塞队列满了就不能生产数据，空了就不能消费了。而且我们对于阻塞队列中的数据可以是任务也可以是讯息等等。我就实现任务的方式：

阻塞队列实现：

#pragma once

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

template <class T>
class block_queue
{
public:
    block_queue(int cap)
        : _capacity(cap)
    {
        pthread_mutex_init(&_mutex, nullptr);
        pthread_cond_init(&_pro_cond, nullptr);
        pthread_cond_init(&_cons_cond, nullptr);
    }

    bool is_full()
    {
        return _capacity == _q.size();
    }
    bool is_empty()
    {
        return _q.size() == 0;
    }

    void push(const T &data) // 生产者
    {
        pthread_mutex_lock(&_mutex); // 防止多执行流干扰结果--上锁(生产者之间是互斥关系)

        while (is_full()) // 数据满了就要进行等待
        {
            pthread_cond_wait(&_pro_cond, &_mutex);
            // 当信号方式是broadcast的话，就会唤醒在该代码处等待所有线程，多线程会竞争信号，而失败的线程依旧在该锁处等待（等待成功会自动上锁），竞争成功的线程执行解锁完毕后，
            // 其他的线程就会被唤醒（伪唤醒）向下执行，如果此时没有数据的话就会出问题,所以将if换成while更合适,每个线程都要再次判断一遍，只有非满才能出循环
        }
        _q.push(data); // 生产后就向消费者发信号
        pthread_cond_signal(&_cons_cond);

        pthread_mutex_unlock(&_mutex);
    }

    void pop(T *data) // 消费者
    {
        pthread_mutex_lock(&_mutex); // 防止多执行流干扰结果--上锁(消费者之间是互斥关系)

        while (is_empty()) // 没数据了要等待
        {
            pthread_cond_wait(&_cons_cond, &_mutex); // 同理
        }
        *data = _q.front();
        _q.pop(); // 消费后就向生产者发信号
        pthread_cond_signal(&_pro_cond);

        pthread_mutex_unlock(&_mutex);
    }

    ~block_queue()
    {
        pthread_mutex_destroy(&_mutex);
        pthread_cond_destroy(&_pro_cond);
        pthread_cond_destroy(&_cons_cond);
    }

private:
    queue<T> _q;
    int _capacity;
    pthread_mutex_t _mutex;//生产者和消费者也是互斥同步关系（访问着同一个阻塞队列），所以需要维护一把锁
    pthread_cond_t _pro_cond;//生产消费者各自等待的条件都不同，所以需要两个条件变量。
    pthread_cond_t _cons_cond;
};

任务：

#pragma once
#include <string>
#include <iostream>
using namespace std;

string opers = "+-*/%^~"; // 将操作符存在一起好生成随机操作符

enum // 枚举常量，标识结果码
{
    correct = 0,
    div_0,
    mol_0,
    none
};
class task
{
public:
    task() // 重载一个默认构造，好让消费者接受数据
    {
    }
    task(int x, int y, char ope)
        : _x(x), _y(y), _operat(ope), _result(0)
    {
    }
    void operator()() // 仿函数
    {
        run();
    }
    void run()
    {
        switch (_operat)
        {
        case '+':
            _result = _x + _y;
            break;
        case '-':
            _result = _x - _y;
            break;
        case '*':
            _result = _x * _y;
            break;
        case '/':
        {
            if (_y == 0)
                _code = div_0;
            _result = _x / _y;
        }
        break;
        case '%':
        {
            if (_y == 0)
                _code = mol_0;
            _result = _x % _y;
        }
        break;
        default:
            _code = none;
        }
    }
    void proc_print()
    {
        cout << "生产者：" << _x << ' ' << _operat << ' ' << _y << " = " << endl;
        ;
    }
    void consu_print()
    {
        cout << "消费者：" << _x << ' ' << _operat << ' ' << _y << " = " << _result << '[' << _code << ']' << endl;
        ;
    }

private:
    int _x;
    int _y;
    int _result;
    char _operat;
    int _code;
};

生产者消费者执行：

#include "block_queue.h"
#include "task.h"

class thread_data
{
    friend void *product(void *argv);
    friend void *consum(void *argv);

public:
    thread_data(block_queue<task> *bq, string name)
        : _bq(bq), _name(name)
    {}

private:
    block_queue<task> *_bq;
    string _name;
};

void *product(void *argv)
{

    thread_data *d = static_cast<thread_data *>(argv);
    while (1)
    {
        int x = rand() % 100;
        usleep(100); // 防止两个数字太接近
        int y = rand() % 100;
        char op = opers[rand() % opers.size()];
        task t(x, y, op);
        d->_bq->push(t); // 传任务
        cout << d->_name << "->";
        t.proc_print();
        sleep(1);
    }
}

void *consum(void *argv)
{
    thread_data *d = static_cast<thread_data *>(argv);
    while (1)
    {
        task t;
        d->_bq->pop(&t); // 接收任务
        // t.run();//处理任务
        t(); // 仿函数的方式来处理任务
        cout << d->_name << "->";
        t.consu_print();
    }
}

int main()
{
    srand((unsigned int)time(nullptr)); // 生成随机数种子（状态是全局的）
    pthread_t tid_1, tid_2, tid_3, tid_4, tid_5;
    block_queue<task> *bq = new block_queue<task>(5);
    thread_data d1(bq, "tid_1");
    thread_data d2(bq, "tid_2");
    thread_data d3(bq, "tid_3");
    thread_data d4(bq, "tid_4");
    thread_data d5(bq, "tid_5");

    pthread_create(&tid_1, nullptr, product, (void *)&d1);
    pthread_create(&tid_2, nullptr, product, (void *)&d2);
    pthread_create(&tid_3, nullptr, product, (void *)&d3);

    pthread_create(&tid_4, nullptr, consum, (void *)&d4);
    pthread_create(&tid_5, nullptr, consum, (void *)&d5);

    pthread_join(tid_1, nullptr);
    pthread_join(tid_2, nullptr);

    return 0;
}

生产消费者模型高效

生产消费者模型的高效其实并不在于生产数据和消费数据上，因为我们任意生产消费者的关系都是互斥的，也就是进行生产消费时只能有唯一的线程进入生产或消费，所以此过程并高效，而是保证了安全性。

真正的高效其实是因为：生产者和消费者线程可以并行执行，相互之间没有依赖关系。也就是当生产者进行生产数据的时候，多个消费者可能同时在处理数据，而当消费者消费接收数据时，多个生产者也可能是在接收数据来源。

生产者和消费者之间通过使用缓冲区来中介传递数据。缓冲区允许生产者生成多个资源，并且消费者可以以自己的速度消耗这些资源。这种缓冲区的使用可以平衡生产者和消费者之间的速度差异，从而提高整体性能。

POSIX信号量

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

信号量的出现其实就是很好的解决了共享资源访问的问题，也就是相较于互斥锁和条件变量的功能，信号量就相当于一把计数器，也就是标识着共享资源的资源个数，而信号量有着PV操作，P操作就是访问是否有资源，V操作就是发出资源，并且对应的信号量加一。

信号量的优势在于可以同一时刻允许进行多个线程访问公共资源，而且由于公共资源有信号量维护个数，所以每个线程进来访问各自的资源，互不干涉，分配不足的话则会进行资源等待。

接口认识

初始化信号量

#include <semaphore.h>
int sem_init(sem_t *sem, int pshared, unsigned int value);
参数：
pshared:0表示线程间共享，非零表示进程间共享
value：信号量初始值

销毁信号量

int sem_destroy(sem_t *sem);

等待信号量（P操作）

功能：等待并分配信号量，成功会将信号量的值减1，失败则等待
int sem_wait(sem_t *sem); //P()

发布信号量（V操作）

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

基于环形队列和信号量实现生产消费者模型

头文件：

#pragma once

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

#define default_value 3
template <class T>
class annular_queue
{
public:
    annular_queue(size_t n = default_value)
        : _v(5), _num(n), _proc(0), _cons(0)
    {
        sem_init(&_space, 0, n);
        sem_init(&_data, 0, 0);
        pthread_mutex_init(&_pmutex,nullptr);
        pthread_mutex_init(&_cmutex,nullptr);
    }
    void push(const T &x)
    {
        sem_wait(&_space); // P操作，分配信号量，成功则-1
        pthread_mutex_lock(&_pmutex);//先分配各个线程的信号量，再一个个访问，提高效率
        _v[_proc++] = x;
        _proc %= _num;
        sem_post(&_data); // V操作，数据++
        pthread_mutex_unlock(&_pmutex);
    }
    void pop(T *ret)
    {
        sem_wait(&_data); // P操作，分配信号量，成功则-1
        pthread_mutex_lock(&_cmutex);
        *ret = _v[_cons++];
        _cons %= _num;
        sem_post(&_space); // V操作，数据++
        pthread_mutex_unlock(&_cmutex);
    }
    ~annular_queue()
    {
        sem_destroy(&_space);
        sem_destroy(&_data);
    }

private:
    vector<T> _v;
    size_t _num;
    int _proc; // 生产者访问下标
    int _cons; // 消费者访问下标
    sem_t _space;
    sem_t _data;
    pthread_mutex_t _pmutex;
    pthread_mutex_t _cmutex;
};

主函数：

#include"add.h"

//信号量
pthread_mutex_t mutex=PTHREAD_MUTEX_INITIALIZER;
void* consum(void* args)
{

    annular_queue<int> *aq=static_cast<annular_queue<int> *>(args);
    int ret;
    while(1)
    {
        aq->pop(&ret);
        cout<<"get a data: "<<ret<<endl;
    }

}
void* produce(void* args)
{
    
    annular_queue<int> *aq=static_cast<annular_queue<int> *>(args);
    int k=100;//公共资源
    while(k)
    {
        pthread_mutex_lock(&mutex);
        aq->push(k);
        cout<<"produce a data: "<<k<<endl;
        k--;
        usleep(10000);
        pthread_mutex_unlock(&mutex);
    }
}
int main()
{
    pthread_t tid_1,tid_2,tid_3,tid_4,tid_5;
    annular_queue<int> *aq=new annular_queue<int> (5);
    pthread_create(&tid_1,nullptr,consum,aq);
    pthread_create(&tid_2,nullptr,consum,aq);
    pthread_create(&tid_3,nullptr,produce,aq);
    pthread_create(&tid_4,nullptr,produce,aq);
    pthread_create(&tid_5,nullptr,produce,aq);

    pthread_join(tid_1,nullptr);
    pthread_join(tid_2,nullptr);

    return 0;
}

我们需要知道该生产消费模式是基于循环队列的方式，主框架就是通过信号量维护循环队列，而生产者和消费者分别用各自的下标进行访问数据。

主要还是在于多生产者多消费者的理解，多个生产者进行访问公共资源时会竞争信号量，当多个线程同时调用sem_wait，如果信号量的值大于等于线程数，则所有线程都能成功进行P操作，并继续执行后续代码。如果信号量的值小于线程数，则只有部分线程能够进行P操作，其他线程会被阻塞。但是我们的代码是有问题的，尽管生产者资源分配合理了，但是我们访问资源的下标却只有一个，所以此时就无法达到互不干扰。所以采用互斥锁来维护公共资源，我们生产者和消费者各自都有访问资源下标，所以不影响之间的同步，所以只有生产者和生产者之间，消费者和消费者之间需要满足互斥，所以就分别各自设置一个锁资源达到互不干涉的效果。

而且还有一点就是申请锁和申请信号量的先后问题。其实我们在申请信号量成功之久就表明该线程有权利访问此资源，而申请锁是让线程之前互斥进行访问资源，所以谁先谁后都没问题。但是唯一一点就是效率的问题，所有线程先申请好信号量就不用每个线程锁上以后再一个个单独访问信号量了，而且临界区的代码越少越好，越少效率越高。