线程同步与互斥
- 一.线程互斥
- 1.互斥相关概念
- 2.互斥锁 Mutex
- 3.互斥锁接口
- 4.互斥锁实现原理
- 5.互斥锁封装
- 二.线程同步
- 1.同步相关概念
- 2.条件变量 Condition Variable
- 3.条件变量接口
- 4.条件变量封装
- 5.信号量 Semaphore
- 6.信号量接口
- 7.信号量封装
- 三.生产者 - 消费者模型
- 1.基于 Blocking Queue 的生产者 - 消费者模型
- 2.基于 Ring Queue 的生产者 - 消费者模型
本节重点：
- 深刻理解线程互斥的原理和操作。
- 深刻理解线程同步。
- 掌握生产消费模型。
一.线程互斥
1.互斥相关概念
- 共享资源：可以被多个进程或线程可以共同访问和使用的资源。
- 临界资源：被保护的共享资源，它在同一时刻只允许一个进程或线程访问该资源。
- 临界区：每个线程内部，访问临界资源的代码，就叫做临界区。
- 原子性：不会被任何调度机制打断的操作，该操作只有两态，要么完成，要么未完成。
- 互斥：在同一时刻，只允许一个线程或进程访问共享资源。确保对共享资源的操作具有原子性，避免多个线程或进程同时对共享资源进行读写操作而导致的数据竞争和不一致问题。
2.互斥锁 Mutex
- 大部分情况，线程使用的数据都是局部变量，变量的地址空间在线程栈空间内，这种情况，变量归属单个线程，其它线程无法获得这种变量。
- 但有时候，很多变量都需要在线程间共享，这样的变量称为共享变量，可以通过数据的共享，完成线程之间的交互。但是多个线程并发的操作共享变量，存在数据安全问题。
#include <iostream>
#include <vector>
#include "Pthread.hpp"
using namespace ThreadModule;#define NUM 4int ticketnum = 10000; // 共享资源void Ticket()
{while(true){if(ticketnum > 0){usleep(1000);// 1.抢票std::cout << "get a new ticket, id: " << ticketnum << std::endl;ticketnum--;// 2.入库模拟// usleep(1000);}else{break;}}
}int main()
{std::vector<Thread> threads;// 1.构建线程对象for(int i = 0; i < NUM; i++){threads.emplace_back(Ticket);}// 2.启动线程for(auto& thread : threads){thread.Start();}// 3.等待线程for(auto& thread : threads){thread.Join();}return 0;
}
xzy@hcss-ecs-b3aa:~$ ./ticket
get a new ticket, id: 10000
get a new ticket, id: 9999
...
get a new ticket, id: 2
get a new ticket, id: 1
get a new ticket, id: 0
get a new ticket, id: -1
get a new ticket, id: -2
为什么可能无法获得正确结果？
- if 语句判断条件为真以后，代码可以并发的切换到其它线程。
- usleep(1000); 这个模拟漫长业务的过程，在这个漫长的业务过程中，可能有很多个线程会进入该代码段。
- ticketnum-- 操作本身就不是一个原子操作。
# 取出ticket--部分的汇编代码
xzy@hcss-ecs-b3aa:~$ objdump -d ticket > ticket.s
xzy@hcss-ecs-b3aa:~$ vim ticket.s
25a3: 8b 05 6b 5a 00 00 mov 0x5a6b(%rip),%eax
25a9: 83 e8 01 sub $0x1,%eax
25ac: 89 05 62 5a 00 00 mov %eax,0x5a62(%rip)
ticketnum-- 操作并不是原子操作，而是对应三条汇编指令：
- load：将共享变量 ticket 从内存加载到寄存器中。
- update：更新寄存器里面的值，执行-1操作。
- store：将新值，从寄存器写回共享变量 ticket 的内存地址。
要解决以上问题，需要做到三点：
- 代码必须要有互斥行为：当代码进入临界区执行时，不允许其它线程进入该临界区。
- 如果多个线程同时要求执行临界区的代码，并且临界区没有线程在执行，那么只能允许一个线程进入该临界区。
- 如果线程不在临界区中执行，那么该线程不能阻止其它线程进入临界区。
要做到这三点，本质上就是需要一把锁。Linux上提供的这把锁叫互斥锁(也叫互斥量)
- 所有对资源的保护，都是对临界区代码的保护，因为资源是通过代码访问的！
- 加锁一定不能大块代码进行加锁，要保证细粒度！
- 锁本身是全局的，也是共享资源。锁保护共享资源，那么谁保证锁？加锁和解锁被设计称为原子！要么执行完，要么未被执行，不需要被保护！
- 二元信号量就是锁，加锁的本质就是对资源展开预定，整体使用资源！
- 如果申请锁的时候，被其它线程拿走了？其它线程要进行阻塞等待，保证只能有一个线程访问资源！
- 线程在访问临界区代码时，线程可以切换？可以切换，但是锁被线程拿走了，其它线程无法进入临界区！串行！原子性！效率低的原因！
3.互斥锁接口
功能: 初始化互斥锁// 静态分配(编译确定内存的大小和位置): 不需要销毁
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;// 动态分配(运行确定内存的大小和位置): 需要销毁
原型: int pthread_mutex_init(pthread_mutex_t *restrict mutex, const pthread_mutexattr_t *restrict attr);
参数mutex: 要初始化的互斥锁指针
参数attr: nullptr功能: 销毁互斥锁
原型: int pthread_mutex_destroy(pthread_mutex_t *mutex);
参数mutex: 要销毁的互斥锁指针
注意：
- 使用 PTHREAD_ MUTEX_ INITIALIZER 初始化的互斥锁不需要销毁。
- 不要销毁一个已经加锁的互斥锁。
- 已经销毁的互斥锁，要确保后面不会有线程再尝试加锁。
功能: 加锁, 其它线程不可以访问共享资源
原型: int pthread_mutex_lock(pthread_mutex_t *mutex);功能: 解锁, ,其它线程可以访问共享资源
原型: int pthread_mutex_unlock(pthread_mutex_t *mutex);
调用 pthread_mutex_lock 时，可能会遇到以下情况：
- 互斥锁处于未锁状态，该函数会将互斥锁锁定，同时返回成功。
- 发起函数调用时，其它线程已经锁定互斥锁，或者存在其它线程同时申请互斥量，但没有竞争到互斥锁，那么 pthread_mutex_lock 调用会陷入阻塞(执行流被挂起，调度其它线程)，等待互斥锁解锁。
// Pthread.hpp 在上一篇博客<线程概念与控制>, 模版封装线程库中// ticket.cc
#include <iostream>
#include <vector>
#include <string>
#include "Pthread.hpp"
using namespace ThreadModule;#define NUM 4// pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER; // 锁也是共享资源
int ticketnum = 1000; // 共享资源class ThreadData
{
public:std::string name;pthread_mutex_t *lock_ptr;
};void Ticket(ThreadData& td)
{while (true){// pthread_mutex_lock(&lock)pthread_mutex_lock(td.lock_ptr); // 加锁if (ticketnum > 0){usleep(1000);// 1.抢票std::cout << td.name << " get a new ticket, id: "<< ticketnum <<std::endl;ticketnum--;// pthread_mutex_unlock(&lock)pthread_mutex_unlock(td.lock_ptr); // 解锁// 2.入库模拟: 耗时, 防止该线程再次抢锁(访问资源)usleep(50);}else{// pthread_mutex_unlock(&lock)pthread_mutex_unlock(td.lock_ptr); // 解锁break;}}
}int main()
{pthread_mutex_t lock;pthread_mutex_init(&lock, nullptr);std::vector<Thread<ThreadData>> threads;// 1.构建线程对象for (int i = 0; i < NUM; i++){ThreadData* td = new ThreadData();td->lock_ptr = &lock;threads.emplace_back(Ticket, *td);td->name = threads[i].Name();}// 2.启动线程for (auto &thread : threads){thread.Start();}// 3.等待线程for (auto &thread : threads){thread.Join();}pthread_mutex_destroy(&lock);return 0;
}
xzy@hcss-ecs-b3aa:~$ ./ticket
Thread-1 get a new ticket, id: 1000
Thread-2 get a new ticket, id: 999
Thread-3 get a new ticket, id: 998
...
Thread-2 get a new ticket, id: 3
Thread-3 get a new ticket, id: 2
Thread-4 get a new ticket, id: 1
4.互斥锁实现原理
- 单纯的 i++ 或者 ++i 都不是原子的，有可能会有数据安全问题。
- 为了实现互斥锁操作，大多数体系结构都提供了 swap 或 exchange 指令，该指令的作用是把寄存器和内存单元的数据相交换，由于只有一条指令，保证了原子性，即使是多处理器平台，访问内存的总线周期也有先后，一个处理器上的交换指令执行时另一个处理器的交换指令只能等待总线周期。现在我们把 lock 和 unlock 的伪代码改一下。
5.互斥锁封装
- RAII 的核心思想是将资源的获取和初始化放在对象的构造函数中，将资源的释放放在对象的析构函数中。
- 实现RAII的加锁方式：构造函数实现加锁，析构函数实现解锁。
// Mutex.hpp
#pragma once#include <pthread.h>namespace MutexModule
{class Mutex{// 互斥锁: 不支持拷贝构造、拷贝赋值Mutex(const Mutex &m) = delete;Mutex &operator=(const Mutex &m) = delete;public:Mutex(){::pthread_mutex_init(&_mutex, nullptr);}~Mutex(){::pthread_mutex_destroy(&_mutex);}pthread_mutex_t *LockAddr() { return &_mutex; }void Lock(){::pthread_mutex_lock(&_mutex);}void Unlock(){::pthread_mutex_unlock(&_mutex);}private:pthread_mutex_t _mutex;};class LockGuard{public:LockGuard(Mutex &mutex): _mutex(mutex){_mutex.Lock();}~LockGuard(){_mutex.Unlock();}private:Mutex &_mutex; // 使用引用: 互斥锁不支持拷贝};
}// Main.cc
#include <string>
#include <unistd.h>
#include "Mutex.hpp"
using namespace MutexModule;int ticket = 1000;
Mutex mtx;void *Ticket(void* args)
{std::string name = static_cast<char*>(args);while(true){// mtx.Lock();LockGuard lg(mtx); // 临时对象: 初始化时自动加锁, 出while循环时自动解锁(RAII风格的加锁方式)if(ticket > 0){usleep(1000);std::cout << name << " buys a ticket: " << ticket << std::endl;ticket--;// mtx.Unlock();}else{// mtx.Unlock();break;}}return nullptr;
}int main()
{pthread_t t1, t2, t3;pthread_create(&t1, nullptr, Ticket, (void*)"thread-1");pthread_create(&t2, nullptr, Ticket, (void*)"thread-2");pthread_create(&t3, nullptr, Ticket, (void*)"thread-3");pthread_join(t1, nullptr);pthread_join(t2, nullptr);pthread_join(t3, nullptr);return 0;
}
xzy@hcss-ecs-b3aa:~$ ./ticket
thread-3 buys a ticket: 1000
thread-3 buys a ticket: 999
thread-3 buys a ticket: 998
...
thread-3 buys a ticket: 3
thread-3 buys a ticket: 2
thread-3 buys a ticket: 1
二.线程同步
1.同步相关概念
在互斥的代码中，发现同一个线程多次访问资源，导致其它进程迟迟访问不到资源，导致进程饥饿问题。
- 同步：在保证数据安全的前提下，让线程能够按照某种特定的顺序访问临界资源，从而有效避免线程饥饿问题，叫做同步。
- 竞态条件：多个线程或进程对共享资源的访问顺序和时间不确定，而导致程序异常。
- 互斥保证在同一时间，只有一个线程/进程访问共享资源，进而保证数据安全性，但是安全不一定合理/高效。同步是在互斥的前提下，让系统变得更加合理/高效。
- 如何做到线程同步？条件变量！
2.条件变量 Condition Variable
- 例如：互斥的三个线程买票时，第一个抢到锁的线程，发现没有票时，它会不断地加锁、什么都做不了、解锁(解锁后最接近加锁条件)，导致其它线程处于饥饿状态。当主线程发票之后，也是该线程抢到票。
#include <iostream>
#include <unistd.h>
#include <pthread.h>int ticket = 0;
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;void *route(void *args)
{std::string name = static_cast<char *>(args);while (1){pthread_mutex_lock(&mutex);if (ticket > 0){usleep(1000); // 微秒std::cout << name << " buys a new ticket: " << ticket << std::endl;ticket--;pthread_mutex_unlock(&mutex);}else{std::cout << name << " do nothing" << std::endl;sleep(1);pthread_mutex_unlock(&mutex);}}return nullptr;
}int main()
{pthread_t t1, t2, t3;pthread_create(&t1, nullptr, route, (void *)"thread 1");pthread_create(&t2, nullptr, route, (void *)"thread 2");pthread_create(&t3, nullptr, route, (void *)"thread 3");int cnt = 3;while(true){sleep(5);ticket += cnt;std::cout << "主线程发票啦, ticket: " << ticket << std::endl;}pthread_join(t1, nullptr);pthread_join(t2, nullptr);pthread_join(t3, nullptr);return 0;
}
xzy@hcss-ecs-b3aa:~$ ./testCond
thread 1 do nothing
thread 1 do nothing
thread 1 do nothing
thread 1 do nothing
thread 1 do nothing
主线程发票啦, ticket: 3
thread 1 buys a new ticket: 3
thread 1 buys a new ticket: 2
thread 1 buys a new ticket: 1
条件变量：通常与互斥锁一起使用。互斥锁用于保护共享资源，防止多个线程同时访问和修改这些资源而导致数据不一致；而条件变量则用于在线程之间传递状态信息，使得线程可以根据特定条件的满足与否来决定是继续执行还是等待。条件变量内部维护的是线程队列，实现线程同步！
3.条件变量接口
功能: 初始化条件变量// 静态分配(编译确定内存的大小和位置): 不需要销毁
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;// 动态分配(运行确定内存的大小和位置): 需要销毁
原型: int pthread_cond_init(pthread_cond_t *restrict cond, const pthread_condattr_t *restrict attr);
参数cond: 要初始化的条件变量指针
参数attr: nullptr功能: 销毁条件变量
原型: int pthread_cond_destroy(pthread_cond_t *cond);
参数cond: 要销毁的条件变量指针
功能: 让当前线程在指定的条件变量上阻塞等待, 直到其它线程通过线程发送信号/广播, 解除线程阻塞, 再在锁上等待, 直到申请锁成功
原型: int pthread_cond_wait(pthread_cond_t *restrict cond, pthread_mutex_t *restrict mutex);
工作原理：
- 当线程调用 pthread_cond_wait() 时，它会自动释放传入的互斥锁 mutex，这是为了让其它线程有机会获取该互斥锁，进而修改共享资源，使得等待的条件有可能得到满足。
- 线程进入阻塞状态，等待在条件变量 cond 上，此时线程不会占用 CPU 资源。
- 当其它线程调用 pthread_cond_signal() 或 pthread_cond_broadcast() 对同一个条件变量 cond 发出信号时，该线程被唤醒。
- 线程被唤醒后，会尝试重新获取之前释放的互斥锁 mutex。若互斥锁当前被其它线程占用，该线程会继续阻塞，直至成功获取互斥锁。一旦获取到锁，线程就会从 pthread_cond_wait() 函数返回，继续执行后续代码。
功能: 该函数用于向指定的条件变量cond发出信号, 唤醒一个正在该条件变量上等待的线程
原型: int pthread_cond_signal(pthread_cond_t *cond);功能: 该函数用于向指定的条件变量cond发出广播信号, 唤醒所有正在该条件变量上等待的线程
原型: int pthread_cond_broadcast(pthread_cond_t *cond);
注意：pthread_cond_signal() 和 pthread_cond_broadcast() 不会自动释放互斥锁，调用该函数的线程仍然持有互斥锁。调用后通常需要手动释放互斥锁，被唤醒的多个线程会竞争获取互斥锁，获取到锁的线程才能继续执行。
#include <iostream>
#include <unistd.h>
#include <pthread.h>pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;void *active(void *args)
{std::string name = static_cast<char *>(args);while (true){pthread_mutex_lock(&mutex);// 没有对资源释放就绪的判定// std::cout << name << " is waiting" << std::endl;pthread_cond_wait(&cond, &mutex); // mutex???std::cout << name << " is active" << std::endl;pthread_mutex_unlock(&mutex);}
}int main()
{pthread_t tid1, tid2, tid3;pthread_create(&tid1, nullptr, active, (void *)"thread-1");pthread_create(&tid2, nullptr, active, (void *)"thread-2");pthread_create(&tid3, nullptr, active, (void *)"thread-3");sleep(1);std::cout << "main thread ctrl begin..." << std::endl;while (true){std::cout << "main wakeup thread..." << std::endl;pthread_cond_signal(&cond);// pthread_cond_broadcast(&cond);sleep(1);}pthread_join(tid1, nullptr);pthread_join(tid2, nullptr);pthread_join(tid3, nullptr);return 0;
}
# 按照线程1、2、3的顺序, 实现同步
xzy@hcss-ecs-b3aa:~$ ./testCond
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
main wakeup thread...
thread-1 is active
main wakeup thread...
thread-2 is active
main wakeup thread...
thread-3 is active
#include <iostream>
#include <unistd.h>
#include <pthread.h>int ticket = 0;
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;void *route(void *args)
{std::string name = static_cast<char *>(args);while (1){pthread_mutex_lock(&mutex);if (ticket > 0){usleep(1000);std::cout << name << " buys a new ticket: " << ticket << std::endl;ticket--;pthread_mutex_unlock(&mutex);}else{pthread_cond_wait(&cond, &mutex);std::cout << "主线程出票完成, " << name << " 醒来" << std::endl;pthread_mutex_unlock(&mutex);}// usleep(50);}return nullptr;
}int main()
{pthread_t t1, t2, t3;pthread_create(&t1, nullptr, route, (void *)"thread 1");pthread_create(&t2, nullptr, route, (void *)"thread 2");pthread_create(&t3, nullptr, route, (void *)"thread 3");int cnt = 10;while(true){sleep(5);ticket += cnt;std::cout << "主线程发票啦, ticket: " << ticket << std::endl;pthread_cond_signal(&cond);// pthread_cond_broadcast(&cond);}pthread_join(t1, nullptr);pthread_join(t2, nullptr);pthread_join(t3, nullptr);return 0;
}
# 按照线程1、2、3的顺序依次抢票, 实现同步
xzy@hcss-ecs-b3aa:~$ ./testCond
主线程发票啦, ticket: 3
主线程出票完成, thread 1 醒来
thread 1 buys a new ticket: 3
thread 1 buys a new ticket: 2
thread 1 buys a new ticket: 1
主线程发票啦, ticket: 3
主线程出票完成, thread 2 醒来
thread 2 buys a new ticket: 3
thread 2 buys a new ticket: 2
thread 2 buys a new ticket: 1
主线程发票啦, ticket: 3
主线程出票完成, thread 3 醒来
thread 3 buys a new ticket: 3
thread 3 buys a new ticket: 2
thread 3 buys a new ticket: 1
4.条件变量封装
// Cond.hpp
#pragma once#include <pthread.h>
#include "Mutex.hpp"
using namespace MutexModule;namespace CondModule
{class Cond{public:Cond() {::pthread_cond_init(&_cond, nullptr);}~Cond() {::pthread_cond_destroy(&_cond);}void Wait(Mutex &mutex) // 线程释放曾经持有的锁, 不能拷贝{::pthread_cond_wait(&_cond, mutex.LockAddr());}void Signal(){::pthread_cond_signal(&_cond);}void Broadcast(){::pthread_cond_broadcast(&_cond);}private:pthread_cond_t _cond;};
}
5.信号量 Semaphore
信号量：一种用于多进程或多线程环境下实现同步与互斥的机制。避免多个进程或线程同时访问共享资源而引发的数据不一致或其他错误。
- SystemV 信号量：涉及内核，系统调用的开销较大，性能可能会受到影响，并且操作复杂。
- POSIX信号量：接口简洁，设计上更注重性能，尤其是对于线程间的同步和互斥操作。
信号量本质上是一个计数器，用于记录系统中某种资源的可用数量，也就是最多允许线程进入共享资源的数量。配合PV两个原子操作来控制对共享资源的访问。PV 原子操作：
- P 操做：如果信号量的值大于 0，将信号量的值减 1，然后继续执行；如果信号量的值为 0，则调用线程会被阻塞，直到信号量的值大于 0。
- V 操作：将信号量的值加 1。如果有其它线程正在等待该信号量，那么会唤醒其中一个等待的线程。
6.信号量接口
功能: 初始化信号量
原型: int sem_init(sem_t *sem, int pshared, unsigned int value);
参数:sem: 信号量对象的指针pshared: 0表示线程间共享, 非0表示进程间共享value: 信号量的初始值功能: 销毁化信号量
原型: int sem_destroy(sem_t *sem);功能: P操作
原型: int sem_wait(sem_t *sem);功能: V操作
原型: int sem_post(sem_t *sem);
- 二元信号量：可用的资源只有一份，只允许一个线程访问共享资源，类似互斥锁。当信号量的值为 1 时，表示资源可用；当值为 0 时，表示资源已被占用。如下用二元信号量写法代替互斥锁：
#include <iostream>
#include <unistd.h>
#include <semaphore.h>int ticket = 1000;
sem_t sem;void* Ticket(void *args)
{std::string name = static_cast<char*>(args);while(true){sem_wait(&sem); // 申请信号量if(ticket > 0){usleep(1000);std::cout << name << " buys a ticket: " << ticket << std::endl;ticket--;sem_post(&sem); // 释放信号量}else{sem_post(&sem); // 释放信号量break;}}return nullptr;
}int main()
{sem_init(&sem, 0, 1);pthread_t t1, t2, t3;pthread_create(&t1, nullptr, Ticket, (void*)"thread-1");pthread_create(&t2, nullptr, Ticket, (void*)"thread-2");pthread_create(&t3, nullptr, Ticket, (void*)"thread-3");pthread_join(t1, nullptr);pthread_join(t2, nullptr);pthread_join(t3, nullptr);sem_destroy(&sem);return 0;
}
xzy@hcss-ecs-b3aa:~$ ./ticket
thread-1 buys a ticket: 1000
thread-1 buys a ticket: 999
thread-1 buys a ticket: 998
...
thread-1 buys a ticket: 3
thread-1 buys a ticket: 2
thread-1 buys a ticket: 1
当允许最多3个线程并发访问共享资源时：如何确保数据安全问题？加锁！
#include <iostream>
#include <unistd.h>
#include <semaphore.h>
#include "Mutex.hpp"int ticket = 1000;
sem_t sem;
Mutex mutex;void* Ticket(void *args)
{std::string name = static_cast<char*>(args);while(true){sem_wait(&sem); // 申请信号量LockGuard lockguard(mutex); // RAII方式加锁if(ticket > 0){usleep(1000);std::cout << name << " buys a ticket: " << ticket << std::endl;ticket--;sem_post(&sem); // 释放信号量}else{sem_post(&sem); // 释放信号量break;}}return nullptr;
}int main()
{sem_init(&sem, 0, 3); // 允许最多3个线程访问共享资源pthread_t t1, t2, t3;pthread_create(&t1, nullptr, Ticket, (void*)"thread-1");pthread_create(&t2, nullptr, Ticket, (void*)"thread-2");pthread_create(&t3, nullptr, Ticket, (void*)"thread-3");pthread_join(t1, nullptr);pthread_join(t2, nullptr);pthread_join(t3, nullptr);sem_destroy(&sem);return 0;
}
xzy@hcss-ecs-b3aa:~$ ./ticket
thread-1 buys a ticket: 1000
thread-1 buys a ticket: 999
thread-1 buys a ticket: 998
...
thread-1 buys a ticket: 3
thread-1 buys a ticket: 2
thread-1 buys a ticket: 1
7.信号量封装
// Sem.hpp
#pragma once
#include <semaphore.h>
namespace SemMoudel
{class Sem{public:Sem(int value = 1): _value(value){::sem_init(&_sem, 0, _value);}~Sem(){::sem_destroy(&_sem);}void P(){::sem_wait(&_sem);}void V(){::sem_post(&_sem);}private:sem_t _sem;int _value;};
}
三.生产者 - 消费者模型
单线程通常是串行执行，多线程通常是单核CPU并发、多核CPU并行。并发比串行效率高，并行比并发效率高。多线程中的某一个线程在执行IO操作时，线程挂起，会释放 CPU 资源，允许操作系统将 CPU 时间片分配给其它就绪的线程，也就是以并发的方式提高 CPU 效率。
- 生产者 - 消费者模型：多线程或多进程协作设计模式，用于解决生产者和消费者之间数据交互问题。
- 生产者和消费者彼此之间不直接通讯，而通过缓冲区来进行通讯。生产者首先检查缓冲区是否还有空闲空间，如果有，则生产一个数据项并将其放入缓冲区；如果缓冲区已满，生产者需要阻塞等待，直到有消费者从缓冲区中取走数据，腾出空间。消费者检查缓冲区是否有可用的数据项，如果有，则从缓冲区中取出一个数据项进行处理；如果缓冲区为空，消费者需要等待，直到生产者向缓冲区中添加了新的数据项。这个缓冲区用来给生产者和消费者解耦的。
生产者 - 消费者模型效率高的原因：
- 解耦：生产者和消费者不需要直接交互，它们只需要与缓冲区进行交互，从而降低了两者之间的耦合度，使得系统的可维护性和可扩展性得到提高。
- 支持并发，但临界区需要同步互斥，防止并发造成数据不一致问题
- 缓存机制：平衡了生产和消费的速度差异。
总结：
- 一个交易场所：临界资源。
- 两个角色：生产者和消费者。
- 三种关系：生产者与生产者：互斥关系。消费者与消费者：互斥关系。生产者与消费者：互斥和同步关系(生产者需要等待缓冲区有空闲空间才能生产数据，消费者需要等待缓冲区有数据才能消费)
1.基于 Blocking Queue 的生产者 - 消费者模型
- 阻塞队列：一种常用于实现生产者 - 消费者模型的数据结构。
- 其与普通的队列区别：当队列为空时，从队列获取元素的操作将会被阻塞，直到队列中被放入了元素。当队列满时，往队列里存放元素的操作也会被阻塞，直到有元素被从队列中取出。生产者和消费者存在同步关系。
- 考虑使用 互斥锁 + 条件变量 实现阻塞队列！
利用 pthread.h 线程库的 Mutex 和 Cond 实现基于 Blocking Queue 的生产者 - 消费者模型：
// BlockQueue.hpp
#pragma once#include <iostream>
#include <queue>
#include <pthread.h>namespace BlockQueueMoudel
{static int gcap = 10;template <class T>class BlockQueue{private:bool IsFull() { return _q.size() == _cap; }bool IsEmpty() { return _q.empty(); }public:BlockQueue(int cap = gcap): _cap(cap), _p_wait_num(0), _c_wait_num(0){pthread_mutex_init(&_mutex, nullptr);pthread_cond_init(&_producer_cond, nullptr);pthread_cond_init(&_consumer_cond, nullptr);}~BlockQueue(){pthread_mutex_destroy(&_mutex);pthread_cond_destroy(&_producer_cond);pthread_cond_destroy(&_consumer_cond);}// 生产者void Push(const T &in){pthread_mutex_lock(&_mutex);// 生产数据是有条件的, 容量不能为满while (IsFull()) // while替代if: 防止伪唤醒{std::cout << "生产者进入等待..." << std::endl;_p_wait_num++;// 生产者线程等待时, 需解锁让线程挂起, 调度消费者线程, 消费一个数据后,// 通知生产者线程条件满足(可以生产数据), 接着需要再阻塞等待加锁pthread_cond_wait(&_producer_cond, &_mutex); // 等待必须在临界区: IsFull访问了临界资源// wait完成后: 生产者线程被唤醒 && 重新申请并持有锁(仍在临界区)_p_wait_num--;std::cout << "生产者已被唤醒..." << std::endl;}// IsFull()不满足 || 生产者线程被唤醒_q.push(in);// 肯定有数据: 若消费者线程在等待, 直接唤醒if (_c_wait_num){// std::cout << "唤醒消费者" << std::endl;pthread_cond_signal(&_consumer_cond);}// 唤醒在解锁后也可以pthread_mutex_unlock(&_mutex);}// 消费者void Pop(T *out){pthread_mutex_lock(&_mutex);// 消费数据是有条件的, 容量不能为空while (IsEmpty()) // while替代if: 防止伪唤醒{std::cout << "消费者进入等待..." << std::endl;_c_wait_num++;// 消费者线程等待时, 需解锁让线程挂起, 调度生产者线程, 生产一个数据后,// 通知消费者线程条件满足(可以消费数据), 接着需要再阻塞等待加锁pthread_cond_wait(&_consumer_cond, &_mutex); // 等待必须在临界区: IsEmpty访问了临界资源// wait完成后: 消费者线程被唤醒 && 重新申请并持有锁(仍在临界区)_c_wait_num--;std::cout << "消费者已被唤醒..." << std::endl;}// IsEmpty()不满足 || 消费者线程被唤醒*out = _q.front();_q.pop();// 肯定有空间: 若生产者线程在等待, 直接唤醒if (_p_wait_num){// std::cout << "唤醒生产者" << std::endl;pthread_cond_signal(&_producer_cond);}// 唤醒在解锁后也可以pthread_mutex_unlock(&_mutex);}private:std::queue<T> _q; // 保存数据, 临界资源int _cap; // bq的最大容量pthread_mutex_t _mutex; // 互斥锁pthread_cond_t _producer_cond; // 生产者条件变量pthread_cond_t _consumer_cond; // 消费者条件变量int _p_wait_num; // 生产者线程等待个数int _c_wait_num; // 消费者线程等待个数};
}// Main.cc
#include <unistd.h>
#include "BlockQueue.hpp"
using namespace BlockQueueMoudel;// 生产者
void *Producer(void *args)
{BlockQueue<int> *bq = static_cast<BlockQueue<int> *>(args);int data = 10;while (true){// sleep(2);// 1.生产到bq队列中bq->Push(data);std::cout << "producer 生产了一个数据: " << data << std::endl;// 2.更新下一个生产的数据data++;}
}// 消费者
void *Consumer(void *args)
{BlockQueue<int> *bq = static_cast<BlockQueue<int> *>(args);while (true){sleep(2);// 1.从bq队列获取数据int data;bq->Pop(&data);// 2.消费数据std::cout << "consumer 消费了一个数据: " << data << std::endl;}
}int main()
{BlockQueue<int> *bq = new BlockQueue<int>(5);// 单生产者、单消费者pthread_t p, c;pthread_create(&p, nullptr, Producer, bq);pthread_create(&c, nullptr, Consumer, bq);pthread_join(p, nullptr);pthread_join(c, nullptr);// 多生产者、多消费者// pthread_t p1, p2, p3, c1, c2;// pthread_create(&p1, nullptr, Producer, bq);// pthread_create(&p2, nullptr, Producer, bq);// pthread_create(&p3, nullptr, Producer, bq);// pthread_create(&c1, nullptr, Consumer, bq);// pthread_create(&c2, nullptr, Consumer, bq);// pthread_join(p1, nullptr);// pthread_join(p2, nullptr);// pthread_join(p3, nullptr);// pthread_join(c1, nullptr);// pthread_join(c2, nullptr);delete bq;return 0;
}
xzy@hcss-ecs-b3aa:~$ ./bq
producer 生产了一个数据: 10
producer 生产了一个数据: 11
producer 生产了一个数据: 12
producer 生产了一个数据: 13
producer 生产了一个数据: 14
生产者进入等待...
consumer 消费了一个数据: 10
生产者已被唤醒...
producer 生产了一个数据: 15
生产者进入等待...
consumer 消费了一个数据: 11
生产者已被唤醒...
producer 生产了一个数据: 16
生产者进入等待...
利用自己封装的 Mutex 和 Cond 实现基于 Blocking Queue 的生产者消费者模型：
// BlockQueue.hpp
#pragma once#include <iostream>
#include <queue>
#include <pthread.h>
#include "Mutex.hpp"
#include "Cond.hpp"
using namespace MutexModule;
using namespace CondModule;namespace BlockQueueMoudel
{static int gcap = 10;template <class T>class BlockQueue{private:bool IsFull() { return _q.size() == _cap; }bool IsEmpty() { return _q.empty(); }public:BlockQueue(int cap = gcap): _cap(cap), _p_wait_num(0), _c_wait_num(0){}~BlockQueue() {}// 生产者void Push(const T &in){LockGuard lockguard(_mutex);while (IsFull()){std::cout << "生产者进入等待..." << std::endl;_p_wait_num++;_producer_cond.Wait(_mutex);_p_wait_num--;std::cout << "生产者已被唤醒..." << std::endl;}_q.push(in);if (_c_wait_num){// std::cout << "唤醒消费者" << std::endl;_consumer_cond.Signal();}}// 消费者void Pop(T *out){LockGuard lockguard(_mutex);while (IsEmpty()){std::cout << "消费者进入等待..." << std::endl;_c_wait_num++;_consumer_cond.Wait(_mutex);_c_wait_num--;std::cout << "消费者已被唤醒..." << std::endl;}*out = _q.front();_q.pop();if (_p_wait_num){// std::cout << "唤醒生产者" << std::endl;_producer_cond.Signal();}}private:std::queue<T> _q; // 保存数据, 临界资源int _cap; // bq的最大容量Mutex _mutex; // 互斥锁Cond _producer_cond; // 生产者条件变量Cond _consumer_cond; // 消费者条件变量int _p_wait_num; // 生产者线程等待个数int _c_wait_num; // 消费者线程等待个数};
}// Main.cc
#include <iostream>
#include <vector>
#include <functional>
#include <unistd.h>
#include <pthread.h>
#include "BlockQueue.hpp"using namespace BlockQueueMoudel;
using task_t = std::function<void()>;std::vector<task_t> tasks;
void Sql() { std::cout << "我是一个数据库任务" << std::endl; }
void UpLoad() { std::cout << "我是一个上传任务" << std::endl; }
void DownLoad() { std::cout << "我是一个下载任务" << std::endl; }// 生产者
void *Producer(void *args)
{BlockQueue<task_t> *bq = static_cast<BlockQueue<task_t> *>(args);int cnt = 0;while (true){// sleep(2);// 1.从tasks数组中获取任务bq->Push(tasks[cnt % 3]);cnt++;// 2.生产任务std::cout << "producer 生产了一个任务" << std::endl;}
}// 消费者
void *Consumer(void *args)
{BlockQueue<task_t> *bq = static_cast<BlockQueue<task_t> *>(args);while (true){sleep(2);// 1.从bq队列中获取任务task_t t;bq->Pop(&t);// 2.处理任务t();std::cout << "consumer 处理完成一个任务" << std::endl;}
}int main()
{tasks.push_back(Sql);tasks.push_back(UpLoad);tasks.push_back(DownLoad);BlockQueue<task_t> *bq = new BlockQueue<task_t>(5);// 单生产者、单消费者pthread_t p, c;pthread_create(&p, nullptr, Producer, bq);pthread_create(&c, nullptr, Consumer, bq);pthread_join(p, nullptr);pthread_join(c, nullptr);delete bq;return 0;
}
xzy@hcss-ecs-b3aa:~$ ./bq
producer 生产了一个任务
producer 生产了一个任务
producer 生产了一个任务
producer 生产了一个任务
producer 生产了一个任务
生产者进入等待...
我是一个数据库任务
consumer 处理完成一个任务
生产者已被唤醒...
producer 生产了一个任务
生产者进入等待...
我是一个上传任务
consumer 处理完成一个任务
生产者已被唤醒...
producer 生产了一个任务
生产者进入等待...
我是一个下载任务
consumer 处理完成一个任务
生产者已被唤醒...
producer 生产了一个任务
生产者进入等待...
2.基于 Ring Queue 的生产者 - 消费者模型
- 环型队列：实现数据高效生产和消费的经典设计模式，用于解决多线程/多进程环境下生产者和消费者之间的数据共享与同步问题。
- 生产者负责将数据放入循环队列，而消费者则从队列中取出数据进行处理。为了确保线程安全和避免数据竞争，需要使用同步机制来控制对队列的访问。当队列已满时，生产者需要等待；当队列为空时，消费者需要等待。
- 考虑使用 信号量 实现循环队列！
通过生产者生产数据：空间-1，数据+1；消费者消费数据：数据-1，空间+1。实现数据和空间的平衡！
单生产者单消费者：
- 当队列为空/满 时：生产者和消费者访问同一个位置(资源)，同步互斥！
- 当队列非空/满 时：生产者和消费者访问不同的位置(资源)，并发！
多单生产者单消费者：
- 生产者和生产者互斥关系，消费者和消费者互斥关系。
- 生产者和消费者同意满足上面的关系。
利用自己封装的 Mutex 和 Sem 实现基于 Ring Queue 的生产者消费者模型：
// RingQueue.hpp
#pragma once#include <iostream>
#include <vector>
#include <pthread.h>
#include "Sem.hpp"
#include "Mutex.hpp"using namespace SemMoudel;
using namespace MutexModule;namespace RingQueueMoudel
{template <class T>class RingQueue{public:RingQueue(int cap): _rq(cap), _cap(cap), _p_pos(0), _c_pos(0), _data_sem(0), _space_sem(cap){}~RingQueue(){}// 生产者void Push(const T &in){// 当队满: 阻塞, 直到消费者消费数据_space_sem.P(); // 申请空间{// 先申请信号量, 再申请锁: 此时信号量的申请是并行的, 效率高一点LockGuard lockguard(_p_mutex);_rq[_p_pos] = in;_p_pos++;_p_pos %= _cap;}_data_sem.V(); // 释放数据}// 消费者void Pop(T *out){// 当队空: 阻塞, 直到生产者生产数据_data_sem.P(); // 申请数据{// 先申请信号量, 再申请锁: 此时信号量的申请是并行的, 效率高一点LockGuard lockguard(_c_mutex);*out = _rq[_c_pos];_c_pos++;_c_pos %= _cap;}_space_sem.V(); // 释放空间}private:std::vector<T> _rq; // 环型队列, 临界资源int _cap; // 最大容量int _p_pos; // 生产者位置int _c_pos; // 消费者位置Sem _data_sem; // 数据信号量Sem _space_sem; // 空间信号量Mutex _p_mutex; // 生产者的锁Mutex _c_mutex; // 消费者的锁};
}// Main.cc
#include <iostream>
#include <pthread.h>
#include <unistd.h>
#include "RingQueue.hpp"using namespace RingQueueMoudel;void *Producer(void *args)
{RingQueue<int> *rq = static_cast<RingQueue<int> *>(args);int data = 0;while(true){// 1.获取数据// 2.生产数据rq->Push(data);std::cout << "producer 生产了一个数据" << data << std::endl;data++;}
}void *Consumer(void *args)
{RingQueue<int> *rq = static_cast<RingQueue<int> *>(args);while(true){sleep(1);// 1.消费数据int data;rq->Pop(&data);// 2.处理数据std::cout << "consumer 消费了一个数据" << data << std::endl;}
}int main()
{RingQueue<int> *rq = new RingQueue<int>(5);pthread_t p, c;pthread_create(&p, nullptr, Producer, rq);pthread_create(&c, nullptr, Consumer, rq);pthread_join(p, nullptr);pthread_join(c, nullptr);// pthread_t p1, p2, c1, c2, c3;// pthread_create(&p1, nullptr, Producer, rq);// pthread_create(&p2, nullptr, Producer, rq);// pthread_create(&c1, nullptr, Consumer, rq);// pthread_create(&c2, nullptr, Consumer, rq);// pthread_create(&c3, nullptr, Consumer, rq);// pthread_join(p1, nullptr);// pthread_join(p2, nullptr);// pthread_join(c1, nullptr);// pthread_join(c2, nullptr);// pthread_join(c3, nullptr);delete rq;return 0;
}
xzy@hcss-ecs-b3aa:~$ ./rq
producer 生产了一个数据0
producer 生产了一个数据1
producer 生产了一个数据2
producer 生产了一个数据3
producer 生产了一个数据4
consumer 消费了一个数据0
producer 生产了一个数据5
consumer 消费了一个数据1
producer 生产了一个数据6
consumer 消费了一个数据2
producer 生产了一个数据7
