Linux系统与系统编程（15）——线程同步与互斥

前言欢迎观看Linux系列文章第15篇主要讲述了线程的互斥与同步的基本概念、互斥与同步的函数接口、POSIX信号量。其中重点讲述了生产者和消费者模型并基于阻塞队列和环形队列实现了该模型。线程互斥基本概念线程是共享地址空间的会共享大部分资源在共享这些公共资源的时候会导致各种情况下的数据不一致。所以就需要有线程的同步和互斥。· 共享资源线程共享地址空间共享大部分资源。· 临界资源多线程执行流被保护的共享的资源。· 临界区每个线程内部访问临界资源的代码。· 互斥任何时刻互斥保证有且只有一个执行流进入临界区访问临界资源通常对临界资源起到保护作用。· 原子性不会被任何调度机制打断的操作操作只有已完成操作和未开始操作两种状态。为了实现互斥我们就需要使用到锁这个工具。int ticket 10000; pthread_mutex_t lock PTHREAD_MUTEX_INITIALIZER;//创建锁 void *routine(void *arg) { char *id (char *)arg; while(1) { pthread_mutex_lock(lock);//加锁 if(ticket 0) { usleep(1000); printf(%s sells ticket:%d\n, id, ticket); ticket--; pthread_mutex_unlock(lock);//开锁 } else { pthread_mutex_unlock(lock);//开锁 break; } } return nullptr; }以上代码模拟了一个抢票过程每个线程执行routine时会将ticket--。如果我们把代码中和锁有关的四条代码删除多线程运行上述代码后我们会发现ticket会最后会变为负数即使我们代码写了ticket在小于等于0时会break出循环。这可不符合理想状况因为我们期望的是ticket刚好减到0即票刚好卖完。为什么会这样呢上述操作中ticket这个全局变量就是临界资源但是ticket--这个操作不是原子的会被调度机制打断。在底层每次ticket--操作起码有三条指令1️⃣ticket载入寄存器、2️⃣ticket在运算器中 - 1、3️⃣ticket写回内存。在没有锁的情况下假设票数已经被减到了1此时有多条线程运行到了ticket0的判断后都被切换了切换到线程a后线程a完成了一次ticket--。之后依次切换回之前被切换走的线程。由于已经经ticket0判断后皆为true所以这多条线程都会在之后执行ticket--的操作。此时就出现了并发问题ticket被多次减1就变成了负数。· 切换线程的时间点1、时间片分配。2、出现I/O阻塞。3、sleep···等操作。出现以上情况时程序会陷入内核态选择新的线程会检查线程状态并返回到用户态。互斥锁的原理1.硬件级实现关闭时钟中断。2.软件级实现大多数体系结构提供了swap或exchange指令作用是寄存器和内存单元数据交换属于原子指令。虽然CPU硬件寄存器是共享的但是里面的内容上下文是线程私有的。哪个线程能够先把占用标志和内存中互斥锁的空闲标志交换谁就占用了那把锁。因为是原子指令其他线程不能打断这个交换。从而形成申请锁竞态。此时互斥锁就成为成功申请的线程的上下文随着线程一起切换。互斥锁的函数· 互斥锁的函数都是原子性的所以即使锁是临界区资源线程竞争申请锁时也不会出现并行问题。· 使用锁之后程序运行会比原来慢因为1.并行变成串行。2.要花时间进行互斥锁竞争。· 互斥锁的本质作用是让临界区代码变得原子化。创建互斥锁全局定义#include pthread.h pthread_mutex_t mutex PTHREAD_MUTEX_INITIALIZER;//全局定义在全局定义的互斥锁使用宏来定义出来。全局的互斥锁不需要主动释放程序运行结束时会自动释放。局部定义#include pthread.h int pthread_mutex_init(pthread_mutex_t *restrict mutex, const pthread_mutexattr_t *restrict attr);初始化一把锁。参数mutex指向互斥锁变量的指针。attr互斥锁的属性可以不设置传入nullptr使用默认属性。释放互斥锁#include pthread.h int pthread_mutex_destroy(pthread_mutex_t *mutex);将局部创建的互斥锁释放全局定义的互斥锁不需要自行释放。前提是锁没有在被使用。参数mutex执行需要释放的互斥锁的指针。加互斥锁阻塞#include pthread.h int pthread_mutex_lock(pthread_mutex_t *mutex);参数mutex执行需要申请的互斥锁的指针。申请已经创建的互斥锁申请成功后加锁锁变为锁定状态否则阻塞挂起。锁本身是临界资源多线程都得先看到锁进行竞争申请互斥锁。该函数后应跟着临界区代码申请锁成功后就直接往后执行临界区代码访问临界区资源。ps实际上就算加了锁线程在运行临界区代码时依旧可能会被切换走。但是因为有锁就算被切换走了其他线程也无法运行只能切换回加锁线程直到其运行完毕。非阻塞#include pthread.h int pthread_mutex_trylock(pthread_mutex_t *mutex);参数mutex执行需要申请的互斥锁的指针。和pthread_mutex_lock的区别是申请失败不会阻塞。解互斥锁#include pthread.h int pthread_mutex_unlock(pthread_mutex_t *mutex);参数mutex执行需要解开的互斥锁的指针。让锁定的互斥锁变为非锁定状态。C11封装的互斥锁C11把上述函数进行了封装用起来更方便一点。#include mutex std::mutex cpp_lock; int main() { //······ //非临界区代码 //······ cpp_lock.lock();//加锁 //······ //临界区代码 //······ cpp_lock.unlock();//开锁 //······ //非临界区代码 //······ }释放和创建锁分别由构造和析构来完成。线程同步基本概念仅靠互斥是有缺点的可能会出现申请到锁的线程多次出现解锁之后又立马加锁的情况不仅多次来回申请会浪费资源而且会导致其他线程拿不到锁就没有办法运行下去。这种情况叫做其他线程的饥饿问题。· 条件变量但一个线程互斥访问某个变量时他可能会发现在其他线程改变状态之前什么都干不了这时就需要用到条件变量。· 同步在保证数据安全的前提下让线程能够按照某种特定的顺序访问临界资源从而有效避免免饥饿问题。生产与消费模型日常场景中我们作为消费者购买商品都是去超市购买而不是在工厂购买因为作为个人需要的商品量低让工厂专门为我们生产会导致效率低下成本也高。所以实际经验是工厂生产大量的商品然后批量发给超市多个用户再去超市购买少量商品。超市作为一个中间交易的场所提高了整体生产消费的效率。1个交易场所、2个角色我们把这个场景类比到多线程场景。一些线程作为生产者一些线程作为消费者而商品就是数据超市里的商品会被工厂和消费者修改所以超市的商品会被工厂和消费者看到也就是说超市是一种临界资源。3种关系生产者之间是典型的互斥和竞争关系因为超市存放商品的数量是有上限的消费者之间也是互斥和竞争关系资源越少越明显生产者和消费者之间依旧是互斥关系消费者要在商品上架完毕之后才可以购买生产者上架商品完成前消费者也还要等待同时生产者和消费者时间是可以协商谁先谁后操作商品也就是也具有同步关系。好处·解耦生产和消费过程。·支持线程之间忙闲不均。·提高效率。实际编写代码时要根据实际场景中存在的角色生产者、消费者和角色数量完善好角色之间的互斥和同步关系。条件变量的函数和互斥锁函数看起来很类似目的是完成线程的同步。创建条件变量全局定义#include pthread.h //全局static定义 pthread_cond_t cond PTHREAD_COND_INITIALIZER;无需主动释放程序结束后会自动释放。局部定义#include pthread.h //局部定义 int pthread_cond_init(pthread_cond_t *restrict cond, const pthread_condattr_t *restrict attr);参数cond指向条件变量的指针。attr条件变量cond的设置属性传入nullptr则设置为默认属性。等待函数无期限的等待#include pthread.h int pthread_cond_wait(pthread_cond_t *restrict cond, pthread_mutex_t *restrict mutex);参数cond指向指定的条件变量会在该条件变量下的等待。mutex指定的互斥锁。当前线程释放互斥锁将线程阻塞挂直到被唤醒为止。被唤醒时线程仍在该函数中还在临界区代码里要继续运行就要重新申请锁如果申请失败就继续阻塞等待直到申请到锁为止。这就是为什么等待函数要传入互斥锁来作为参数的原因。限时等待#include pthread.h int pthread_cond_timedwait(pthread_cond_t *restrict cond, pthread_mutex_t *restrict mutex, const struct timespex *restrict abstime);参数cond指向指定的条件变量会在该条件变量下的等待。mutex指定的互斥锁。abstime绝对时间点。timespex结构体如下struct timespex { time_t tv_sec; long tv_nsec; }tv_sec为时间戳tv_nsec为纳秒数范围为0~999,999,999其实就是秒数的小数部分。作用和无限期等待唯一的区别是超时的等待会返回错误码ETIMEOUT。唤醒函数唤醒一个线程#include pthread.h int pthread_cond_signal(pthread_cont_t *cond);唤醒在该条件变量下的等待的一个线程通常先唤醒等得最久那个。唤醒所有线程#include pthread.h int pthread_cond_broadcast(pthread_cont_t *cond);唤醒所有等待的线程这些线程会一起竞争互斥锁。基于blockqueue阻塞队列的生产消费模型POSIX信号量和之前学过的systemV类似但是POSIX比较新systemV基本已经被淘汰了。多线程使用资源有两种场景1.将目标资源整体使用【互斥和2元信号量2元信号量本身就有互斥的特性】2.将目标资源按“块”分批使用【多元信号量】要根据实际场景选择维护互斥同步的方式。所有的线程都得看到信号量sem、计数器、sem或sem--信号量本身也是临界资源。P操作计数器--原子V操作计数器原子POSIX信号量的函数初始化信号量#include semaphore.h int sem_init(sem_t *sem, int pshared, unsigned int value);参数pshared0表示线程间共享非0表示进程间共享。value信号量初始值销毁信号量#include semaphore.h int sem_destroy(sem_t *sem);等待信号量#include semaphore.h int sem_wait(sem_t *sem);等待信号量会把等待的信号量值-1。也就是P操作。发布信号量#include semaphore.h int sem_post(sem_t *sem);发布信号量表示资源使用完毕归还资源将信号量值1。也就是V操作。基于环形队列的生产者消费者模型环形队列中设置CP两个指针指向消费生产数据的位置。规则1.空时生产者先运行。2.满时消费者先运行。3.生产者不能把消费者套一个圈以上。4.消费者不能超过生产者。特点1.只要C、P指针不访问同一个位置就可以同时进行生产消费。2.只有为空和为满时C、P指针才会在同一个位置。即非空非满时生产消费就可以同时进行。3.为空或为满时生产和消费就需要同步和互斥。关键代码Roundqueue.hpp环形队列代码#pragma once #include iostream #include vector #include Sem.hpp #include Mutex.hpp static const int gcap 5; // for debug using namespace SemModule; using namespace MutexModule; template typename T class Roundqueue { public: Roundqueue(int cap gcap) : _cap(cap), _rq(cap), _blank_sem(cap), _p_step(0), _data_sem(0), _c_step(0) { } void Equeue(const T in) { // 生产者 // 1. 申请信号量空位置信号量 _blank_sem.P(); { LockGuard lockguard(_pmutex); // 2. 生产 _rq[_p_step] in; // 3. 更新下标 _p_step; // 4. 维持环形特性 _p_step % _cap; } _data_sem.V(); } void Pop(T *out) { // 消费者 // 1. 申请信号量数据信号量 _data_sem.P(); { LockGuard lockguard(_cmutex); // 2. 消费 *out _rq[_c_step]; // 3. 更新下标 _c_step; // 4. 维持环形特性 _c_step % _cap; } _blank_sem.V(); } private: std::vectorT _rq; int _cap; // 生产者 Sem _blank_sem; // 空位置 int _p_step; // 消费者 Sem _data_sem; // 数据 int _c_step; // 维护多生产多消费, 2把锁 Mutex _cmutex; Mutex _pmutex; };其他头文件Mutex.hpp、Sem.hpp#pragma once #include iostream #include pthread.h namespace MutexModule { class Mutex { public: Mutex() { pthread_mutex_init(_mutex, nullptr); } void Lock() { int n pthread_mutex_lock(_mutex); (void)n; } void Unlock() { int n pthread_mutex_unlock(_mutex); (void)n; } ~Mutex() { pthread_mutex_destroy(_mutex); } pthread_mutex_t *Get() { return _mutex; } private: pthread_mutex_t _mutex; }; class LockGuard { public: LockGuard(Mutex mutex):_mutex(mutex) { _mutex.Lock(); } ~LockGuard() { _mutex.Unlock(); } private: Mutex _mutex; }; }#include iostream #include pthread.h #include semaphore.h namespace SemModule { const int defaultvalue 1; class Sem { public: Sem(unsigned int sem_value defaultvalue) { sem_init(_sem, 0, sem_value); } 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; }; }❤~~本文完结感谢观看接下来更精彩欢迎来我博客做客~~❤