Linux学习笔记（十九）--生产消费模型与线程安全

生产者消费者模型

概念

生产者与消费者模型是一种经典的并发编程模型，用于解决多线程/多进程间共享数据的同步与互斥问题。该模型的核心在于：生产者负责生成数据并放入缓冲区，消费者负责从缓冲区取出数据并处理。

为什么需要生产者消费者模型

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

优点

（1）解耦

（2）支持并发

（3）支持忙闲不均

基于BlockingQueue的生产者消费者模型

概念

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

代码示例

#include <iostream> #include <queue> #include <stdlib.h> #include <pthread.h> #include <unistd.h> // 用于 sleep() #include <ctime> // 用于 time() #define NUM 8 class BlockQueue { private: std::queue<int> q; int cap; pthread_mutex_t lock; pthread_cond_t full; pthread_cond_t empty; private: void LockQueue() { pthread_mutex_lock(&lock); } void UnLockQueue() { pthread_mutex_unlock(&lock); } void ProductWait() { pthread_cond_wait(&full, &lock); } void ConsumeWait() { pthread_cond_wait(&empty, &lock); } void NotifyProduct() { pthread_cond_signal(&full); } void NotifyConsume() { pthread_cond_signal(&empty); } bool IsEmpty() { return q.size() == 0 ? true : false; } bool IsFull() { return q.size() == cap ? true : false; } public: BlockQueue(int _cap = NUM) : cap(_cap) { pthread_mutex_init(&lock, NULL); pthread_cond_init(&full, NULL); pthread_cond_init(&empty, NULL); } void PushData(const int &data) { LockQueue(); while (IsFull()) { NotifyConsume(); // 通知消费者消费，让出空间 std::cout << "queue full, notify consume data, product stop." << std::endl; ProductWait(); // 生产者阻塞等待 } q.push(data); NotifyConsume(); // 通知消费者有新数据 UnLockQueue(); } void PopData(int &data) { LockQueue(); while (IsEmpty()) { NotifyProduct(); // 通知生产者生产 std::cout << "queue empty, notify product data, consume stop." << std::endl; ConsumeWait(); // 消费者阻塞等待 } data = q.front(); q.pop(); NotifyProduct(); // 通知生产者有空位 UnLockQueue(); } ~BlockQueue() { pthread_mutex_destroy(&lock); pthread_cond_destroy(&full); pthread_cond_destroy(&empty); } }; void *consumer(void *arg) { BlockQueue *bqp = (BlockQueue *)arg; int data; for (; ; ) { bqp->PopData(data); std::cout << "consume data done : " << data << std::endl; } return nullptr; } // more faster void *producer(void *arg) { BlockQueue *bqp = (BlockQueue *)arg; srand((unsigned long)time(NULL)); for (; ; ) { int data = rand() % 1024; bqp->PushData(data); std::cout << "product data done: " << data << std::endl; // sleep(1); // 可选：控制生产速度 } return nullptr; } int main() { BlockQueue bq; pthread_t c, p; pthread_create(&c, NULL, consumer, (void *)&bq); pthread_create(&p, NULL, producer, (void *)&bq); pthread_join(c, NULL); pthread_join(p, NULL); return 0; }

POSIX信号量

概念

POSIX 信号量是一种用于进程间或线程间同步的机制，用于控制对共享资源的访问。与互斥锁相比，信号量可以允许多个线程/进程同时访问资源，而互斥锁一次只允许一个。

初始化信号量

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

参数：

（1）sem：信号量指针
（2）pshared：0表示线程间共享，非0表示进程间共享
（3）value：信号量初始值

销毁信号量

int sem_destroy(sem_t *sem);

等待信号量

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

发布信号量

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

获取信号量

#include <semaphore.h> int sem_getvalue(sem_t *sem, int *sval);

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

概念

环形队列（Circular Buffer/Ring Buffer）是一种高效的生产者消费者实现方式，特别适合固定大小缓冲区的场景。它通过头尾指针循环使用内存空间，避免了数据的搬移开销。

代码示例

#include <iostream> #include <vector> #include <mutex> #include <condition_variable> #include <thread> #include <cstdlib> #include <ctime> #include <unistd.h> // for sleep() template <typename T> class RingBuffer { private: std::vector<T> buffer_; // 环形队列的底层存储（数组） size_t capacity_; // 队列容量（固定大小） size_t head_ = 0; // 消费者读取位置（读指针） size_t tail_ = 0; // 生产者写入位置（写指针） size_t size_ = 0; // 当前元素数量（原子变量保证线程安全） std::mutex mtx_; // 互斥锁，保护共享资源（队列、指针、size） std::condition_variable not_empty_; // 队列不空时唤醒消费者 std::condition_variable not_full_; // 队列不满时唤醒生产者 // 辅助函数：计算下一个位置（环形） size_t next_pos(size_t pos) const { return (pos + 1) % capacity_; } public: // 构造函数：初始化队列容量 explicit RingBuffer(size_t capacity) : capacity_(capacity), buffer_(capacity) { if (capacity == 0) { throw std::invalid_argument("RingBuffer capacity must be > 0"); } } // 生产者写入数据（阻塞直到队列不满） void push(const T& data) { std::unique_lock<std::mutex> lock(mtx_); // 队列满时，等待“队列不满”的条件（not_full_被唤醒） not_full_.wait(lock, [this]() { return size_ < capacity_; }); // 写入数据到tail位置 buffer_[tail_] = data; tail_ = next_pos(tail_); // 移动写指针（环形） size_++; // 元素数量+1 // 唤醒所有等待“队列不空”的消费者 not_empty_.notify_one(); } // 消费者读取数据（阻塞直到队列不空） T pop() { std::unique_lock<std::mutex> lock(mtx_); // 队列空时，等待“队列不空”的条件（not_empty_被唤醒） not_empty_.wait(lock, [this]() { return size_ > 0; }); // 从head位置读取数据 T data = buffer_[head_]; head_ = next_pos(head_); // 移动读指针（环形） size_--; // 元素数量-1 // 唤醒所有等待“队列不满”的生产者 not_full_.notify_one(); return data; } // 析构函数：可选实现（本例中无需特殊操作） ~RingBuffer() = default; }; // 消费者线程函数：循环从队列取数据并打印 void consumer_task(RingBuffer<int>* rb) { while (true) { int data = rb->pop(); std::cout << "[消费者] 获取数据: " << data << std::endl; sleep(1); // 模拟消费耗时 } } // 生产者线程函数：循环生成随机数并放入队列 void producer_task(RingBuffer<int>* rb) { srand(time(nullptr)); // 初始化随机数种子 while (true) { int data = rand() % 1000; // 生成0~999的随机数 rb->push(data); std::cout << "[生产者] 生产数据: " << data << std::endl; sleep(1); // 模拟生产耗时（可根据需求调整） } } int main() { const size_t QUEUE_CAPACITY = 8; // 环形队列容量 RingBuffer<int> rb(QUEUE_CAPACITY); // 初始化环形队列 // 创建生产者和消费者线程 std::thread producer(producer_task, &rb); std::thread consumer(consumer_task, &rb); // 等待线程结束（本例中线程是无限循环，需手动终止，或按Ctrl+C退出） producer.join(); consumer.join(); return 0; }

应用场景

1.数据流处理

2.网络数据包缓冲

3.实时数据采集

4.任务调度系统

线程池

概念

线程池用于管理和复用线程的编程模式。它的核心思想是预先创建一组线程，放入“池”中，然后等待任务队列分配任务。当有任务需要执行时，线程池会分配一个空闲线程来处理，任务完成后线程返回池中等待下一个任务，而不是被销毁。这避免了频繁创建和销毁线程的开销，提高了系统性能和响应速度。

主要组件

任务队列：一个线程安全的数据结构（如链表），用于存放待处理的任务。生产者（如主线程）将任务放入队列，消费者（线程池中的工作线程）从队列中取出任务执行。
工作线程池：一组预先创建并启动的线程。它们的工作循环就是从任务队列中获取任务并执行。
线程池管理器：负责线程池的初始化、创建/销毁线程、动态调整线程数量、关闭线程池等。

基本工作流程

初始化：创建指定数量的工作线程，并启动它们。这些线程启动后会阻塞在任务队列上，等待任务。
提交任务：当有新的任务需要处理时，将任务（通常是一个函数指针和其参数）封装成结构体，放入任务队列。
​任务调度​：空闲的工作线程被唤醒，从任务队列中取出一个任务。
​执行任务​：工作线程执行任务函数。
​任务完成​：任务执行完毕后，线程返回，继续等待下一个任务。
​销毁线程池​：当不再需要线程池时，发送停止信号，等待所有工作线程处理完当前任务后退出，并回收资源。

线程池示例

1.创建固定数量线程池，循环从任务队列中获取任务对象。

2.获取到任务对象后，执行任务对象中的任务接口

#ifndef __M_TP_H__ #define __M_TP_H__ #include <iostream> #include <queue> #include <pthread.h> #define MAX_THREAD 5 typedef bool (*handler_t)(int); class ThreadTask { private: int _data; handler_t _handler; public: ThreadTask():_data(-1), _handler(NULL) {} ThreadTask(int data, handler_t handler) { _data = data; _handler = handler; } void SetTask(int data, handler_t handler) { _data = data; _handler = handler; } void Run() { _handler(_data); } }; class ThreadPool { private: int _thread_max; int _thread_cur; bool _tp_quit; std::queue<ThreadTask *> _task_queue; pthread_mutex_t _lock; pthread_cond_t _cond; private: void LockQueue() { pthread_mutex_lock(&_lock); } void UnLockQueue() { pthread_mutex_unlock(&_lock); } void wakeUpOne() { pthread_cond_signal(&_cond); } void wakeUpAll() { pthread_cond_broadcast(&_cond); } void ThreadQuit() { _thread_cur--; UnLockQueue(); pthread_exit(NULL); } void ThreadWait(){ if (_tp_quit) { ThreadQuit(); } pthread_cond_wait(&_cond, &_lock); } bool IsEmpty() { return _task_queue.empty(); } static void *thr_start(void *arg) { ThreadPool *tp = (ThreadPool*)arg; while(1) { tp->LockQueue(); while(tp->IsEmpty()) { tp->ThreadWait(); } ThreadTask *tt; tp->PopTask(&tt); tp->UnLockQueue(); tt->Run(); delete tt; } return NULL; } public: ThreadPool(int max=MAX_THREAD):_thread_max(max), _thread_cur(max), _tp_quit(false) { pthread_mutex_init(&_lock, NULL); pthread_cond_init(&_cond, NULL); } ~ThreadPool() { pthread_mutex_destroy(&_lock); pthread_cond_destroy(&_cond); } bool PoolInit() { pthread_t tid; for (int i = 0; i < _thread_max; i++) { int ret = pthread_create(&tid, NULL, thr_start, this); if (ret != 0) { std::cout<<"create pool thread error\n"; return false; } } return true; } bool PushTask(ThreadTask *tt) { LockQueue(); if (_tp_quit) { UnLockQueue(); return false; } _task_queue.push(tt); wakeUpOne(); UnLockQueue(); return true; } bool PopTask(ThreadTask **tt) { *tt = _task_queue.front(); _task_queue.pop(); return true; } bool PoolQuit() { LockQueue(); _tp_quit = true; UnLockQueue(); while(_thread_cur > 0) { wakeUpAll(); usleep(1000); } return true; } }; #endif

#include "threadpool.hpp" #include <unistd.h> #include <cstdlib> #include <ctime> bool handler(int data) { srand(time(NULL)); int n = rand() % 5; printf("thread: %p run task: %d--sleep %d sec\n", pthread_self(), data, n); sleep(n); return true; } int main() { int i; ThreadPool pool; pool.PoolInit(); for (i = 0; i < 10; i++) { ThreadTask *tt = new ThreadTask(i, handler); pool.PushTas(tt); } pokol.PoolQuit(); return 0; }

优点

（1）降低资源消耗：通过复用已创建的线程，减少线程创建和销毁的系统开销。
（2）提高响应速度：当任务到达时，无需等待线程创建即可立即执行。
（3）提高线程的可管理性：线程是稀缺资源，无限制创建会消耗大量系统资源并降低稳定性。使用线程池可以进行统一的分配、调优和监控。
（4）提供任务排队和拒绝机制：当任务过多时，可以通过队列缓冲，或实现拒绝策略防止系统过载。

应用场景

1. 需要大量的线程来完成任务，且完成任务的时间比较短。 WEB服务器完成网页请求这样的任务，使用线程池技术是非常合适的。因为单个任务小，而任务数量巨大，你可以想象一个热门网站的点击次数。 但对于长时间的任务，比如一个Telnet连接请求，线程池的优点就不明显了。因为Telnet会话时间比线程的创建时间大多了。
2. 对性能要求苛刻的应用，比如要求服务器迅速响应客户请求。
3. 接受突发性的大量请求，但不至于使服务器因此产生大量线程的应用。突发性大量客户请求，在没有线程池情况下，将产生大量线程，虽然理论上大部分操作系统线程数目最大值不是问题，短时间内产生大量线程可能使内存到达极限，出现错误。

线程安全的单例模式

单例模式

线程安全的单例模式是一种在多线程环境下保证单例对象只被创建一次的常见设计模式。

饿汉实现方式与懒汉实现方式

饿汉方式实现单例模式

template <typename T> class Singleton { static T data; public: static T* GetInstance() { return &data; } };

懒汉方式实现单例模式

template <typename T> class Singleton { static T* inst; public: static T* GetInstance() { if (inst == NULL) { inst = new T(); } return inst; } };

懒汉模式实现单例模式（线程安全版本）

// 懒汉模式, 线程安全 template <typename T> class Singleton { volatile static T* inst; // 需要设置 volatile 关键字, 否则可能被编译器优化. static std::mutex lock; public: static T* GetInstance() { if (inst == NULL) { // 双重判定空指针, 降低锁冲突的概率, 提高性能. lock.lock(); // 使用互斥锁, 保证多线程情况下也只调用一次 new. if (inst == NULL) { inst = new T(); } lock.unlock(); } return inst; } };

STL与线程安全

STL中的容器是否是线程安全的?

不是. 原因是, STL 的设计初衷是将性能挖掘到极致, 而一旦涉及到加锁保证线程安全, 会对性能造成巨大的影响. 而且对于不同的容器, 加锁方式的不同, 性能可能也不同(例如hash表的锁表和锁桶). 因此 STL 默认不是线程安全. 如果需要在多线程环境下使用, 往往需要调用者自行保证线程安全.

智能指针与线程安全

智能指针是否是线程安全的?

对于 unique_ptr, 由于只是在当前代码块范围内生效, 因此不涉及线程安全问题. 对于 shared_ptr, 多个对象需要共用一个引用计数变量, 所以会存在线程安全问题. 但是标准库实现的时候考虑到了这 个问题, 基于原子操作(CAS)的方式保证 shared_ptr 能够高效, 原子的操作引用计数。

其他常见的各种锁

悲观锁：在每次取数据时，总是担心数据会被其他线程修改，所以会在取数据前先加锁（读锁，写锁，行锁等），当其他线程想要访问数据时，被阻塞挂起。
乐观锁：每次取数据时候，总是乐观的认为数据不会被其他线程修改，因此不上锁。但是在更新数据前，会判断其他数据在更新前有没有对数据进行修改。主要采用两种方式：版本号机制和CAS操作。
CAS操作：当需要更新数据时，判断当前内存值和之前取得的值是否相等。如果相等则用新值更新。若不等则失败，失败则重试，一般是一个自旋的过程，即不断重试。

除了上述锁还有自旋锁，公平锁和非公平锁等。