1线程互斥1进程线程间的互斥相关背景以及概念临界资源多线程执⾏流共享的资源就叫做临界资源临界区每个线程内部访问临界资源的代码就叫做临界区互斥任何时刻互斥保证有且只有⼀个执⾏流进⼊临界区访问临界资源通常对临界资源起保护作⽤原⼦性后⾯讨论如何实现不会被任何调度机制打断的操作该操作只有两态要么完成要么未完成2互斥量mutex⼤部分情况线程使⽤的数据都是局部变量变量的地址空间在线程栈空间内这种情况变量归属单个线程其他线程⽆法获得这种变量。但有时候很多变量都需要在线程间共享这样的变量称为共享变量可以通过数据的共享完成线程之间的交互。多个线程并发的操作共享变量会带来⼀些问题。#include stdio.h #include pthread.h #include unistd.h // 1. 定义临界资源总票数假设初始有10张票 int tickets 10; // 2. 定义互斥锁全局变量所有线程可访问 pthread_mutex_t mutex; // 线程函数黄牛抢票逻辑 void *scalper_grab_ticket(void *arg) { // 获取当前线程的编号区分不同黄牛 int scalper_id *(int *)arg; while (1) { // 3. 进入临界区前加锁保证同一时间只有一个线程操作tickets pthread_mutex_lock(mutex); // 临界区访问/修改临界资源tickets if (tickets 0) { // 模拟抢票耗时比如联网验证、生成订单 usleep(100000); // 休眠0.1秒 printf(黄牛%d抢到了第%d张票\n, scalper_id, tickets); tickets--; // 票数减1 } else { // 票已抢完解锁后退出循环 pthread_mutex_unlock(mutex); break; } // 4. 离开临界区后解锁释放互斥锁让其他线程可以抢票 pthread_mutex_unlock(mutex); // 模拟黄牛抢票间隔非临界区无需加锁 usleep(50000); // 休眠0.05秒 } printf(黄牛%d票抢完了收工\n, scalper_id); pthread_exit(NULL); // 退出当前线程 } int main() { // 定义5个黄牛线程可调整数量 pthread_t scalper_threads[5]; int scalper_ids[5] {1, 2, 3, 4, 5}; // 黄牛编号 // 初始化互斥锁默认属性 pthread_mutex_init(mutex, NULL); // 创建5个黄牛线程 for (int i 0; i 5; i) { pthread_create(scalper_threads[i], NULL, scalper_grab_ticket, scalper_ids[i]); } // 等待所有黄牛线程执行完毕 for (int i 0; i 5; i) { pthread_join(scalper_threads[i], NULL); } // 销毁互斥锁释放资源 pthread_mutex_destroy(mutex); printf(所有票已抢完程序结束\n); return 0; }if语句判断条件为真以后代码可以并发的切换到其他线程usleep这个模拟漫⻓业务的过程在这个漫⻓的业务过程中可能有很多个线程会进⼊该代码段--ticket操作本⾝就不是⼀个原⼦操作互斥量的接口初始化互斥量初始化互斥量有两种⽅法1静态分配pthread_mutex_t mutex PTHREAD_MUTEX_INITIALIZER2动态分配int pthread_mutex_init(pthread_mutex_t *restrict mutex, const pthread_mutexattr_t *restrict attr); /*参数 mutex要初始化的互斥量 attrNULL*/销毁互斥量销毁互斥量需要注意使⽤PTHREAD_ MUTEX_ INITIALIZER初始化的互斥量不需要销毁不要销毁⼀个已经加锁的互斥量已经销毁的互斥量要确保后⾯不会有线程再尝试加锁int pthread_mutex_destroy(pthread_mutex_t *mutex);互斥量加锁和解锁int pthread_mutex_lock(pthread_mutex_t *mutex); int pthread_mutex_unlock(pthread_mutex_t *mutex); //返回值:成功返回0,失败返回错误号调⽤pthread_ lock时可能会遇到以下情况:互斥量处于未锁状态该函数会将互斥量锁定同时返回成功发起函数调⽤时其他线程已经锁定互斥量或者存在其他线程同时申请互斥量但没有竞争到互斥量那么pthread_ lock调⽤会陷⼊阻塞(执⾏流被挂起)等待互斥量解锁。3:互斥量实现原理探究经过上⾯的例⼦⼤家已经意识到单纯的i或者i都不是原⼦的有可能会有数据⼀致性问题为了实现互斥锁操作,⼤多数体系结构都提供了swap或exchange指令,该指令的作⽤是把寄存器和 内存单元的数据相交换,由于只有⼀条指令,保证了原⼦性,即使是多处理器平台,访问内存的 总线周 期也有先后,⼀个处理器上的交换指令执⾏时另⼀个处理器的交换指令只能等待总线周期。 现在我们把lock和unlock的伪代码改⼀下2线程同步1条件变量当⼀个线程互斥地访问某个变量时它可能发现在其它线程改变状态之前它什么也做不了。例如⼀个线程访问队列时发现队列为空它只能等待只到其它线程将⼀个节点添加到队列中。这种情况就需要⽤到条件变量。2同步概念和竞态条件同步在保证数据安全的前提下让线程能够按照某种特定的顺序访问临界资源从⽽有效避免 饥饿问题叫做同步竞态条件因为时序问题⽽导致程序异常我们称之为竞态条件。在线程场景下这种问题也不难理解3条件变量函数初始化int pthread_cond_init(pthread_cond_t *restrict cond,const pthread_condattr_t*restrictattr); /*参数 cond要初始化的条件变量 attrNULL*/销毁int pthread_cond_destroy(pthread_cond_t *cond)等待条件满足int pthread_cond_wait(pthread_cond_t *restrict cond,pthread_mutex_t *restrictmutex); /*参数 cond要在这个条件变量上等待 mutex互斥量后⾯详细解释*/唤醒等待int pthread_cond_broadcast(pthread_cond_t *cond); int pthread_cond_signal(pthread_cond_t *cond);4生产消费模型1为何要使用生产者消费模型⽣产者消费者模式就是通过⼀个容器来解决⽣产者和消费者的强耦合问题。⽣产者和消费者彼此之间不直接通讯⽽通过阻塞队列来进⾏通讯所以⽣产者⽣产完数据之后不⽤等待消费者处理直接扔给阻塞队列消费者不找⽣产者要数据⽽是直接从阻塞队⾥取阻塞队列就相当于⼀个缓冲区平衡了⽣产者和消费者的处理能⼒。这个阻塞队列就是⽤来⽣产者和消费者解耦的。2生产者消费模型的优点解耦支持并发支持忙闲不均5基于BlockingQueue的生产消费者模型1BlockingQueue在多线程编程中阻塞队列(Blocking Queue)是⼀种常⽤于实现⽣产者和消费者模型的数据结构。其与普通的队列区别在于当队列为空时从队列获取元素的操作将会被阻塞直到队列中被放⼊了元素当队列满时往队列⾥存放元素的操作也会被阻塞直到有元素被从队列中取出(以上的操作都是基于不同的线程来说的线程在对阻塞队列进程操作时会被阻塞)2Cqueue模拟阻塞队列的生产消费模型单生产单消费//BlockQueue.hpp #ifndef __BLOCK_QUEUE_H #define __BLOCK_QUEUE_H #include iostream #include queue #include pthread.h #include unistd.h #include cstdlib #include ctime const int defaultcap 5; template class T class BlockQueue { public: BlockQueue(int cap defaultcap) : _cap(cap) { pthread_mutex_init(_mutex, nullptr); pthread_cond_init(_consumer_cond, nullptr); pthread_cond_init(_productor_cond, nullptr); // _blockqueue_high_water _cap * 2 / 3; // _blockqueue_low_water _cap * 1 / 3; sleep_consumer_num 0; sleep_productor_num 0; } void Enqueue(T in) { pthread_mutex_lock(_mutex); if (_dp.size() _cap) { sleep_productor_num; pthread_cond_wait(_productor_cond, _mutex); sleep_productor_num--; } _dp.push(in); // if (_dp.size() _blockqueue_high_water) if (sleep_consumer_num 0) pthread_cond_signal(_consumer_cond); pthread_mutex_unlock(_mutex); } void pop(T *out) // consumer { pthread_mutex_lock(_mutex); if (_dp.empty()) { sleep_consumer_num; pthread_cond_wait(_consumer_cond, _mutex); sleep_consumer_num--; } *out _dp.front(); _dp.pop(); // if(_dp.size()_blockqueue_low_water) if (sleep_productor_num 0) 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: std::queueT _dp; int _cap; pthread_mutex_t _mutex; pthread_cond_t _consumer_cond; pthread_cond_t _productor_cond; // int _blockqueue_low_water; // int _blockqueue_high_water; int sleep_productor_num; int sleep_consumer_num; }; #endif6:为什么pthread_cond_wait需要互斥量条件等待是线程间同步的⼀种⼿段如果只有⼀个线程条件不满⾜⼀直等下去都不会满⾜所以必须要有⼀个线程通过某些操作改变共享变量使原先不满⾜的条件变得满⾜并且友好的通知等待在条件变量上的线程。条件不会⽆缘⽆故的突然变得满⾜了必然会牵扯到共享数据的变化。所以⼀定要⽤互斥锁来保护。没有互斥锁就⽆法安全的获取和修改共享数据。、按照上⾯的说法我们设计出如下的代码先上锁发现条件不满⾜解锁然后等待在条件变量上不就⾏了如下代码:// 错误的设计 pthread_mutex_lock(mutex); while (condition_is_false) { pthread_mutex_unlock(mutex); // 解锁之后等待之前条件可能已经满⾜信号已经发出但是该信号可能被错过 pthread_cond_wait(cond); pthread_mutex_lock(mutex); } pthread_mutex_unlock(mutex);由于解锁和等待不是原⼦操作。调⽤解锁之后pthread_cond_wait之前如果已经有其他线程获取到互斥量摒弃条件满⾜发送了信号那么 pthread_cond_wait将错过这个信号可能会导致线程永远阻塞在这个 pthread_cond_wait。所以解锁和等待必须是⼀个原⼦操作。int pthread_cond_wait(pthread_cond_ t *cond,pthread_mutex_ t *mutex); 进⼊该函数后会去看条件量等于0不等于就把互斥量变成1直到cond_ wait返回把条件量改成1把互斥量恢复成原样。7条件变量使用范围等待条件代码pthread_mutex_lock(mutex);while(条件为假pthread_cond_wait(cond, mutex);修改条件pthread_mutex_unlock(mutex);给条件发送信号代码pthread_mutex_lock(mutex);设置条件为真pthread_cond_signal(cond);pthread_mutex_unlock(mutex);3POSIX信号量POSIX信号量和SystemV信号量作⽤相同都是⽤于同步操作达到⽆冲突的访问共享资源⽬的。但POSIX可以⽤于线程间同步。初始化信号量#include semaphore.h int sem_init(sem_t *sem, int pshared, unsigned int value); /*参数 pshared:0表⽰线程间共享⾮零表⽰进程间共享 value信号量初始值*/销毁信号量int sem_destroy(sem_t *sem);等待信号量//功能等待信号量会将信号量的值减1 int sem_wait(sem_t *sem); //P()发布信号量//功能发布信号量表⽰资源使⽤完毕可以归还资源了。将信号量值加1。 int sem_post(sem_t *sem);//V()上⼀节⽣产者-消费者的例⼦是基于queue的,其空间可以动态分配,现在基于固定⼤⼩的环形队列重写这个程序POSIX信号量:基于环形队列的生产消费模型环形队列采⽤数组模拟⽤模运算来模拟环状特性环形结构起始状态和结束状态都是⼀样的不好判断为空或者为满所以可以通过加计数器或者标记位来判断满或者空。另外也可以预留⼀个空的位置作为满的状态但是我们现在有信号量这个计数器就很简单的进⾏多线程间的同步过程。#pragma once #include iostream #include string #include vector #include pthread.h #include unistd.h #include Sem.hpp #include Mutex.hpp const int defaultcap 5; template class T class RingQueue { public: RingQueue(int cap defaultcap) : _cap(cap), _rq(cap), _consumer_step(0), _productor_step(0), _blank_sem(cap), _data_sem(0) { } void Enqueue(T in) { _pmutex.Lock(); _blank_sem.P(); _rq[_productor_step] in; _productor_step % _cap; _data_sem.V(); _pmutex.Unlock(); } void Pop(T *out) { _cmutex.Lock(); _data_sem.P(); *out _rq[_consumer_step]; _consumer_step % _cap; _blank_sem.V(); _cmutex.Unlock(); } ~RingQueue() { } private: int _cap; std::vectorT _rq; int _consumer_step; int _productor_step; Sem _blank_sem; Sem _data_sem; Mutex _cmutex; Mutex _pmutex; };4线程池下⾯开始我们结合我们之前所做的所有封装进⾏⼀个线程池的设计。在写之前我们要做如准备准备线程的封装准备锁和条件变量的封装引⼊⽇志对线程进⾏封装1日志与策略模式什么事设计模式IT⾏业这么⽕, 涌⼊的⼈很多. 俗话说林⼦⼤了啥⻦都有. ⼤佬和菜鸡们两极分化的越来越严重. 为了让菜鸡们不太拖⼤佬的后腿, 于是⼤佬们针对⼀些经典的常⻅的场景, 给定了⼀些对应的解决⽅案这个就是设计模式日志认识计算机中的⽇志是记录系统和软件运⾏中发⽣事件的⽂件主要作⽤是监控运⾏状态、记录异常信息帮助快速定位问题并⽀持程序员进⾏问题修复。它是系统维护、故障排查和安全管理的重工具。日志格式必须要有1时间戳2日志等级问题的严重程度3日志内容非必要日志指标1文件名行号2进程线程相关id信息。。。日志代码#ifndef __LOGGER_HPP #define __LOGGER_HPP #include iostream #include string #include sys/time.h #include time.h #include filesystem #include fstream #include unistd.h #include sys/types.h #include sstream #include Mutex.hpp namespace NS_LOG_MOUDLE { enum class LogLevel { INFO, WARING, ERROR, FATAL, DEBUG }; std::string LogLevel2Message(LogLevel level) { switch (level) { case LogLevel::INFO: return Info; case LogLevel::WARING: return Waring; case LogLevel::ERROR: return Error; case LogLevel::FATAL: return Fatal; case LogLevel::DEBUG: return Debug; } } std::string GetCurrentTime() { struct timeval current_time; int n gettimeofday(current_time, nullptr); (void)n; struct tm struct_time; localtime_r((current_time.tv_sec), struct_time); char timestr[128]; snprintf(timestr, sizeof(timestr), %d-%d-%d %d:%d:%d.%d, struct_time.tm_year 1900, struct_time.tm_mon 1, struct_time.tm_mday, struct_time.tm_hour, struct_time.tm_min, struct_time.tm_sec, current_time.tv_sec); return timestr; } class LogStrategy { public: virtual ~LogStrategy() default; virtual void SyncLog(const std::string message) 0; }; class ConsoleStrategy : public LogStrategy { public: void SyncLog(const std::string message) { LockGuard lockguard(_mutex); std::cerr message std::endl; } ~ConsoleStrategy() {} private: Mutex _mutex; }; const std::string defaultpath ./log; const std::string defaultfilename log.txt; // 文件策略 class FileLogstrategy : public LogStrategy { public: FileLogstrategy(const std::string logpath defaultpath, const std::string logfilename defaultfilename) : _logpath(logpath), _logfilename(logfilename) { LockGuard lockguard(_mutex); if (std::filesystem::exists(_logpath)) return; try { std::filesystem::create_directories(_logpath); } catch (const std::filesystem::filesystem_error e) { std::cerr e.what() \n; } if (!_logpath.empty() _logpath.back() ! /) { _logpath /; } } virtual void SyncLog(const std::string message) override { std::string targetlog _logpath / _logfilename; std::ofstream out(targetlog, std::ios::app); if (!out.is_open()) { std::cerr open file std::endl; return; } out message \n; out.close(); } private: std::string _logpath; std::string _logfilename; Mutex _mutex; }; class Logger { public: Logger() { } void UseConsoleStrategy() { _strategy std::make_uniqueConsoleStrategy(); } void UserFileStrategy() { _strategy std::make_uniqueFileLogstrategy(); } class LogMessage { public: LogMessage(LogLevel level, std::string filename, int line, Logger logger) : _level(level), _current_time(GetCurrentTime()), _pid(getpid()), _filename(filename), _line(line), _logger(logger) { std::stringstream ss; ss [ _current_time ] [ LogLevel2Message(_level) ] [ _pid ] [ _filename ] [ _line ]; _loginfo ss.str(); } template class T LogMessage operator(const T info) { std::stringstream ss; ss info; _loginfo ss.str(); return *this; } ~LogMessage() { if (_logger._strategy) { _logger._strategy-SyncLog(_loginfo); } } private: LogLevel _level; std::string _current_time; pid_t _pid; std::string _filename; int _line; std::string _loginfo; Logger _logger; }; LogMessage operator()(LogLevel level, std::string filename, int line) { return LogMessage(level, filename, line, *this); } ~Logger() { } private: std::unique_ptrLogStrategy _strategy; }; Logger logger; #define ENABLE_CONSOLE_LOG_STRATEGY() logger.UseConsoleStrategy(); #define ENABLE_FILE_LOG_STRATEGY() logger.UseFileStrategy(); #define LOG(level) logger(level, __FILE__, __LINE__) } #endif2线程池设计线程池一种线程使用模式线程过多会带来调度开销进⽽影响缓存局部性和整体性能。⽽线程池维护着多个线程等待着监督管理者分配可并发执⾏的任务。这避免了在处理短时间任务时创建与销毁线程的代价。线程池不仅能够保证内核的充分利⽤还能防⽌过分调度。可⽤线程数量应该取决于可⽤的并发处理器、处理器内核、内存、⽹络sockets等的数量。线程池的应用场景1需要⼤量的线程来完成任务且完成任务的时间⽐较短。 ⽐如WEB服务器完成⽹⻚请求这样的任务使⽤线程池技术是⾮常合适的。因为单个任务⼩⽽任务数量巨⼤你可以想象⼀个热⻔⽹站的点击次数。 但对于⻓时间的任务⽐如⼀个Telnet连接请求线程池的优点就不明显了。因为 Telnet会话时间⽐线程的创建时间⼤多了。2对性能要求苛刻的应⽤⽐如要求服务器迅速响应客⼾请求。3接受突发性的⼤量请求但不⾄于使服务器因此产⽣⼤量线程的应⽤。突发性⼤量客⼾请求在没有线程池情况下将产⽣⼤量线程虽然理论上⼤部分操作系统线程数⽬最⼤值不是问题短时间内产⽣⼤量线程可能使内存到达极限出现错误.线程池种类创建固定数量线程池循环从任务队列中获取任务对象获取到任务对象后执⾏任务对象中的任务接⼝浮动线程池其他同上线程池代码#pragma once #include vector #include queue #include Thread.hpp #include Logger.hpp #include Cond.hpp namespace NS_THREAD_POOL_MODULE { using namespace NS_LOG_MODULE; using namespace NS_THREAD_MODULE; const int defaultnum 5; template class T class ThreadPool { public: void HandlerTask() { char name[128]; pthread_getname_np(pthread_self(), name, sizeof(name)); while (1) { T task; { LockGuard lkg(_mutex); // 检测任务。不休眠1. 队列不为空 2. 线程池退出 - 队列为空 线程池不退出 while (_tasks.empty() _isrunning) { _slavers_sleep_count; _cond.Wait(_mutex); _slavers_sleep_count--; } // 线程池退出了- while 就要break - 不能 // 1. 线程池退出 _tasks empty if (!_isrunning _tasks.empty()) { break; } task _tasks.front(); _tasks.pop(); } LOG(LogLevel::INFO) name 处理ing; task(); LOG(LogLevel::DEBUG) task.Result(); } } public: ThreadPool(int slaver_num defaultnum) : _isrunning(false), _slavers_sleep_count(0), _slaver_num(slaver_num) { for (int idx 0; idx _slaver_num; idx) { _slavers.emplace_back([this]() { this-HandlerTask(); }); } } void Start() { if (_isrunning) { LOG(LogLevel::WARING) Thread Pool Is Already Running; return; } _isrunning true; for (auto slave : _slavers) { slave.Start(); } } void Stop() { _mutex.Lock(); _isrunning false; if (_slavers_sleep_count 0) _cond.Broadcast(); _mutex.Unlock(); } void Wait() { for (auto slave : _slavers) { slave.Join(); } } void Enqueue(T in) { _mutex.Lock(); _tasks.push(in); if (_slavers_sleep_count 0) { _cond.Signal(); } _mutex.Unlock(); } ~ThreadPool() { } private: bool _isrunning; int _slaver_num; std::vectorThread _slavers; std::queueT _tasks; Mutex _mutex; Cond _cond; int _slavers_sleep_count; }; }条件变量和线程自己可以封装。5线程安全和重入问题概念线程安全就是多个线程在访问共享资源时能够正确地执⾏不会相互⼲扰或破坏彼此的执⾏结 果。⼀般⽽⾔多个线程并发同⼀段只有局部变量的代码时不会出现不同的结果。但是对全局变量 或者静态变量进⾏操作并且没有锁保护的情况下容易出现该问题。重入同⼀个函数被不同的执⾏流调⽤当前⼀个流程还没有执⾏完就有其他的执⾏流再次进⼊ 我们称之为重⼊。⼀个函数在重⼊的情况下运⾏结果不会出现任何不同或者任何问题则该函数被称为可重⼊函数否则是不可重⼊函数。学到现在其实我们已经能理解重入其实可以分为两种情况1多线程重入函数2信号导致一个执行流重复进入函数常⻅的线程不安全的情况不保护共享变量的函数函数状态随着被调⽤状态发⽣变化的函数返回指向静态变量指针的函数调⽤线程不安全函数的函数常⻅不可重⼊的情况调⽤了malloc/free函数因为malloc函数 是⽤全局链表来管理堆的调⽤了标准I/O库函数标准I/O库的很多实 现都以不可重⼊的⽅式使⽤全局数据结构可重⼊函数体内使⽤了静态的数据结构常见的线程安全情况每个线程对全局变量或者静态变量只有读取 的权限⽽没有写⼊的权限⼀般来说这些 线程是安全的类或者接⼝对于线程来说都是原⼦操作多个线程之间的切换不会导致该接⼝的执⾏ 结果存在⼆义性常见可重入情况不使⽤全局变量或静态变量不使⽤ malloc或者new开辟出的空间不调⽤不可重⼊函数不返回静态或全局数据所有数据都有函数的调⽤者提供使⽤本地数据或者通过制作全局数据的本地拷⻉来保护全局数据结论