文章目录
- 一、线程封装
- 二、Linux线程互斥
- 进程线程间的互斥相关背景概念
- 互斥量mutex
- 为什么上面的抢票代码可能无法获得正确结果？（票数为负）
- 互斥量的接口
- 1. 初始化互斥量
- 方法一：静态分配
- 方法二：动态分配:
- 2. 销毁互斥量
- 3. 互斥量加锁和解锁
- 三、常见锁概念
- 死锁
- 死锁四个必要条件
- 避免死锁的方法
- 避免死锁的算法
- 四、生产消费者模型
- 为何要使用生产者消费者模型？
- 生产者消费者模型优点
- 基于BlockingQueue的生产者消费者模型
- BlockingQueue是什么
- C++ queue模拟阻塞队列的生产消费模型
一、线程封装
模拟封装C++11的thread：
#pragma once
#include <pthread.h>
#include <iostream>
#include <string>
#include <functional>
template<class T>
using func_t = std::function<void(T)>;
template<class T>
class Thread
{
public:
Thread(const std::string& threadname, func_t<T> func, T data)
:_tid(0)
, _isrunning(false)
, _threadname(threadname)
, _func(func)
, _data(data)
{}
static void* ThreadRoutine(void* args) // 类内方法，隐含了第一个形参this，改成static并把args作为this传
{
(void)args; // 使用一下，避免编译器告警
Thread* ts = static_cast<Thread*>(args);
ts->_func(ts->_data);
return nullptr;
}
bool Start()
{
int n = pthread_create(&_tid, nullptr, ThreadRoutine, this);
if (n == 0)
{
_isrunning = true;
return true;
}
else
{
return false;
}
}
bool Join()
{
if (!_isrunning) return true;
int n = pthread_join(_tid, nullptr);
if (n == 0)
{
_isrunning = false;
return true;
}
return false;
}
std::string ThreadName()
{
return _threadname;
}
bool IsRunning()
{
return _isrunning;
}
~Thread() = default;
private:
pthread_t _tid;
std::string _threadname;
bool _isrunning;
func_t<T> _func;
T _data;
};
主文件：
- 创建一个线程
#include <iostream>
#include <stdio.h>
#include <unistd.h>
#include <vector>
#include "Thread.hpp"
std::string GetThreadName()
{
static int number = 1;
char name[64];
snprintf(name, sizeof(name), "Thread - %d", number++);
return name;
}
void Print(int num) // 传一个打印次数
{
while (true)
{
std::cout << "hello world：" << num-- << std::endl;
sleep(1);
}
}
int main()
{
Thread<int> t(GetThreadName(), Print, 10);
std::cout << "Is thread running? " << t.IsRunning() << std::endl;
t.Start();
std::cout << "Is thread running? " << t.IsRunning() << std::endl;
t.Join();
return 0;
}
- 模拟一个多线程抢票的情况：
#include <iostream>
#include <stdio.h>
#include <unistd.h>
#include <vector>
#include "Thread.hpp"
std::string GetThreadName()
{
static int number = 1;
char name[64];
snprintf(name, sizeof(name), "Thread - %d", number++);
return name;
}
// 不加锁的版本
int ticket = 10000;
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
void GetTicket(std::string name)
{
while (true)
{
if (ticket > 0)
{
usleep(100); // 充当抢票花费的时间
printf("%s get a ticket: %d \n", name.c_str(), ticket);
ticket--;
}
else
{
break;
}
}
}
int main()
{
std::string name1 = GetThreadName();
Thread<std::string> t1(name1, GetTicket, name1);
std::string name2 = GetThreadName();
Thread<std::string> t2(name2, GetTicket, name2);
std::string name3 = GetThreadName();
Thread<std::string> t3(name3, GetTicket, name3);
std::string name4 = GetThreadName();
Thread<std::string> t4(name4, GetTicket, name4);
t1.Start();
t2.Start();
t3.Start();
t4.Start();
t1.Join();
t2.Join();
t3.Join();
t4.Join();
}
为什么减到0了？？？我们设置的是
ticket > 0停止了啊！因为线程未加锁。
二、Linux线程互斥
进程线程间的互斥相关背景概念
- 临界资源：多线程执行流共享的公共资源就叫做临界资源
- 临界区：每个线程内部，访问临界资源的代码，就叫做临界区
- 互斥：任何时刻，互斥保证有且只有一个执行流进入临界区，访问临界资源，通常对临界资源起保护作用，串行访问！
- 原子性（后面讨论如何实现）：不会被任何调度机制打断的操作，该操作只有两态，要么完成，要么未完成
互斥量mutex
- 大部分情况，线程使用的数据都是局部变量，变量的地址空间在线程栈空间内，这种情况，变量归属单个线程，其他线程无法获得这种变量。
- 但有时候，很多变量都需要在线程间共享，这样的变量称为共享变量，可以通过数据的共享，完成线程之间的交互。
- 多个线程并发的操作共享变量，会带来一些问题。
为什么上面的抢票代码可能无法获得正确结果？（票数为负）
if语句判断条件为真以后，代码可以并发的切换到其他线程。usleep模拟漫长业务的过程，在这个漫长的业务过程中，可能有很多个线程会进入该代码段。--ticket操作本身就不是一个原子操作，因为-- 操作并不是原子操作，而是对应三条汇编指令：
load ：将共享变量ticket从内存加载到寄存器中
update : 更新寄存器里面的值，执行-1操作
store ：将新值，从寄存器写回共享变量ticket的内存地址400fc7: 8b 05 17 21 20 00 mov 0x202117(%rip),%eax # 6030e4 <ticket> 400fcd: 83 e8 01 sub $0x1,%eax 400fd0: 89 05 0e 21 20 00 mov %eax,0x20210e(%rip) # 6030e4 <ticket>
要解决以上问题，需要做到三点：
- 代码必须要有互斥行为：当代码进入临界区执行时，不允许其他线程进入该临界区。
- 如果多个线程同时要求执行临界区的代码，并且临界区没有线程在执行，那么只能允许一个线程进入该临界区。
- 如果线程不在临界区中执行，那么该线程不能阻止其他线程进入临界区。
要做到以上这三点，本质上就是需要一把锁。Linux上提供的这把锁叫互斥量。
互斥量的接口
1. 初始化互斥量
初始化互斥量有两种方法：
方法一：静态分配
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER
方法二：动态分配:
int pthread_mutex_init(pthread_mutex_t *restrict mutex,
const pthread_mutexattr_t *restrictattr);
参数：
mutex：要初始化的互斥量
attr：NULL
2. 销毁互斥量
销毁互斥量需要注意：
- 使用
PTHREAD_ MUTEX_ INITIALIZER初始化的互斥量不需要销毁 - 不要销毁一个已经加锁的互斥量
- 已经销毁的互斥量，要确保后面不会有线程再尝试加锁
int pthread_mutex_destroy(pthread_mutex_t *mutex)；
3. 互斥量加锁和解锁
int pthread_mutex_lock(pthread_mutex_t *mutex);
int pthread_mutex_unlock(pthread_mutex_t *mutex);
返回值:成功返回0,失败返回错误号
调用 pthread_mutex_lock 时，可能会遇到以下情况：
- 互斥量处于未锁状态，该函数会将互斥量锁定，同时返回成功
- 发起函数调用时，其他线程已经锁定互斥量，或者存在其他线程同时申请互斥量，但没有竞争到互斥量，那么
pthread_mutex_lock调用会陷入阻塞（执行流被挂起），等待互斥量解锁。
我们在上述的抢票系统中引入互斥量，每一个线程要进入临界区之前都必须先申请锁，只有申请到锁的线程才可以进入临界区对临界资源进行访问，并且当线程出临界区的时候需要释放锁，这样才能让其余要进入临界区的线程继续竞争锁。
改进的售票系统：
// 加锁版本
// 加锁：
// 1. 我们要尽可能给少的代码块加锁，加锁牺牲效率，提高安全性
// 2. 一般加锁，都是给临界区加锁
int ticket = 10000; // 全局共享
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER; // 全局锁
void GetTicket(std::string name)
{
while (true)
{
// 2. 是由程序员自己保证的！规则都必须先申请锁
// 3. 根据互斥的定义，任何时刻，只允许一个线程申请锁成功！
// 多个线程申请锁失败，失败的线程怎么办？在mutex上进行阻塞，本质就是等待！
pthread_mutex_lock(&mutex); // 1. 申请锁本身是安全的，原子的，为什么？因为申请锁的底层汇编指令只有一条
if (ticket > 0) // 4. 一个线程在临界区中访问临界资源的时候，可不可能发生切换？ 可能！
{
usleep(100); // 充当抢票花费的时间
printf("%s get a ticket: %d \n", name.c_str(), ticket);
ticket--;
pthread_mutex_unlock(&mutex);
}
else
{
pthread_mutex_unlock(&mutex);
break;
}
}
}
三、常见锁概念
死锁
死锁是指在一组进程中的各个进程均占有不会释放的资源，但因互相申请被其他进程所站用不会释放的资源而处于的一种永久等待状态。
单执行流可能产生死锁吗？
单执行流也有可能产生死锁，如果某一执行流连续申请了两次锁，那么此时该执行流就会被挂起。因为该执行流第一次申请锁的时候是申请成功的，但第二次申请锁时因为该锁已经被申请过了，于是申请失败导致被挂起直到该锁被释放时才会被唤醒，但是这个锁本来就在自己手上，自己现在处于被挂起的状态根本没有机会释放锁，所以该执行流将永远不会被唤醒，此时该执行流也就处于一种死锁的状态。例如，在下面的代码中我们让主线程创建的新线程连续申请了两次锁。
#include <stdio.h> #include <pthread.h> pthread_mutex_t mutex; void* Routine(void* arg) { pthread_mutex_lock(&mutex); pthread_mutex_lock(&mutex); pthread_exit((void*)0); } int main() { pthread_t tid; pthread_mutex_init(&mutex, NULL); pthread_create(&tid, NULL, Routine, NULL); pthread_join(tid, NULL); pthread_mutex_destroy(&mutex); return 0; }
运行代码，此时该程序实际就处于一种被挂起的状态。用
ps命令查看该进程时可以看到，该进程当前的状态是Sl+，其中的l实际上就是lock的意思，表示该进程当前处于一种死锁的状态。
例如，当某一个进程在被CPU调度时，该进程需要用到锁的资源，但是此时锁的资源正在被其他进程使用：
- 那么此时该进程的状态就会由R状态变为某种阻塞状态，比如S状态。并且该进程会被移出运行等待队列，被链接到等待锁的资源的资源等待队列，而CPU则继续调度运行等待队列中的下一个进程。
- 此后若还有进程需要用到这一个锁的资源，那么这些进程也都会被移出运行等待队列，依次链接到这个锁的资源等待队列当中。
- 直到使用锁的进程已经使用完毕，也就是锁的资源已经就绪，此时就会从锁的资源等待队列中唤醒一个进程，将该进程的状态由S状态改为R状态，并将其重新链接到运行等待队列，等到CPU再次调度该进程时，该进程就可以使用到锁的资源了。
锁本质就是一种软件资源，当我们申请锁时，锁当前可能并没有就绪，可能正在被其他线程所占用，此时当其他线程再来申请锁时，就会被放到这个锁的资源等待队列当中。
死锁四个必要条件
- 互斥条件：一个资源每次只能被一个执行流使用
- 请求与保持条件：一个执行流因请求资源而阻塞时，对已获得的资源保持不放
- 不剥夺条件：一个执行流已获得的资源，在末使用完之前，不能强行剥夺
- 循环等待条件：若干执行流之间形成一种头尾相接的循环等待资源的关系
这是死锁的四个必要条件，也就是说只有同时满足了这四个条件才可能产生死锁。
避免死锁的方法
- 破坏死锁的四个必要条件
- 加锁顺序一致
- 避免锁未释放的场景
- 资源一次性分配
避免死锁的算法
- 死锁检测算法（了解）
- 银行家算法（了解）
四、生产消费者模型
为何要使用生产者消费者模型？
生产者消费者模式就是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。这个阻塞队列就是用来给生产者和消费者解耦的。
三种关系：
- 生产者之间是什么关系？竞争 - 互斥
- 消费者之间是什么关系？竞争 - 互斥
- 生产和消费者之间是什么关系？互斥 && 同步
“321”原则：
- 3：三种关系
- 2：两种角色，生产者（1 or n），消费者（1 or n），他们可以是线程或进程
- 1：一个交易场所：内存空间
生产者消费者模型优点
基于阻塞队列的生产者消费者模型具备以下优点：
- 解耦
- 支持并发
- 支持忙闲不均
-
解耦：
- 阻塞队列作为生产者和消费者之间的缓冲区，使它们能够独立运行，不需要直接相互通信。
- 生产者和消费者只需与队列进行交互，而无需了解对方的存在或状态。
- 这种解耦性简化了系统的设计和维护，使得生产者和消费者之间的关系更加松散，降低了耦合度。
-
支持并发：
- 阻塞队列通常是线程安全的数据结构，能够支持多个生产者和消费者并发地对其进行读写操作。
- 多个线程可以同时向队列中添加数据或者从队列中取出数据，而不会出现数据不一致或竞争条件等并发问题。
-
支持忙闲不均：
- 当生产者和消费者的速度不匹配时，阻塞队列能够缓解生产者和消费者之间的压力差异。
- 当生产者生产速度过快或消费者消费速度过慢时，队列可以暂时存储数据，避免生产者因为没有消费者而被阻塞，或者消费者因为没有生产者而处于忙碌状态。
- 阻塞队列可以根据需求设置不同的容量，以调节生产者和消费者之间的速度差异，使系统能够更好地适应不同的负载情况。
如果我们在主函数中调用某一函数，那么我们必须等该函数体执行完后才继续执行主函数的后续代码，因此函数调用本质上是一种紧耦合。
对应到生产者消费者模型中，函数传参实际上就是生产者生产的过程，而执行函数体实际上就是消费者消费的过程，但生产者只负责生产数据，消费者只负责消费数据，在消费者消费期间生产者可以同时进行生产，因此生产者消费者模型本质是一种松耦合。
基于BlockingQueue的生产者消费者模型
BlockingQueue是什么
在多线程编程中阻塞队列(Blocking Queue)是一种常用于实现生产者和消费者模型的数据结构。其与普通的队列区别在于，当队列为空时，从队列获取元素的操作将会被阻塞，直到队列中被放入了元素；当队列满时，往队列里存放元素的操作也会被阻塞，直到有元素被从队列中取出（以上的操作都是基于不同的线程来说的，线程在对阻塞队列进程操作时会被阻塞）
C++ queue模拟阻塞队列的生产消费模型
为了方便理解，可以先以单生产者、单消费者为例进行实现，然后再增加生产者和消费者数量：
其中的BlockQueue就是生产者消费者模型当中的交易场所，我们可以用C++STL库当中的queue进行实现。
Makefile：
bq:Main.cc
g++ -o bq Main.cc -std=c++11 -lpthread
.PHONY:clean
clean:
rm -f bq;
主文件Main.cc：
#include "BlockQueue.hpp"
#include "LockGuard.hpp"
#include "Task.hpp"
#include <pthread.h>
#include <unistd.h>
#include <ctime>
void* consumer(void* args)
{
BlockQueue<Task>* bq = static_cast<BlockQueue<Task>*>(args);
while (true)
{
Task t;
// 1. 消费数据
bq->Pop(&t);
t();
std::cout << "productor data: " << t.PrintResult() << std::endl;
// 注意：消费者没有sleep！！！
}
return nullptr;
}
void* productor(void* args)
{
BlockQueue<Task>* bq = static_cast<BlockQueue<Task>*>(args);
while (true)
{
// 1. 有数据
// 生产前，你的任务从哪里来的呢？？？
int data1 = rand() % 10; // [1, 10] // 将来深刻理解生产消费，就要从这里入手，TODO
usleep(rand() % 123);
int data2 = rand() % 10; // [1, 10] // 将来深刻理解生产消费，就要从这里入手，TODO
usleep(rand() % 123);
char oper = operators[rand() % (operators.size())];
Task t(data1, data2, oper);
std::cout << "productor task: " << t.PrintTask() << std::endl;
// 2. 进行生产
bq->Push(t);
}
}
int main()
{
srand((uint16_t)time(NULL) ^ getpid() ^ pthread_self());
BlockQueue<Task>* bq = new BlockQueue<Task>();
pthread_t c[3], p[2]; // 生产者和消费者
pthread_create(&c[0], nullptr, consumer, bq);
pthread_create(&c[1], nullptr, productor, bq);
pthread_create(&c[2], nullptr, productor, bq);
pthread_create(&p[0], nullptr, consumer, bq);
pthread_create(&p[1], nullptr, productor, bq);
pthread_join(c[0], nullptr);
pthread_join(c[1], nullptr);
pthread_join(c[2], nullptr);
pthread_join(p[0], nullptr);
pthread_join(p[1], nullptr);
return 0;
}
阻塞队列的实现BlockQueue.hpp：
#pragma once
#include <iostream>
#include <queue>
#include <pthread.h>
#include "LockGuard.hpp"
const int defalutcap = 5;
template<class T>
class BlockQueue
{
public:
BlockQueue(int cap = defalutcap)
:_capacity(cap)
{
pthread_mutex_init(&_mutex, nullptr);
pthread_cond_init(&_p_cond, nullptr);
pthread_cond_init(&_c_cond, nullptr);
}
bool IsFull()
{
return _q.size() == _capacity;
}
bool IsEmpty()
{
return _q.size() == 0;
}
void Push(const T& in) // 生产者的
{
pthread_mutex_lock(&_mutex);
if (IsFull())
{
// 阻塞等待
pthread_cond_wait(&_p_cond, &_mutex);
}
_q.push(in);
pthread_cond_signal(&_c_cond);
pthread_mutex_unlock(&_mutex);
}
void Pop(T* out) // 消费者的
{
pthread_mutex_lock(&_mutex);
if (IsEmpty())
{
// 阻塞等待
pthread_cond_wait(&_c_cond, &_mutex);
}
*out = _q.front();
_q.pop();
pthread_cond_signal(&_p_cond);
pthread_mutex_unlock(&_mutex);
}
~BlockQueue()
{
pthread_mutex_destroy(&_mutex);
pthread_cond_destroy(&_p_cond);
pthread_cond_destroy(&_c_cond);
}
private:
std::queue<T> _q;
int _capacity; // _q.size() == _capacity
pthread_mutex_t _mutex;
pthread_cond_t _p_cond; // 给生产者的条件变量，有数据了来叫它
pthread_cond_t _c_cond; // 给消费者的条件变量，有空间了来叫它
int _consumer_water_line;
int _productor_water_line;
};
生产消费的任务类Task.hpp：
#pragma once
#include <iostream>
#include <string>
const int defaultvalue = 0;
enum
{
ok = 0,
div_zero = 1,
mod_zero = 2,
unknown = 3
};
const std::string operators = "+-*/%)(&";
class Task
{
public:
Task() = default;
~Task() = default;
Task(int x, int y, char op)
: data_x(x)
, data_y(y)
, _operator(op)
, result(defaultvalue)
, code(ok)
{}
void Run()
{
switch (_operator)
{
case '+':
result = data_x + data_y;
break;
case '-':
result = data_x - data_y;
break;
case '*':
result = data_x * data_y;
break;
case '/':
{
if (data_y == 0)
{
code = div_zero;
}
else
{
result = data_x / data_y;
}
}
break;
case '%':
{
if (data_y == 0)
{
code = mod_zero;
}
}
break;
default:
code = unknown;
break;
}
}
void operator()()
{
Run();
}
std::string PrintTask()
{
std::string s;
s = std::to_string(data_x);
s += _operator;
s += std::to_string(data_y);
s += "=?";
return s;
}
std::string PrintResult()
{
std::string s;
s = std::to_string(data_x);
s += _operator;
s += std::to_string(data_y);
s += "=";
s += std::to_string(result);
s += " [";
s += std::to_string(code);
s += "]";
return s;
}
private:
int data_x;
int data_y;
char _operator;
int result;
int code; // 结果码，0：结果可信 !0：结果不可信
};
锁的生命周期管理LockGuard.hpp：
#pragma once
#include <pthread.h>
class Mutex
{
public:
Mutex(pthread_mutex_t* lock)
:_lock(lock)
{}
void Lock()
{
pthread_mutex_lock(_lock);
}
void Unlock()
{
pthread_mutex_unlock(_lock);
}
~Mutex() = default;
private:
pthread_mutex_t* _lock;
};
class LockGuard
{
public:
LockGuard(pthread_mutex_t* lock)
:_mutex(lock)
{
_mutex.Lock();
}
~LockGuard()
{
_mutex.Unlock();
}
private:
Mutex _mutex;
};
相关说明：
- 阻塞队列是会被生产者和消费者同时访问的临界资源，因此我们需要用一把互斥锁将其保护起来。
- 生产者线程要向阻塞队列当中Push数据，前提是阻塞队列里面有空间，若阻塞队列已经满了，那么此时该生产者线程就需要进行等待，直到阻塞队列中有空间时再将其唤醒。
- 消费者线程要从阻塞队列当中Pop数据，前提是阻塞队列里面有数据，若阻塞队列为空，那么此时该消费者线程就需要进行等待，直到阻塞队列中有新的数据时再将其唤醒。
- 因此在这里我们需要用到两个条件变量，一个条件变量用来描述队列为空，另一个条件变量用来描述队列已满。当阻塞队列满了的时候，要进行生产的生产者线程就应该在full条件变量下进行等待；当阻塞队列为空的时候，要进行消费的消费者线程就应该在empty条件变量下进行等待。
- 不论是生产者线程还是消费者线程，它们都是先申请到锁进入临界区后再判断是否满足生产或消费条件的，如果对应条件不满足，那么对应线程就会被挂起。但此时该线程是拿着锁的，为了避免死锁问题，在调用pthread_cond_wait函数时就需要传入当前线程手中的互斥锁，此时当该线程被挂起时就会自动释放手中的互斥锁，而当该线程被唤醒时又会自动获取到该互斥锁。
- 当生产者生产完一个数据后，意味着阻塞队列当中至少有一个数据，而此时可能有消费者线程正在empty条件变量下进行等待，因此当生产者生产完数据后需要唤醒在empty条件变量下等待的消费者线程。
- 同样的，当消费者消费完一个数据后，意味着阻塞队列当中至少有一个空间，而此时可能有生产者线程正在full条件变量下进行等待，因此当消费者消费完数据后需要唤醒在full条件变量下等待的生产者线程。
生产者消费者步调一致
运行代码后我们可以看到生产者和消费者的执行步调是一致的：
![[AIGC] 在Spring Boot中指定请求体格式](https://img-blog.csdnimg.cn/direct/cd575d2000a34da69f729f18eddfa1ff.png)
![[Uni-app] 微信小程序的圆环进度条](https://img-blog.csdnimg.cn/direct/1fc0bcdd2d2e4a14bf64fc87509cf7dd.png)
