【Linux】线程同步和生产者-消费者模型

一. 线程同步

线程同步: 在保证数据安全的前提下, 使线程能够按照某种特定的顺序访问临界资源,避免饥饿问题;

例:
当去掉休眠后, 1 号线程由于先运行, 竞争锁的能力比较强, 直接抢占了大部分的资源;

#include "Thread.hpp"

pthread_mutex_t mtx = PTHREAD_MUTEX_INITIALIZER;
int tickets = 50;

void* func(Thread<int*>* td, int* tickets)
{
    while (1)
    {
        //sleep(1);   // 休眠, 避免一个线程直接抢完
        LockGuard guard(&mtx);  // 申请加锁, 离开作用域自动解锁
        if (*tickets > 0)
            cout << "次线程: " << td->get_name() << ",  " << (*tickets)-- << endl;
        else
            break;
    }
    return 0;
}

int main()
{
    int n = 10;

    vector<Thread<int*> > threads;
    for (int i=1; i<n; i++)
        threads.emplace_back(func, &tickets, "thread-"+to_string(i));

    for (int i=1; i<n; i++)
        threads[i-1].start();

    for (int i=1; i<n; i++)
        threads[i-1].join();
    cout << "---------" << endl;
    cout << tickets << endl;

    return 0;
}

线程运行是没有问题的, 但是不符合期望, 在原生线程库中提供了条件变量这种方式来实现线程同步;

1. 条件变量

条件变量相当于一个队列, 若线程不满足运行条件, 那么推入当前条件变量的队列当中, 等待唤醒; 当其他线程唤醒时, 从当前条件变量的队列中推出线程; 通常条件变量和互斥锁同时使用;

条件变量与互斥锁不同, 互斥锁是线程自动竞争锁资源, 而条件变量是诱发的;

2. 条件变量接口

条件变量的创建及初始化

条件变量的类型为 pthread_cond_t, 在创建后需要进行初始化;

#include <pthread.h>

 // 定义条件变量
pthread_cond_t cond;

// 全局/静态的条件变量初始化
cond = PTHREAD_COND_INITIALIZER;

// 条件变量初始化
int pthread_cond_init(pthread_cond_t *cond, pthread_condattr_t *cond_attr);

参数:

• cond: 需要条件变量的指针;
• cond_attr: 初始化时的相关属性, 设置为 nullptr 表示使用默认属性;

返回值:

• 若成功, 返回 0; 若失败, 返回 error number;

全局/静态的条件变量和互斥锁相同, 也可以使用静态初始化, 自动初始化, 自动销毁;

条件变量的销毁

#include <pthread.h>

int pthread_cond_destroy(pthread_cond_t *cond);

参数:

• cond: 条件变量的地址;

返回值:

• 若成功, 返回 0; 若失败, 返回 error number;

条件变量等待

pthread_cond_wait() 函数, 将等待当前线程, 并且释放当前线程申请的锁资源(避免当前锁资源出现死锁, 唤醒时自动竞争锁资源);
pthread_cond_timedwait() 函数, 和 pthread_cond_wait() 函数相同, 不过会限制等待时间, 超时自动唤醒, 避免死锁;

#include <pthread.h>

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

int pthread_cond_timedwait(pthread_cond_t *cond, pthread_mutex_t *mutex, const struct timespec *abstime);

参数:

• cond: 条件变量的地址;
• mutex: 互斥锁的地址;
• abstime: 指定等待的时间(其值为系统时间 + 等待时间);

返回值:

• 若成功, 返回 0; 若失败, 返回 error number;

条件变量唤醒

pthread_cond_signalt() 函数, 唤醒指定条件变量等待的队头线程;

#include <pthread.h>

int pthread_cond_signal(pthread_cond_t *cond);

参数:

• cond: 条件变量的地址;

返回值:

• 若成功, 返回 0; 若失败, 返回 error number;

pthread_cond_broadcast() 函数, 唤醒指定条件变量等待的所有线程;

#include <pthread.h>

int pthread_cond_broadcast(pthread_cond_t *cond);

参数:

• cond: 条件变量的地址;

返回值:

• 若成功, 返回 0; 若失败, 返回 error number;

3. 条件变量同步解决抢占问题

将线程推入同一条件变量队列中, 一个一个的唤醒, 这样就保证了资源分配均匀;

• Thread.hpp

#pragma once
#include <iostream>
#include <pthread.h>
#include <unistd.h>
#include <functional>
#include <vector>

using namespace std;


class LockGuard
{
    public:
    LockGuard(pthread_mutex_t* mutex)
    :_mutex(mutex)
    {
        pthread_mutex_lock(_mutex);
    }

    ~LockGuard()
    {
        pthread_mutex_unlock(_mutex);
    }

    private:
    pthread_mutex_t* _mutex;
};


template<class T>
class Thread
{
    typedef function<void*(Thread<T>*, T&)> func_t;

public:
    Thread(func_t func = nullptr, const T& args = T(), const string& name = "none")
        :_tid(0), _func(func), _args(args), _name(name)
    {}

    static void* threadroutine(void* td)
    {
        auto it = static_cast<Thread<T>*>(td);
        it->_func(it, it->_args);
        return 0;
    }

    bool start()
    {
        int flag = pthread_create(&_tid, nullptr, threadroutine, this);
        if (flag)
        {
            _tid = 0;
            return false;
        }
        return true;
    }

    void join()
    {
        if (_tid)
        {
            void* msg;
            int flag = pthread_join(_tid, &msg);
            if (flag)
            {
                cerr << _name << "  join fail "<< endl;
                exit(1);
            }
        }
        _tid = 0;
    }

    ~Thread()
    {
        if (_tid)
            join();
    }

    const string& get_name()
    { return _name; }

private:
    pthread_t _tid;
    func_t _func;
    string _name;
    T _args;
};

• test.cpp

#include "Thread.hpp"

pthread_mutex_t mtx = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
int tickets = 50;

void* func(Thread<int*>* td, int* tickets)
{
    LockGuard guard(&mtx);  // 申请加锁, 离开作用域自动解锁
    while (*tickets)
    {
        cout << "次线程: " << td->get_name() << ",  " << (*tickets)-- << endl;
        pthread_cond_wait(&cond, &mtx);	// 线程等待, 自动释放锁资源, 下一个线程此时获取锁资源
        pthread_cond_signal(&cond);		// 唤醒下一个线程
    }
    return 0;
}

void wait(vector<Thread<int*> >& threads, int n)
{
    for (int i=0; i<n; i++)
        threads[i].join();
}

void start(vector<Thread<int*> >& threads, int n)
{
    for (int i=0; i<n; i++)
        threads[i].start();
    
    sleep(2);
    pthread_cond_signal(&cond);
}

void init(vector<Thread<int*> >& threads, int n)
{
    for (int i=0; i<n; i++)
        threads.emplace_back(func, &tickets, "thread-"+to_string(i+1));
}

int main()
{
    int n = 10;
    vector<Thread<int*> > threads;
    init(threads, n);
    start(threads, n);
    wait(threads, n);

    cout << "---------" << endl;
    cout << tickets << endl;

    return 0;
}

二. 生产者-消费者模型

1. 什么是生产者-消费者模型

假设有两个线程 A, B 和一个缓冲区; A 线程向缓冲区中写入数据, B 线程从缓冲区中读取数据, 这就是一个简单的生产者-消费者模型, A 为生产者, B 为消费者;

2. 为什么要使用生产者-消费者模型

• 解耦
  假设生产者和消费者分别是两个类; 若使生产者直接调用消费者的某个方法, 那么生产者对于消费者就会产生依赖(也就是耦合); 若消费者的代码改变, 就可能会影响到生产者; 而若两者不直接调用或通信, 两者之间也就不会直接依赖, 耦合也就相应降低了;

• 支持并发
  假设生产者直接调用消费者的某个方法, 由于函数调用是同步的, 那么生产者就需要等待消费者处理方法; 而在生产者-消费者模型中, 两者为并发的线程/进程, 只需要关心缓冲区的状态, 在缓冲区 非空&&非满 的情况下, 不会互相影响;

• 支持忙闲不均

3. 生产者-消费者模型特点

生产者和消费者的三种关系:

• 生产者与消费者的关系:
  同步: 当缓冲区满的时候, 生产者会进入休眠状态, 当下次消费者开始消耗缓冲区的数据时, 生产者才会被唤醒, 开始往缓冲区中添加数据; 当缓冲区空的时候, 消费者也会进入休眠状态, 直到生产者往缓冲区中添加数据时才会被唤醒;
  互斥: 同一时间, 生产者或消费者只能有一方向缓冲区添加或消耗数据;

• 生产者与生产者的关系: 互斥;

• 消费者与消费者的关系: 互斥;

4. 基于阻塞队列实现生产者-消费者模型

阻塞队列(Blocking Queue)是一种常用于实现生产者-消费者模型的数据结构;

阻塞队列的特定: 阻塞队列的是有容量的;

使用阻塞队列实现的生产者-消费者模型类似管道; 其同步和互斥特性使用条件变量和互斥锁实现;

单生产-单消费

根据生产者-消费者模型特点, 搭建出所需的框架;

#include "Thread.hpp"

template<class T>
class BlockingQueue
{
public:
    BlockingQueue(int cap = 10)
        :_cap(cap)
    {
        // 初始化锁和条件变量
        pthread_mutex_init(&_mutex);
        pthread_cond_init(&_product_cond);
        pthread_cond_init(&_consum_cond);
    }

    void Push(const T& data)
    { }

    const T& Pop()
    { }

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

    bool IsEmpty()
    { return _blcok_queue.size() == 0; }


    ~BlockingQueue()
    {
        // 销毁锁和条件变量
        pthread_mutex_destroy(&_mutex);
        pthread_cond_destroy(&_product_cond);
        pthread_cond_destroy(&_consum_cond);
    }


private:
    queue<T> _blcok_queue;
    int _cap;

    pthread_mutex_t _mutex;  // 消费者之间的互斥锁
    pthread_cond_t _product_cond;   // 生产者的条件变量
    pthread_cond_t _consum_cond;    // 消费者的条件变量
};

当生产者生产数据时, 也就是插入数据时, 条件为是否存在空间, 若没有, 就需要阻塞等待; 若有空间, 那么直接插入数据, 并且需要通知消费者插入了数据;

void Push(const T &in)
{
	pthread_mutex_lock(&_mutex); // 申请加锁
	
	if (IsFull()) // 插入判满
		pthread_cond_wait(&_product_cond, &_mutex);
	
	_blcok_queue.push(in); // 插入数据
	
	pthread_cond_signal(&_consum_cond);    // 唤醒消费者
	pthread_mutex_unlock(&_mutex); // 解锁
}

消费者同理

void Pop(T &out)
{
     pthread_mutex_lock(&_mutex);

     if (IsEmpty()) // 删除判空
         pthread_cond_wait(&_consum_cond, &_mutex);

     out = _blcok_queue.front();
     _blcok_queue.pop(); // 插入数据

     pthread_cond_signal(&_product_cond);    // 唤醒生产者
     pthread_mutex_unlock(&_mutex);
 }

创建测试, 生产者先运行 2 秒, 然后消费者开始消费;

#include "BlockingQueue.hpp"

template<class T>
void* Product(Thread<T>* self, T& args)
{
    BlockingQueue<int>* blcok_queue = (BlockingQueue<int>*)args;
    int n = 1;
    while (1)
    {
        cout << self->get_name() << " " << n << endl;
        cout << "---------" << endl;
        blcok_queue->Push(n++);
        sleep(1);
    }
    return 0;
}

template<class T>
void* Consum(Thread<T>* self, T& args)
{
    int n;
    BlockingQueue<int>* blcok_queue = (BlockingQueue<int>*)args;
    while (1)
    {
        blcok_queue->Pop(n);
        cout << self->get_name() << " " << n << endl;
        cout << "---------" << endl;
        sleep(1);
    }
    return 0;
}


template<class T>
void Wait(vector<Thread<T> >& threads, int n)
{
    for (int i=0; i<n; i++)
        threads[i].join();
}

template<class T>
void Start(vector<Thread<T> >& threads, int n)
{
    for (int i=0; i<n; i++)
        threads[i].start();
    
    //sleep(2);
}

template<class T>
void Init(vector<Thread<T> >& threads, int n, T func(Thread<T>*,T&), void* data = nullptr, const char* name = "thread-")
{
    for (int i=0; i<n; i++)
        threads.emplace_back(func, data, name+to_string(i+1));
}

int main()
{
    int n = 1;
    int m = 1;
    // vector<Thread<void*> > threads;

    BlockingQueue<int> blcok_queue;

    vector<Thread<void*> > product;
    vector<Thread<void*> > consum;
    Init(product, n, Product, &blcok_queue, "product-");
    Init(consum, m, Consum, &blcok_queue, "consum-");
    Start(product, n);
    Start(consum, m);
    Wait(product, n);
    Wait(consum, m);

  
    cout << "---------" << endl;

    return 0;
}

多生产-多消费

在单生产-单消费模型中, 插入和删除数据的条件判断使用的 if 判断, 那么当 pthread_cond_wait() 函数被唤醒时, 就会直接向下执行; 这种判断在实际上是有误的, 因为 pthread_cond_wait() 函数可能存在调用失败(误唤醒, 伪唤醒)的情况;

而在 多生产-多消费 模型中, pthread_cond_wait() 函数调用失败 或 调用pthread_cond_broadcast() 函数导致非法唤醒的情况更多, 所以需要将条件判断的 if 改为 while, 持续进行条件判断, 不合法的唤醒需要重新堵塞等待;

void Push(const T &in)
{
    pthread_mutex_lock(&_mutex); // 申请加锁

    while (IsFull()) // 插入判满
        pthread_cond_wait(&_product_cond, &_mutex);

    _blcok_queue.push(in); // 插入数据

    pthread_cond_signal(&_consum_cond);    // 唤醒消费者
    pthread_mutex_unlock(&_mutex); // 解锁
}

void Pop(T &out)
{
    pthread_mutex_lock(&_mutex);

    while (IsEmpty()) // 删除判空
        pthread_cond_wait(&_consum_cond, &_mutex);

    out = _blcok_queue.front();
    _blcok_queue.pop(); // 插入数据

    pthread_cond_signal(&_product_cond);    // 唤醒生产者
    pthread_mutex_unlock(&_mutex);
}

4. POSIX 信号量

信号量的本质就是一个计数器; 信号量的 PV 操作是原子的, 可以使用信号量实现l临界资源的互斥和同步;

信号量主要作用是描述临界资源中的资源数目;

• 若申请信号量成功, 计数器 - - (P 操作)
• 若释放信号量成功, 计数器 ++ (V 操作)

若将临界资源看作一个整体, 这种信号量就是二元信号量, 类似互斥锁, 信号量只有两种状态: 0, 1;

若将临界资源中的资源数目划分为多份, 这种信号量就是多元信号量, 类似条件变量, 只有申请资源成功的, 才可以进行临界区操作;

信号量的初始化及销毁

#include <semaphore.h>

// 创建信号量
sem_t sem;	

// 初始化信号量
int sem_init(sem_t *sem, int pshared, unsigned int value);

// 销毁信号量
int sem_destroy(sem_t *sem);

参数:

• sem: 指定的信号量;
• pshared: 当前信号量的共享状态. 传递 0 表示线程间共享, 传递 非0 表示进程间共享;
• value: 信号量的初始值, 相当于资源的数目;

返回值: 若成功返回 0; 若失败返回 -1, 并设置错误码;

信号量的申请及释放

#include <semaphore.h>

// 申请信号量
int sem_wait(sem_t *sem);		// 堵塞等待直至申请成功
int sem_trywait(sem_t *sem);	// 只会申请一次
int sem_timedwait(sem_t *sem, const struct timespec *abs_timeout);	// 若失败, 等待 abs_timeout 后再次申请;

// 释放信号量
int sem_post(sem_t *sem);

参数:

• sem: 指定的信号量;
• abs_timeout: 休眠时间, 若申请失败, 会在 abs_timeout 后再次申请;

返回值: 若成功返回 0; 若失败返回 -1, 并设置错误码;

5. 基于环形队列实现生产者-消费者模型

生产者-消费者模型的缓冲区也可以使用环形队列进行实现;

在环形队列中, 生产者只关心是否有空间放数据, 消费者只关心是否能从空间中取到数据, 只要申请资源成功, 生产者可以和消费者并发访问环形队列;
那么可以分别记录两者的下标, 若下标位置不同, 那么双方一定是有资源的;

若下标位置相同, 那么只可能为一方空, 一方满;

单生产-单消费

单生产-单消费模型中, 当两者信号量都不为 0 时, 两者可以并发执行;
当生产者信号量为 0 时, 生产者阻塞等待, 等待消费者消费; 当消费者信号量为 0 时, 消费者会阻塞等待, 等待生产者生产; 当对方生产/消费后, 自己就会唤醒, 保证了同步和互斥;

#include "Thread.hpp"

template <class T>
class RingQueue
{

private:
    void P(sem_t& sem)
    {
        sem_wait(&sem);
    }

    void V(sem_t& sem)
    {
        sem_post(&sem);
    }

public:
    RingQueue(int cap = 10)
        : _cap(cap), _pro_pos(0), _con_pos(0)
    {
        _queue.resize(cap);
        // 初始化信号量
        sem_init(&_pro_sem, 0, cap);
        sem_init(&_con_sem, 0, 0);
    }

    void Push(const T &in)
    {
        
        P(_pro_sem);    // 申请信号量

        // 至当前位置, 一定申请成功, 就一定会有资源

        _queue[_pro_pos++] = in; // 插入数据
        _pro_pos %= _cap;

        V(_con_sem);    // 释放信号量, 但释放的是消费者的信号量, 增加消费者可用资源数目
    }

    void Pop(T &out)
    {
        P(_con_sem);    // 申请信号量

        // 至当前位置, 一定申请成功, 就一定会有资源

        out = _queue[_con_pos++]; // 删除数据
        _con_pos %= _cap;

        V(_pro_sem);    // 释放信号量, 但释放的是生产者的信号量, 增加生产者可用资源数目
    }

    ~RingQueue()
    {
        // 销毁信号量
        sem_destroy(&_pro_sem);
        sem_destroy(&_con_sem);
    }

private:
    vector<T> _queue;
    int _cap;
    size_t _pro_pos; // 生产者下标
    size_t _con_pos; // 消费者下标

    sem_t _pro_sem;  // 生产者的信号量
    sem_t _con_sem;  // 消费者的信号量
};

#include "BlockingQueue.hpp"
#include "RingQueue.hpp"

LockGuard guard;

template<class T>
void* Product(Thread<T>* self, T& args)
{
    RingQueue<int>* _queue = (RingQueue<int>*)args;
    int n = 1;
    while (1)
    {
        cout << self->get_name() << " " << n << endl;
        cout << "---------" << endl;
        _queue->Push(n++);
        sleep(1);
    }
    return 0;
}

template<class T>
void* Consum(Thread<T>* self, T& args)
{
    int n;
    RingQueue<int>* _queue = (RingQueue<int>*)args;
    while (1)
    {
        _queue->Pop(n);
        cout << self->get_name() << " " << n << endl;
        cout << "---------" << endl;
        sleep(1);
    }
    return 0;
}


template<class T>
void Wait(vector<Thread<T> >& threads, int n)
{
    for (int i=0; i<n; i++)
        threads[i].join();
}

template<class T>
void Start(vector<Thread<T> >& threads, int n)
{
    for (int i=0; i<n; i++)
        threads[i].start();
    
    //sleep(2);
}

template<class T>
void Init(vector<Thread<T> >& threads, int n, T func(Thread<T>*,T&), void* data = nullptr, const char* name = "thread-")
{
    for (int i=0; i<n; i++)
        threads.emplace_back(func, data, name+to_string(i+1));
}

int main()
{
    int n = 1;
    int m = 1;
    // vector<Thread<void*> > threads;

    //BlockingQueue<int> blcok_queue;
    RingQueue<int> _queue;

    vector<Thread<void*> > product;
    vector<Thread<void*> > consum;
    Init(product, n, Product, &_queue, "product-");
    Init(consum, m, Consum, &_queue, "consum-");
    Start(product, n);
    Start(consum, m);
    Wait(product, n);
    Wait(consum, m);
   
    cout << "---------" << endl;

    return 0;
}

多生产-多消费

但在多生产-多消费中需要注意, 由于生产者和生产者, 消费者和消费者之间存在互斥关系, 所以需要增加两把锁;

#include "Thread.hpp"

template <class T>
class RingQueue
{

private:
    void P(sem_t &sem)
    {
        sem_wait(&sem);
    }

    void V(sem_t &sem)
    {
        sem_post(&sem);
    }

public:
    RingQueue(int cap = 10)
        : _cap(cap), _pro_pos(0), _con_pos(0)
    {
        _queue.resize(cap);
        // 初始化信号量
        sem_init(&_pro_sem, 0, cap);
        sem_init(&_con_sem, 0, 0);
    }

    void Push(const T &in)
    {

        P(_pro_sem);       // 申请信号量
        _pro_mutex.Lock(); // 申请加锁

        // 至当前位置, 一定申请成功

        _queue[_pro_pos++] = in; // 插入数据
        _pro_pos %= _cap;

        _pro_mutex.Unlock(); // 申请解锁
        V(_con_sem);         // 释放信号量, 但释放的是消费者的信号量, 增加消费者可用资源数目
    }

    void Pop(T &out)
    {
        P(_con_sem);       // 申请信号量
        _con_mutex.Lock(); // 申请加锁

        // 至当前位置, 一定申请成功

        out = _queue[_con_pos++]; // 删除数据
        _con_pos %= _cap;

        _con_mutex.Unlock(); // 申请解锁
        V(_pro_sem);         // 释放信号量, 但释放的是生产者的信号量, 增加生产者可用资源数目
    }

    ~RingQueue()
    {
        // 销毁信号量
        sem_destroy(&_pro_sem);
        sem_destroy(&_con_sem);
    }

private:
    vector<T> _queue;
    int _cap;
    size_t _pro_pos; // 生产者下标
    size_t _con_pos; // 消费者下标

    sem_t _pro_sem; // 生产者的信号量
    sem_t _con_sem; // 消费者的信号量

    LockGuard _pro_mutex; // 生产者的信号量
    LockGuard _con_mutex; // 消费者的信号量
};