【Linux】线程同步与互斥
文章目录
- 1. 线程互斥
- 1.1 进程线程间的互斥相关背景概念
- 1.2 互斥量mutex
- 1.3 相关操作
- 1.4 互斥量实现原理
- 1.5 互斥量的封装
- 2. 线程同步
- 2.1 条件变量
- 2.2 生产者消费者模型
- 2.3 基于BlockingQueue的生产者消费者模型
- 2.4 信号量
- 2.5 基于环形队列的生产消费模型
- 3. 线程池
- 3.1 日志
- 3.2 线程池设计
- 4. 线程安全与重入问题
1. 线程互斥
1.1 进程线程间的互斥相关背景概念
- 临界资源：多线程执⾏流共享的资源就叫做临界资源
- 临界区：每个线程内部，访问临界资源的代码，就叫做临界区
- 互斥：任何时刻，互斥保证有且只有⼀个执⾏流进⼊临界区，访问临界资源，通常对临界资源起保护作⽤
- 原子性（后⾯讨论如何实现）：不会被任何调度机制打断的操作，该操作只有两态，要么完成， 要么未完成。
1.2 互斥量mutex
- 大部分情况，线程使用的数据都是局部变量，变量的地址空间在线程栈空间内，这种情况，变量归属单个线程，其他线程无法获得这种变量。
- 但有时候，很多变量都需要在线程间共享，这样的变量称为共享变量，可以通过数据的共享，完成线程之间的交互。
多个线程并发的操作共享变量，会带来⼀些问题：
在上方代码中，由于每个线程在运行时可能被切换(特别是正在访问共享资源时)，并且每个线程访问共享资源的代码不是原子的，所以会导致对资源的“重复”操作。
- -操作并不是原⼦操作，而是对应三条汇编指令：
- load ：将共享变量g_ticket从内存加载到寄存器中
- update : 更新寄存器里面的值，执行-1操作
- store ：将新值，从寄存器写回共享变量g_ticket的内存地址
要解决以上问题，需要做到三点：
- 代码必须要有互斥行为：当代码进⼊临界区执行时，不允许其他线程进⼊该临界区。
- 如果多个线程同时要求执行临界区的代码，并且临界区没有线程在执行，那么只能允许⼀个线程进⼊该临界区。
- 如果当前线程不在临界区中执行，那么该线程不能阻止其他线程进⼊临界区。
要做到这三点，本质上就是需要⼀把锁，Linux上提供的这把锁叫互斥量。
1.3 相关操作
在Linux中，互斥量是一个类型为pthread_mutex_t类型的变量。
- 锁的初始化与销毁
锁的初始化有两种方式：
- 静态分配
pthread_mutex_t mutex= PTHREAD_MUTEX_INITIALIZER
- 动态分配
锁的销毁：
销毁互斥量需要注意：
- 使用 PTHREAD_ MUTEX_ INITIALIZER 初始化的互斥量不需要销毁
- 不要销毁⼀个已经加锁的互斥量
- 已经销毁的互斥量，要确保后面不会有线程再尝试加锁
- 加锁与解锁的方法
lock是阻塞式的加锁；trylock若加锁失败，则会返回一个错误，下次什么时候再加锁，取决于程序员。
加了锁以后，对共享资源的访问就是安全的了。
调⽤ pthread_ lock 时，可能会遇到以下情况:
- 互斥量处于未锁状态，该函数会将互斥量锁定，同时返回成功
- 发起函数调⽤时，其他线程已经锁定互斥量，或者存在其他线程同时申请互斥量，但没有竞争到互斥量，那么
pthread_ lock调⽤会陷⼊阻塞(执⾏流被挂起)，等待互斥量解锁。
1.4 互斥量实现原理
上面的代码中，锁是全局的，它保护临界资源，可是谁来保护锁呢？
所以pthread_mutex_loack 与pthread_mutex_unloack必须是原子的！
- 经过上⾯的例⼦，⼤家已经意识到单纯的 i-- 或者 i++ 都不是原⼦的，有可能会有数据⼀致性问题
- 为了实现互斥锁操作，⼤多数体系结构都提供了swap或exchange指令，该指令的作⽤是把寄存器和内存单元的数据相交换，由于只有⼀条指令，保证了原⼦性，即使是多处理器平台，访问内存的总线周期也有先后，⼀个处理器上的交换指令执⾏时另⼀个处理器的交换指令只能等待总线周期。
下面是lock和unlock的伪代码分析：
1.5 互斥量的封装
#include <iostream>
using namespace std;
#include <pthread.h>
namespace MyMutexModule
{
class Mutex
{
public:
Mutex(const Mutex &) = delete;
const Mutex &operator=(const Mutex &) = delete;
Mutex()
{
pthread_mutex_init(&_mutex, nullptr);
}
~Mutex()
{
pthread_mutex_destroy(&_mutex);
}
void lock()
{
int n = pthread_mutex_lock(&_mutex);
if (n != 0)
{
cout << "lock fail" << endl;
exit(1);
}
}
void unlock()
{
int n = pthread_mutex_unlock(&_mutex);
if (n != 0)
{
cout << "unlock fail" << endl;
exit(1);
}
}
pthread_mutex_t *getAddr()
{
return &_mutex;
}
private:
pthread_mutex_t _mutex;
};
//采用RAII风格，进行锁的管理
class MutexGuard
{
public:
MutexGuard(Mutex &mutex)
: _mutex(mutex)
{
_mutex.lock();
}
~MutexGuard()
{
_mutex.unlock();
}
private:
Mutex &_mutex;
};
}
2. 线程同步
在上方抢票的案例中，我们发现总是会出现同一个线程一直在抢票的情况。这种情况可以但是不合乎情理，为了解决该情况，引入了线程同步的概念。
线程同步：在保证数据安全的前提下，让线程能够按照某种特定的顺序访问临界资源，从而有效避免饥饿问题
2.1 条件变量
条件变量就是用来实现线程同步效果的，内部要维护一个等待队列。
当一个线程到来时，它发现没有资源可访问，那它就要去条件变量下等，直至条件满足，它被唤醒。
- 条件变量是一个
pthread_cond_t类型的 - 初始化与销毁
- 等待
为什么 pthread_cond_wait 需要互斥量?- - 后面说
- 唤醒
测试：
此时，利用条件变量，就可以实现线程同步的问题了。
封装：
#pragma once
#include<pthread.h>
using namespace std;
#include"MyMute.hpp"
using namespace MutexModule;
namespace MyCondModule
{
class Cond
{
public:
Cond()
{
pthread_cond_init(&_cond,nullptr);
}
//要让线程释放曾经拥有的锁！！
void wait(Mutex& mutex)
{
int n = pthread_cond_wait(&_cond,mutex.getAddr());
(void)n;
}
void signal()
{
int n = pthread_cond_signal(&_cond);
(void)n;
}
void signalAll()
{
int n = pthread_cond_broadcast(&_cond);
(void)n;
}
~Cond()
{
pthread_cond_destroy(&_cond);
}
private:
pthread_cond_t _cond;
};
}
2.2 生产者消费者模型
生产者消费者模式：就是通过⼀个容器来解决生产者和消费者的强耦合问题，处理多个生产者线程和多个消费者线程之间的协作问题。
生产者和消费者彼此之间不直接通讯，而是通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。
这个阻塞队列就是用来给生产者和消费者解耦的。
利用 “321”原则 牢记生产者消费者模式
- 生产者消费者模式是多线程并发的一种模式，它有3 种关系，即生产者与生产者（
互斥）、消费者与消费者（互斥）、消费者与生产者（互斥且同步）- 有2 种角色，生产则与消费者，通常由线程承担
有1 种特定的数据结构提供的缓冲区，充当交易场所
2.3 基于BlockingQueue的生产者消费者模型
在多线程编程中阻塞队列(Blocking Queue)是⼀种常用于实现生产者和消费者模型的数据结构。
其与普通的队列区别在于：
- 当队列为空时，从队列获取元素的操作将会被阻塞，直到队列中被放⼊了元素；
- 当队列为满时，往队列⾥存放元素的操作也会被阻塞，直到有元素被从队列中取出
实现：
#pragma once
#include <iostream>
#include <pthread.h>
#include <queue>
using namespace std;
namespace BlockQueueModule
{
static const int gcap = 10;
template <class T>
class BlockQueue
{
private:
bool isFull()
{
return _q.size() == _cap;
}
bool isEmpty()
{
return _q.empty();
}
public:
BlockQueue(int cap = 10)
: _cap(cap)
,_cwait_num(0)
,_pwait_num(0)
{
pthread_mutex_init(&_mutex,nullptr);
pthread_cond_init(&consumer_cond,nullptr);
pthread_cond_init(&productor_cond,nullptr);
}
void Push(const T &data)
{
pthread_mutex_lock(&_mutex);
while(isFull()) //为避免造成伪唤醒问题，应使用while循环判断
{
_pwait_num++;
cout << "生产者进入等待..." << endl;
//等待时，持有锁去等，会造成死锁问题；
//因此wait内部会先释放锁，被唤醒后需要再次申请锁
pthread_cond_wait(&productor_cond,&_mutex);
cout << "生产者被唤醒..." << endl;
_pwait_num--;
}
//不满 || 被唤醒，一定能生产！
_q.push(data);
//有资源能消费，唤醒消费者★★★★
if(_cwait_num)
{
cout << "唤醒消费者..." << endl;
pthread_cond_signal(&consumer_cond);
//多生产多消费时，若在解锁之后唤醒被唤醒的线程未竞争过其它线程，
//其它线程将资源使用完了，当前线程才被唤醒，然后去错误的使用资源，会造成伪唤醒问题。
}
pthread_mutex_unlock(&_mutex);
}
void Pop(T *out)
{
pthread_mutex_lock(&_mutex);
//若空，消费者应去其条件变量下去等
while(isEmpty()) //为避免造成伪唤醒问题，应使用while循环判断
{
_cwait_num++;
cout << "消费者进入等待..." << endl;
pthread_cond_wait(&consumer_cond,&_mutex);
cout << "消费者被唤醒..." << endl;
_cwait_num--;
}
//不空 || 被唤醒，一定能消费！
*out = _q.front();
_q.pop();
//没有有资源能消费，唤醒生产者
if(_pwait_num)
{
cout << "唤醒生产者..." << endl;
pthread_cond_signal(&productor_cond);
}
pthread_mutex_unlock(&_mutex);
}
~BlockQueue()
{
pthread_mutex_destroy(&_mutex);
pthread_cond_destroy(&consumer_cond);
pthread_cond_destroy(&productor_cond);
}
private:
queue<T> _q; // 充当交易场所
int _cap; // 阻塞队列的最大容量
pthread_mutex_t _mutex; // 互斥
pthread_cond_t consumer_cond; // 消费者的条件变量
pthread_cond_t productor_cond; // 生产者的条件变量
int _cwait_num;
int _pwait_num;
};
}
为什么 pthread_cond_wait 需要传递互斥量呢？?
- 条件等待是线程间同步的⼀种⼿段，如果只有⼀个线程，条件不满⾜，⼀直等下去都不会满⾜，所以必须要有⼀个线程通过某些操作，改变共享变量，使原先不满⾜的条件变得满⾜，并且友好的通知等待在条件变量上的线程。
- 在上面的代码中，由于生产者与消费者是互斥的，那他们俩用的就是同一把锁；如果某一方 pthread_cond_wait时，是持有锁去等的话，另一方就永远无法获得锁，也就无法改变共享资源，从而无法改变条件变量，就会造成死锁得局面。
- 所以，在一方进行等待时，需要先将持有的锁释放，让另一方能够有机会去改变条件变量，从而唤醒等待的一方；等待的一方被唤醒后，需再次持有锁对共享资源进行操作。
在我们上面所实现的生产者消费者模型中，生产者与消费者是互斥的，也就是说在任何一个时刻，只能有一方访问共享资源，那不就是串行执行了吗？它高效在哪里呢？
生产者向交易场所中生产数据，消费者从交易场所中获取数据确实是串行的。
可是生产者的数据从哪里来呢？消费者获取的数据又如何处理呢?
2.4 信号量
信号量的使用主要是用来保护共享资源，使得资源在一个时刻只有一个进程（线程）所拥有。
POSIX信号量和SystemV信号量作⽤相同，但POSIX可以⽤于线程间同步。
在我们前面所写的代码中，都是对资源进行整体使用的，生产者用消费者就不能用；可以使用信号量对资源进行拆分，使得生产者与消费者某些时候可以同时使用共享资源。
相关接口：
- 初始化
参数：
- pshared：0表示线程间共享，非零表示进程间共享
- value：信号量初始值(信号量的个数)
- 销毁
- 发布信号量
功能：发布信号量，表⽰资源使⽤完毕，可以归还资源了。将信号量值加1。
- 等待信号量
功能：等待信号量，会将信号量的值减1
封装：
#pragma once
#include <semaphore.h>
using namespace std;
namespace MySemModule
{
const int g_value = 10;
class Sem
{
public:
Sem(int value = g_value)
:_value_num(value)
{
sem_init(&_sem,0,_value_num);
}
void P()
{
int n = sem_wait(&_sem);
(void)n;
}
void V()
{
int n = sem_post(&_sem);
(void)n;
}
~Sem()
{
sem_destroy(&_sem);
}
private:
sem_t _sem;
int _value_num;
};
} // namespace MySemModule
2.5 基于环形队列的生产消费模型
#pragma once
#include <semaphore.h>
#include <iostream>
#include <pthread.h>
using namespace std;
#include <vector>
#include "MyMute.hpp"
#include "MySem.hpp"
namespace RingBufferModule
{
using namespace MyMutexModule;
using namespace MySemModule;
const int g_default_cap = 10;
template <class T>
class RingBuffer
{
public:
RingBuffer(int cap = g_default_cap)
: _ring(cap)
, _cap(cap)
, _c_head(0)
, _p_tail(0)
, _space_sem(cap) // 初始化空间信号量
, _data_sem(0) // 初始化数据信号量
{
// sem_init(&_data_sem, 0, 0); // 初始化数据信号量
// sem_init(&_space_sem, 0, _cap); // 初始化空间信号量
}
void Push(T &data)
{
// 1. 先获取信号量 -->有，获取成功；没有，阻塞
// sem_wait(&_space_sem);
_space_sem.P();
// 多生产者时，不上锁，会造成空间数据的丢失（覆盖），可能会使用同一个_p_tail
{
LockGuard lockguard(_space_lock);
_ring[_p_tail] = data;
_p_tail++;
_p_tail %= _cap;
}
// sem_post(&_data_sem); // 数据信号量增加
_data_sem.V();
}
void Pop(T *out)
{
// sem_wait(&_data_sem);
_data_sem.P();
// 多消费者时，不上锁，会造成空间数据的重复消费,可能会使用同一个_c_head
{
LockGuard lockguard(_data_lock);
*out = _ring[_c_head];
_c_head++;
_c_head %= _cap;
}
_space_sem.V();
// sem_post(&_space_sem); // 空间信号量增加
}
~RingBuffer()
{
// sem_destroy(&_data_sem);
// sem_destroy(&_space_sem);
}
private:
vector<T> _ring; // 环，临界资源
int _cap; // 总容量
int _c_head;
int _p_tail;
Sem _data_sem; // 数据信号量
Sem _space_sem; // 空间信号量
Mutex _space_lock; // 空间锁
Mutex _data_lock; // 数据锁
};
} // namespace RingBufferModule
3. 线程池
下⾯开始，我们结合我们之前所做的所有封装，进⾏⼀个线程池的设计。在写之前，我们要做如下准备
- 准备线程的封装 （已有）
- 准备锁和条件变量的封装 （已有）
- 引入日志，对线程进行封装
3.1 日志
计算机中的日志是记录系统和软件运行中发生事件的文件，主要作用是监控运行状态、记录异常信息，帮助快速定位问题并⽀持程序员进行问题修复。它是系统维护、故障排查和安全管理的重要⼯具。
日志格式以下几个指标是必须得有的：
- 时间戳
- 日志等级
- 日志内容
以下几个指标是可选的
- 文件名
- 行号
- 进程，线程相关id信息等
日志有现成的解决放案，如：spdlog、glog、Boost.Log、Log4cxx等等，我们依旧采用自定义日志的方式。
这里我们采用设计模式，策略模式（基类规定方法，子类去重写）来进行日志的设计:
- 控制台日志策略
- 文件日志策略
// 基类
class LogStrategy
{
public:
virtual ~LogStrategy() = default;
virtual void FlushLog(const std::string &message) = 0; // 派生类必须重写
};
// 控制台 派生类
class ConsoleStrategy : public LogStrategy
{
public:
virtual ~ConsoleStrategy() override
{
// std::cout << "~ConsoleStrategy" << std::endl;
}
virtual void FlushLog(const std::string &message) override // 派生类重写基类的虚函数
{
LockGuard LockGuard(_mutex);
std::cerr << message << std::endl;
}
private:
Mutex _mutex; // 显示器也是临界资源，保证输出线程的安全
};
// 文件 派生类
class FileStrategy : public LogStrategy
{
public:
FileStrategy(const std::string logPaht = defaultLogPath, const std::string logFileName = defaulFiletName)
: _logPath(defaultLogPath), _logFileName(defaulFiletName)
{
LockGuard LockGuard(_mutex);
// filesystem c++17
if (std::filesystem::exists(_logPath)) // 检测当前路径是否存在
return;
try
{
std::filesystem::create_directories(_logPath);
}
catch (const std::filesystem::filesystem_error &e)
{
std::cerr << e.what() << std::endl;
}
}
virtual ~FileStrategy() override
{
std::cout << "~FileStrategy" << std::endl;
}
virtual void FlushLog(const std::string &message) override // 派生类重写基类的虚函数
{
LockGuard lockguard(_mutex);
std::string filename = _logPath + _logFileName;
// 使用ofstream打开文件filename,以append方式
std::ofstream out(filename.c_str(), std::ofstream::app);
if (!out.is_open())
{
std::cerr << filename << " 文件打开失败" << std::endl;
return; // 打开失败
}
out << message << "\n"; // 向文件中写
}
private:
std::string _logPath;
std::string _logFileName;
Mutex _mutex; // 文件也是临界资源，保证输出线程的安全
};
具体的日志类：
注意点：
- 内部类使用外部类，内部类中必须是外部类的引用。
- 引用必须在初始化列表内初始化
class Logger
{
private:
std::unique_ptr<LogStrategy> _strategy; // 日志的写入策略
public:
Logger()
{
UseConsoleStrategy(); // 默认使用控制台策略
}
~Logger()
{
}
void UseConsoleStrategy()
{
_strategy = std::make_unique<ConsoleStrategy>();
}
void UseFileStrategy()
{
_strategy = std::make_unique<FileStrategy>();
}
// 内部类
//[可读性很好的时间] [⽇志等级] [进程pid] [打印对应⽇志的⽂件名][⾏号] - 消息内容（⽀持可变参数）
class LogMessage
{
private:
std::string _cur_time; // 日志写入时间
LogGrade _grade; // 日志等级
pid_t _pid; // 进程pid
std::string _fileName; // 文件名
int _lineNum; // 行号
Logger &_logger; // 方便进行日志的刷新，一定是引用
std::string _completeMessage; // 完整的信息
public:
LogMessage(LogGrade grade, std::string filename, int lineNum, Logger &logger)
: _cur_time(GetCurTime())
, _grade(grade), _pid(getpid())
, _fileName(filename)
, _lineNum(lineNum)
, _logger(logger)
{
// stringstream不允许拷⻉,所以这⾥就当做格式化功能使⽤
std::stringstream ssbuffer;
ssbuffer << "[" << _cur_time << "] "
<< "[" << GradeToString(_grade) << "] "
<< "[" << _pid << "] "
<< "[" << _fileName << "] "
<< "[" << _lineNum << "] "
<< " - ";
_completeMessage = ssbuffer.str(); // 信息格式以初始化完成
}
// LogMessage(const LogMessage& lg)
// :_logger(lg._logger)
// {
// _cur_time = lg._cur_time;
// _grade = lg._grade;
// _pid = lg._pid;
// _fileName = lg._fileName;
// _lineNum = lg._lineNum;
// cout <<"拷贝构造:LogMessage(const LogMessage& lg)" << endl;
// }
// RAII⻛格，析构的时候进⾏⽇志持久化，采⽤指定的策略
~LogMessage()
{
if (_logger._strategy)
{
_logger._strategy->FlushLog(_completeMessage);
cout <<"~LogMessage" << endl;
}
}
// 为支持连续的 LogMessage << "11" << "222";
template <class T>
LogMessage &operator<<(const T &message)
{
stringstream ssbuffer;
ssbuffer << message;
_completeMessage += ssbuffer.str(); // 每次将输入信息添加到日子信息后
return *this;
}
}; // class LogMessage
// 故意拷⻉，形成LogMessage临时对象，临时对象内包含独⽴⽇志数据,后续在被<<时，会被持续引⽤，
// 直到完成输⼊,才会⾃动析构临时LogMessage，⾄此也完成了⽇志的显⽰或者刷新
// 未来采⽤宏替换，进⾏⽂件名和代码⾏数的获取
public:
LogMessage operator()(LogGrade grade, const std::string &filename, int lineNum)
{
return LogMessage(grade, filename, lineNum, *this);
//传值返回，若没有对象接收，拷贝构造会被编译器优化掉！！！！！！！
}
}; // class Logger
// 定义全局的Logger 对象
Logger logger;
// 使⽤宏，可以进⾏代码插⼊，⽅便随时获取⽂件名和⾏号
#define LOG(grade) logger(grade, __FILE__, __LINE__)
// 提供选择使⽤何种⽇志策略的⽅法
#define ENABLE_CONSOLE_LOG_STRATEGY() logger.UseConsoleStrategy()
#define ENABLE_FILE_LOG_STRATEGY() logger.UseFileStrategy()
3.2 线程池设计
线程池：⼀种线程使用模式。
线程过多会带来调度开销，进而影响缓存局部性和整体性能。而线程池维护着多个线程，等待着监督管理者分配可并发执行的任务。这避免了在处理短时间任务时创建与销毁线程的代价。
-
线程池的应用场景：
- 需要大量的线程来完成任务，且完成任务的时间比较短。 比如WEB服务器完成网页请求这样的任务，使用线程池技术是非常合适的。因为单个任务小，而任务数量巨大，你可以想象⼀个热门网站的点击次数。 但对于长时间的任务，比如⼀个Telnet连接请求，线程池的优点就不明显了。因为Telnet会话时间比线程的创建时间大多了。
- 对性能要求苛刻的应用，比如要求服务器迅速响应客户请求。
-
线程池的种类
- 创建固定数量线程池，循环从任务队列中获取任务对象，获取到任务对象后，执⾏任务对象中的任务接口
- 浮动线程池，其他同上
在实现线程池时，我们全部都使用自己封装过的接口。
#include <iostream>
#include <memory>
#include <queue>
#include <vector>
#include "Log.hpp"
#include "MyMute.hpp"
#include "myCond.hpp"
#include "Thread.hpp"
namespace ThreadPoolModule
{
using namespace MyCondModule;
using namespace MyLogModule;
using namespace MyMutexModule;
using namespace MyThreadModule;
static const int default_count = 10;
using thread_t = shared_ptr<MyThread>;
void DefaultTest()
{
while (true)
{
LOG(NORMAL) << "测试线程的执行方法";
sleep(1);
}
}
template <class T>
class ThreadPool
{
private:
queue<T> _task_q; // 任务队列
vector<thread_t> _threads; // 管理线程的结构
int _threadCount; // 线程的数量
int _wait_num;
Mutex _lock;
Cond _cond;
bool _isRunning;
bool isEmpty()
{
return _task_q.empty();
}
void HandleTask(std::string name)
{
LOG(NORMAL) << name << "进入HandleTask方法";
// 线程要一直做该任务
while (true)
{
// 1. 拿任务
T t;
{
LockGuard lockguard(_lock);
//任务队列为空，线程池没退，则进程必须等
while (isEmpty() && _isRunning)
{
_wait_num++;
_cond.wait(_lock);
_wait_num--;
}
//任务队列为空，线程池退了，则进程自己退出
if(isEmpty() && !_isRunning)
break;
//任务队列不为空，线程池退/没退，线程都执行任务完任务
t = _task_q.front();
_task_q.pop();
}
// 2. 处理任务
t(name);
}
}
public:
ThreadPool(int count = default_count)
: _threadCount(count), _wait_num(0), _isRunning(false)
{
// 构建线程对象
for (int i = 0; i < _threadCount; i++)
{
// 构造MyThread时，将HandleTask方法与其绑定，传递this指针是为了访问它。
//_threads.push_back(make_shared<MyThread>(bind(&ThreadPool::HandleTask, this, std::placeholders::_1)));
_threads.push_back(make_shared<MyThread>([this](std::string name)
{ this->HandleTask(name); }));
LOG(NORMAL) << "构建线程对象" << _threads.back()->name() << "...成功";
}
}
~ThreadPool()
{
}
void Start()
{
if (_isRunning)
return;
_isRunning = true; // 线程池运行标记
// 启动线程
for (auto &thread_ptr : _threads)
{
thread_ptr->create();
LOG(NORMAL) << "启动线程" << thread_ptr->name() << "...成功";
}
}
void Wait()
{
// 等待线程
for (auto &thread_ptr : _threads)
{
thread_ptr->join();
LOG(NORMAL) << "等待线程" << thread_ptr->name() << "...成功";
}
}
void Stop()
{
LockGuard lockguard(_lock);
if (_isRunning)
{
_isRunning = false;
// 1.不能再次放任务了--标记位
// 2.让线程自己结束
if(_wait_num)
_cond.signalAll(); //唤醒所有线程
// 3.历史任务都要被执行完了
}
LOG(NORMAL) << "线程池stop..";
}
void Push(T in)
{
LockGuard lockguard(_lock);
if (!_isRunning)
return;
_task_q.push(in);
if (_wait_num)
_cond.signal();
}
};
} // namespace ThreadPoolModule
4. 线程安全与重入问题
线程安全：就是多个线程在访问共享资源时，能够正确地执行，不会相互干扰或破坏彼此的执行结果。
⼀般而言，多个线程并发同⼀段只有局部变量的代码时，不会出现不同的结果。但是对全局变量或者静态变量进⾏操作，并且没有锁保护的情况下，容易出现该问题。
函数是可重入的，那就是线程安全的!!!
线程安全函数：若线程执行的函数是线程安全的函数，该函数叫做线程安全函数。
可重入函数是线程安全函数的⼀种。
注意：
- 线程安全不⼀定是可重入的，而可重⼊函数则⼀定是线程安全的。
如果对临界资源的访问加上锁，则这个函数是线程安全的；但如果这个重入函数在锁还未释放之前再次进入，则会产⽣死锁，因此是不可重入的（例如递归、信号导致⼀个执行流重复进⼊函数）
我们上面所写的线程池还存在一些问题，例如线程池的个数不是想创建多少就创建多少的，这还取决于OS；为了避免线程安全问题，还需要对线程池的创建进行加锁。
因此我们可使用单例模式，设计一个只允许创建一个的线程池，并对其加锁。
STL中的容器是否是线程安全的?
不是。
- 原因是，STL 的设计初衷是将性能挖掘到极致，而⼀旦涉及到加锁保证线程安全,，会对性能造成巨⼤的影响。而且对于不同的容器,，加锁方式的不同,，性能可能也不同(例如hash表的锁表和锁桶)，因此 STL 默认不是线程安全。
- 如果需要在多线程环境下使用, 往往需要调⽤者自行保证线程安全.
智能指针是否是线程安全的?
- 对于 unique_ptr, 由于只是在当前代码块范围内⽣效, 因此不涉及线程安全问题。
- 对于 shared_ptr，多个对象需要共⽤⼀个引⽤计数变量， 所以会存在线程安全问题。但是标准库实现的时候考虑到了这个问题, 基于原⼦操作(CAS)的⽅式保证shared_ptr 能够⾼效，原⼦的操作引用计数。
- 注意：shared_ptr是线程安全的，但是使用其对某些资源进行操作不是，需要操作者自行保证。
