【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;elsebreak;}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; // 消费者的信号量
};