Linux多线程【线程互斥】

Linux线程互斥

进程线程间的互斥相关背景概念

• 临界资源：多线程执行流共享的资源就叫做临界资源
• 临界区：每个线程内部，访问临界资源的代码，就叫做临界区
• 互斥：任何时刻，互斥保证有且只有一个执行流进入临界区，访问临界资源，通常对临界资源起保护作用
• 原子性（后面讨论如何实现）：不会被任何调度机制打断的操作，该操作只有两态，要么完成，要么未完成

---

互斥量mutex

• 大部分情况，线程使用的数据都是局部变量，变量的地址空间在线程栈空间内，这种情况，变量归属单个 线程，其他线程无法获得这种变量。
• 但有时候，很多变量都需要在线程间共享，这样的变量称为共享变量，可以通过数据的共享，完成线程之 间的交互。
• 多个线程并发的操作共享变量，会带来一些问题。

---

模拟抢票代码

• 我们看下面这段代码：

#define NUM 4// 线程名字
class ThreadData
{
public:ThreadData(int num){threadname = "thread-" + to_string(num);}public:string threadname;
};int tickets = 1000;// 买票操作
void *getTicket(void *args)
{ThreadData *td = static_cast<ThreadData *>(args);const char *name = td->threadname.c_str();while (true){if (tickets > 0){usleep(1000);printf("who=%s, get a ticket: %d\n", name, tickets); // ?tickets--;}elsebreak;}printf("%s ... quit\n", name);return nullptr;
}int main()
{vector<pthread_t> tids;vector<ThreadData *> thread_datas;for (int i = 1; i <= NUM; i++){pthread_t tid;ThreadData *td = new ThreadData(i);thread_datas.push_back(td);pthread_create(&tid, nullptr, getTicket, thread_datas[i - 1]);tids.push_back(tid);}// 等待for (auto e : tids){pthread_join(e, nullptr);}// 释放for (auto e : thread_datas){delete e;}return 0;
}

• 造成了数据不一致问题，肯定是和多线程并发有关系

• 一个全局变量进行多线程并发++或者--操作是否安全？–>不安全！

• if 语句判断条件为真以后，代码可以并发的切换到其他线程

• usleep 这个模拟漫长业务的过程，在这个漫长的业务过程中，可能有很多个线程会进入该代码段

• --ticket 操作本身就不是一个原子操作

---

取出ticket--部分的汇编代码

objdump -d mythread > test.objdump

14e6:	8b 05 24 5b 00 00    	mov    0x5b24(%rip),%eax        # 7010 <tickets>
14ec:	83 e8 01             	sub    $0x1,%eax
14ef:	89 05 1b 5b 00 00    	mov    %eax,0x5b1b(%rip)        # 7010 <tickets>

-- 操作并不是原子操作，而是对应三条汇编指令：

• load ：将共享变量ticket从内存加载到寄存器中
• update : 更新寄存器里面的值，执行-1操作
• store ：将新值，从寄存器写回共享变量ticket的内存地址

要解决以上问题，需要做到三点：

• 代码必须要有互斥行为：当代码进入临界区执行时，不允许其他线程进入该临界区。
• 如果多个线程同时要求执行临界区的代码，并且临界区没有线程在执行，那么只能允许一个线程进入该临界区。
• 如果线程不在临界区中执行，那么该线程不能阻止其他线程进入临界区。

要做到这三点，本质上就是需要一把锁。Linux上提供的这把锁叫互斥量。

互斥量的接口

• 在Ubuntu下查看需要先安装手册

sudo apt-get install manpages-posix-dev

初始化互斥量

#include <pthread.h>
int pthread_mutex_init(pthread_mutex_t *restrict mutex,const pthread_mutexattr_t *restrict attr);
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

• 第一个是动态分配

int pthread_mutex_init(pthread_mutex_t *restrict mutex,const pthread_mutexattr_t *restrict attr);

• 第二个是静态分配

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

销毁互斥量

#include <pthread.h>int pthread_mutex_destroy(pthread_mutex_t *mutex);
int pthread_mutex_init(pthread_mutex_t *restrict mutex,const pthread_mutexattr_t *restrict attr);
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

销毁互斥量需要注意：

• 使用PTHREAD_ MUTEX_ INITIALIZER初始化的互斥量不需要销毁
• 不要销毁一个已经加锁的互斥量
• 已经销毁的互斥量，要确保后面不会有线程再尝试加锁

互斥量加锁和解锁

#include <pthread.h>int pthread_mutex_lock(pthread_mutex_t *mutex);
int pthread_mutex_trylock(pthread_mutex_t *mutex);
int pthread_mutex_unlock(pthread_mutex_t *mutex);

调用 pthread_ lock 时，可能会遇到以下情况:

• 互斥量处于未锁状态，该函数会将互斥量锁定，同时返回成功
• 发起函数调用时，其他线程已经锁定互斥量，或者存在其他线程同时申请互斥量，但没有竞争到互斥量， 那么pthread_ lock调用会陷入阻塞(执行流被挂起)，等待互斥量解锁。

加锁的本质：是用时间来换取安全
加锁的表现：线程对于临界区代码串执行
加锁原则：尽量的要保证临界区代码，越少越好

---

改进模拟抢票代码（加锁）

• 然后就可以使用上面的性质来进行改进模拟抢票代码

#define NUM 4
pthread_mutex_t lock;// 线程名字
class ThreadData
{
public:ThreadData(int num, pthread_mutex_t *mutex){lock = mutex; // 初始化锁threadname = "thread-" + to_string(num);}public:string threadname;pthread_mutex_t *lock; // 定义一个锁指针
};int tickets = 1000;// 买票操作
void *getTicket(void *args)
{ThreadData *td = static_cast<ThreadData *>(args);const char *name = td->threadname.c_str();while (true){// 加锁 -->pthread_mutex_lock(td->lock); // 申请锁成功，才能往后执行，不成功，阻塞等待。if (tickets > 0){usleep(1000);printf("who=%s, get a ticket: %d\n", name, tickets);tickets--;// 解锁 -->pthread_mutex_unlock(td->lock);}else{// 解锁 --> 在else执行流要注意break，之前要解锁pthread_mutex_unlock(td->lock);break;}usleep(13); // 执行得到票之后的后续动作，如果不加usleep会导致 饥饿问题}printf("%s ... quit\n", name);return nullptr;
}int main()
{vector<pthread_t> tids;vector<ThreadData *> thread_datas;// 初始化锁pthread_mutex_init(&lock, nullptr);for (int i = 1; i <= NUM; i++){pthread_t tid;ThreadData *td = new ThreadData(i, &lock);thread_datas.push_back(td);pthread_create(&tid, nullptr, getTicket, thread_datas[i - 1]);tids.push_back(tid);}// 等待for (auto e : tids){pthread_join(e, nullptr);}// 释放for (auto e : thread_datas){delete e;}return 0;
}

---

• 或者使用全局初始化也可以

#define NUM 4// 直接初始化锁
pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;// 线程名字
class ThreadData
{
public:ThreadData(int num){threadname = "thread-" + to_string(num);}public:string threadname;
};int tickets = 1000;// 买票操作
void *getTicket(void *args)
{ThreadData *td = static_cast<ThreadData *>(args);const char *name = td->threadname.c_str();while (true){pthread_mutex_lock(&lock);if (tickets > 0){usleep(1000);printf("who=%s, get a ticket: %d\n", name, tickets);tickets--;// 解锁 -->pthread_mutex_unlock(&lock);}else{// 解锁 -->pthread_mutex_unlock(&lock);break;}usleep(13); // 执行得到票之后的后续动作，如果不加usleep会导致 饥饿问题}printf("%s ... quit\n", name);return nullptr;
}int main()
{vector<pthread_t> tids;vector<ThreadData *> thread_datas;// 初始化锁pthread_mutex_init(&lock, nullptr);for (int i = 1; i <= NUM; i++){pthread_t tid;ThreadData *td = new ThreadData(i);thread_datas.push_back(td);pthread_create(&tid, nullptr, getTicket, thread_datas[i - 1]);tids.push_back(tid);}// 等待for (auto e : tids){pthread_join(e, nullptr);}// 释放for (auto e : thread_datas){delete e;}return 0;
}

• 如果纯互斥环境，如果锁分配不够合理，容易被其他线程的饥饿问题！
• 不是说只要有互斥，必有饥饿
• 适合纯互斥的场景，就用互斥
• 如果是按照一定的顺序性获取资源，就叫同步

---

• 其中在临界区中，线程可以被切换吗？ 当然可以

  • 在线程被切出去的时候，是持有锁走的
  • 不在临界区之间，照样没有线程可以进入临界区访问临界资源

• 对于其他线程来讲，一个线程要么没有锁，要么释放锁。

• 当线程访问临界资源区的过程，对于其他线程是原子的。

---

小结

1. 所以，加锁的范围，粒度一定要小
2. 任何线程，要进行抢票，都得先申请锁，原则上，不应该有例外
3. 所有线程申请锁，前提是所有线程都得看到这把锁，锁本身也是共享资源，加锁的过程，必须是原子的！原子性：要么不做，要做就做完，没有中间状态，就是原子性
4. 如果线程申请锁失败了，我的线程要被阻塞
5. 如果线程申请成功了，继续向后运行
6. 如果线程申请锁成功了，执行临界区的代码了，执行临界区的代码是可以被切换的，其他线程无法进入，因为被切换了，但是没有释放，可以放心的执行完毕，没有任何线程能打扰。

结论：所有对于其他线程，要么没有申请锁，要么释放了锁，对于其他线程才有意义

---

什么是线程互斥，为什么需要互斥：

• 线程互斥指的是在多个线程间对临界资源进行争抢访问时有可能会造成数据二义，因此通过保证同一时间只有一个线程能够访问临界资源的方式实现线程对临界资源的访问安全性

---

对锁封装 lockGuard.hpp

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(){}private:pthread_mutex_t *_lock;
};class LockGuard
{
public:LockGuard(pthread_mutex_t *lock) : _mutex(lock){_mutex.Lock();}~LockGuard(){_mutex.Unlock();}private:Mutex _mutex;
};

mythread.cc

void *getTicket(void *args)
{ThreadData *td = static_cast<ThreadData *>(args);const char *name = td->threadname.c_str();while (true){{ // 对锁进行临时对象控制// 定义了一个临时的锁对象LockGuard lockGuard(&lock); // RAII风格的锁if (tickets > 0){usleep(1000);printf("who=%s, get a ticket: %d\n", name, tickets);tickets--;}elsebreak;}usleep(13); // 执行得到票之后的后续动作，如果不加usleep会导致 饥饿问题}printf("%s ... quit\n", name);return nullptr;
}

互斥量实现原理探究

• 经过上面的例子，已经意识到单纯的 i++ 或者 ++i 都不是原子的，有可能会有数据一致性问题
• 为了实现互斥锁操作,大多数体系结构都提供了swap或exchange指令,该指令的作用是把寄存器和内存单元的数据相交换,由于只有一条指令,保证了原子性,即使是多处理器平台，访问内存的 总线周期也有先后，一 个处理器上的交换指令执行时另一个处理器的交换指令只能等待总线周期。 现在我们把lock和unlock的伪代码改一下

1. CPU的寄存器只有一套，被所有的线程共享，但是寄存器里面的数据，属于执行流的上下文，属于执行流私有的数据。
2. CPU在执行代码的时候，一定要有对应的执行载体，线程和进程
3. 数据在内存中没被所有线程共享

结论：把数据从内存移动到CPU寄存器中，本质是把数据从共享变成线程私有！！

---

• 竞争锁本质是在谁先把xchgb做完

• mutex简单理解就是一个0/1的计数器，用于标记资源访问状态：

  • 0表示已经有执行流加锁成功，资源处于不可访问，
  • 1表示未加锁，资源可访问。

可重入VS线程安全

概念

• 线程安全：多个线程并发同一段代码时，不会出现不同的结果。常见对全局变量或者静态变量进行操作， 并且没有锁保护的情况下，会出现该问题。
• 重入：同一个函数被不同的执行流调用，当前一个流程还没有执行完，就有其他的执行流再次进入，我们 称之为重入。一个函数在重入的情况下，运行结果不会出现任何不同或者任何问题，则该函数被称为可重 入函数，否则，是不可重入函数。

常见的线程不安全的情况

• 不保护共享变量的函数
• 函数状态随着被调用，状态发生变化的函数
• 返回指向静态变量指针的函数
• 调用线程不安全函数的函数

常见的线程安全的情况

• 每个线程对全局变量或者静态变量只有读取的权限，而没有写入的权限，一般来说这些线程是安全的
• 类或者接口对于线程来说都是原子操作
• 多个线程之间的切换不会导致该接口的执行结果存在二义性

常见不可重入的情况

• 调用了malloc/free函数，因为malloc函数是用全局链表来管理堆的
• 调用了标准I/O库函数，标准I/O库的很多实现都以不可重入的方式使用全局数据结构
• 可重入函数体内使用了静态的数据结构

常见可重入的情况

• 不使用全局变量或静态变量
• 不使用用malloc或者new开辟出的空间
• 不调用不可重入函数 不返回静态或全局数据，所有数据都有函数的调用者提供
• 使用本地数据，或者通过制作全局数据的本地拷贝来保护全局数据

可重入与线程安全联系

• 函数是可重入的，那就是线程安全的
• 函数是不可重入的，那就不能由多个线程使用，有可能引发线程安全问题
• 如果一个函数中有全局变量，那么这个函数既不是线程安全也不是可重入的。

可重入与线程安全区别

• 可重入函数是线程安全函数的一种
• 线程安全不一定是可重入的，而可重入函数则一定是线程安全的。
• 如果将对临界资源的访问加上锁，则这个函数是线程安全的，但如果这个重入函数若锁还未释放则会产生死锁，因此是不可重入的。

---

常见锁概念

死锁

• 死锁是指在一组进程中的各个进程均占有不会释放的资源，但因互相申请被其他进程所站用不会释放的资源而处于的一种永久等待状态。

死锁四个必要条件

• 互斥条件：一个资源每次只能被一个执行流使用
• 请求与保持条件：一个执行流因请求资源而阻塞时，对已获得的资源保持不放
• 不剥夺条件：一个执行流已获得的资源，在末使用完之前，不能强行剥夺
• 循环等待条件：若干执行流之间形成一种头尾相接的循环等待资源的关系

避免死锁

• 破坏死锁的四个必要条件
• 加锁顺序一致
• 避免锁未释放的场景
• 资源一次性分配

避免死锁算法

• 死锁检测算法
• 银行家算法

Linux线程同步

条件变量

• 当一个线程互斥地访问某个变量时，它可能发现在其它线程改变状态之前，它什么也做不了。
• 例如一个线程访问队列时，发现队列为空，它只能等待，只到其它线程将一个节点添加到队列中。这种情 况就需要用到条件变量。

同步概念与竞态条件

• 同步：在保证数据安全的前提下，让线程能够按照某种特定的顺序访问临界资源，从而有效避免饥饿问题，叫做同步
• 竞态条件：因为时序问题，而导致程序异常，我们称之为竞态条件。在线程场景下，这种问题也不难理解

条件变量函数

初始化

#include <pthread.h>int pthread_cond_init(pthread_cond_t *restrict cond,const pthread_condattr_t *restrict attr);
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;

• cond：要初始化的条件变量
• attr：NULL

销毁

int pthread_cond_destroy(pthread_cond_t *cond);

等待条件满足

#include <pthread.h>int pthread_cond_timedwait(pthread_cond_t *restrict cond,pthread_mutex_t *restrict mutex,const struct timespec *restrict abstime);
int pthread_cond_wait(pthread_cond_t *restrict cond,pthread_mutex_t *restrict mutex);

• cond：要在这个条件变量上等待
• mutex：互斥量，后面详细解释

唤醒等待

#include <pthread.h>int pthread_cond_broadcast(pthread_cond_t *cond);
int pthread_cond_signal(pthread_cond_t *cond);

案例：

#include <iostream>
#include <unistd.h>
#include <pthread.h>// 临界资源
int cnt = 0;// 初始化
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;void *Count(void *args)
{uint64_t number = (uint64_t)args;std::cout << "pthread: " << number << " create sucess! " << std::endl;while (true){// 加锁pthread_mutex_lock(&mutex);// 必须先加锁后等待pthread_cond_wait(&cond, &mutex);std::cout << "pthread: " << number << ", cnt: " << cnt++ << std::endl;// 解锁pthread_mutex_unlock(&mutex);}// 分离线程pthread_detach(pthread_self());
}int main()
{for (uint64_t i = 0; i < 5; i++){pthread_t tid;pthread_create(&tid, nullptr, Count, (void *)i); // 注意这里细节不能传i的地址usleep(1000);                                    // 创建出来的线程打印就保持一致了}sleep(3);std::cout << "main thread ctrl begin: " << std::endl;while (true){sleep(1);// 唤醒单个线程pthread_cond_signal(&cond);std::cout << "signal thread..." << std::endl;}return 0;
}

• 也还可以唤醒多个线程

#include <iostream>
#include <unistd.h>
#include <pthread.h>// 临界资源
int cnt = 0;// 初始化
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;void *Count(void *args)
{uint64_t number = (uint64_t)args;std::cout << "pthread: " << number << " create sucess! " << std::endl;while (true){// 加锁pthread_mutex_lock(&mutex);// 你怎么知道临界资源是就绪还是不就绪的？你判断出来的！判断是访问临界资源吗？必须是的，也就是判断必须在加锁之后！！！// 必须先加锁后等待pthread_cond_wait(&cond, &mutex); // pthread_cond_wait让线程等待的时候，会自动释放锁！std::cout << "pthread: " << number << ", cnt: " << cnt++ << std::endl;// 解锁pthread_mutex_unlock(&mutex);}// 分离线程pthread_detach(pthread_self());
}int main()
{for (uint64_t i = 0; i < 5; i++){pthread_t tid;pthread_create(&tid, nullptr, Count, (void *)i); // 注意这里细节不能传i的地址usleep(1000);                                    // 创建出来的线程打印就保持一致了}sleep(3);std::cout << "main thread ctrl begin: " << std::endl;while (true){sleep(1);// 唤醒多个线程pthread_cond_broadcast(&cond);std::cout << "boardcast thread..." << std::endl;}return 0;
}

为什么 pthread_cond_wait需要互斥量?

• 条件等待是线程间同步的一种手段，如果只有一个线程，条件不满足，一直等下去都不会满足，所以必须要有一个线程通过某些操作，改变共享变量，使原先不满足的条件变得满足，并且友好的通知等待在条件变量上的线程。
• 条件不会无缘无故的突然变得满足了，必然会牵扯到共享数据的变化。所以一定要用互斥锁来保护。没有互斥锁就无法安全的获取和修改共享数据。

按照上面的说法，我们设计出如下的代码：先上锁，发现条件不满足，解锁，然后等待在条件变量上不就行了

• 由于解锁和等待不是原子操作。调用解锁之后， 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，把互斥量恢复成原样。

生产者消费者模型

• 3种关系、2种角色、1个交易场所：

  • 3种关系：生产者和生产者（互斥），消费者和消费者（互斥），生产者和消费者（互斥 / 同步）

  • 2种角色：生产者、消费者（线程承担）

  • 1个交易场所：内存中特定的一种内存结构。

• 生产者消费者模式就是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者 要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。这个阻塞队 列就是用来给生产者和消费者解耦的。

优点：

1. 支持忙闲不闲
2. 生产和消费进行解耦
3. 支持并发

• 生产者vs生产者：互斥
• 消费者和消费者：互斥
• 生产者和消费者：互斥，同步

基于BlockingQueue的生产者消费者模型

BlockingQueue

• 在多线程编程中阻塞队列(Blocking Queue)是一种常用于实现生产者和消费者模型的数据结构。其与普通的队列区别 在于，当队列为空时，从队列获取元素的操作将会被阻塞，直到队列中被放入了元素；当队列满时，往队列里存放元 素的操作也会被阻塞，直到有元素被从队列中取出(以上的操作都是基于不同的线程来说的，线程在对阻塞队列进程 操作时会被阻塞)

C++ queue模拟阻塞队列的生产消费模型

blockqueue.hpp

#include <iostream>
#include <queue>template <class T>
class BlockQueue
{static const int defaultnum = 5;public:BlockQueue(int maxcap = defaultnum): maxcap_(maxcap){// 初始化锁pthread_mutex_init(&mutex_, nullptr);pthread_cond_init(&c_cond_, nullptr);pthread_cond_init(&p_cond_, nullptr);}T Pop(){// 加锁pthread_mutex_lock(&mutex_);while (q_.size() == 0) // 因为判断临界资源调试是否满足，也是在访问临界资源！判断资源是否就绪，是通过再临界资源内部判断的。{// 进行等待，调用的时候，自动释放锁，因为唤醒而返回的时候，重新持有锁pthread_cond_wait(&c_cond_, &mutex_);}// 取出队头数据，进行处理T out = q_.front();q_.pop();// 唤醒生产pthread_cond_signal(&p_cond_);// 如果线程wait时，被误唤醒了呢？？ -->// pthread_cond_broadcast(&c_cond_, &mutex_); // 唤醒所有线程// 解锁pthread_mutex_unlock(&mutex_);return out;}void Push(const T &in){// 加锁pthread_mutex_lock(&mutex_);while (q_.size() == maxcap_) // 因为判断临界资源调试是否满足，也是在访问临界资源！判断资源是否就绪，是通过再临界资源内部判断的。{pthread_cond_wait(&p_cond_, &mutex_);}// 入队q_.push(in);// 唤醒消费pthread_cond_signal(&c_cond_);// 解锁pthread_mutex_unlock(&mutex_);}// 析构~BlockQueue(){pthread_cond_destroy(&c_cond_);pthread_cond_destroy(&p_cond_);pthread_mutex_destroy(&mutex_);}private:// 一个队列std::queue<T> q_;// 一个极值，队列到达极值就不可入队int maxcap_;// 一个消费者，一个生产者pthread_cond_t c_cond_;pthread_cond_t p_cond_;// 锁pthread_mutex_t mutex_;
};

Task.hpp

#pragma once
#include <iostream>
#include <string>std::string opers = "+-*/%";enum
{DivZero = 1,ModZero,Unknown
};class Task
{
public:Task(int x, int y, char op): data1_(x), data2_(y), oper_(op){}void Run(){switch (oper_){case '+':result_ = data1_ + data2_;break;case '-':result_ = data1_ - data2_;break;case '*':result_ = data1_ * data2_;break;case '/':{if (data2_ == 0)exitcode_ = DivZero;elseresult_ = data1_ / data2_;}break;case '%':{if (data2_ == 0)exitcode_ = ModZero;elseresult_ = data1_ % data2_;}break;default:exitcode_ = Unknown;break;}}void operator()(){Run();}std::string GetResult() // 这里可以使用stringstream{std::string r = std::to_string(data1_);r += oper_;r += std::to_string(data2_);r += "=";r += std::to_string(result_);r += "[code: ";r += std::to_string(exitcode_);r += "]";return r;}std::string GetTask(){std::string r = std::to_string(data1_);r += oper_;r += std::to_string(data2_);r += "=?";return r;}~Task(){}private:int data1_; // 操作数int data2_; // 操作数char oper_; // 操作符int result_;   // 结果int exitcode_; // 错误码
};

Task.hpp（改进版本）

• 我们也可以使用包装器和stringstream来改进一下

#pragma once
#include <iostream>
#include <string>
#include <sstream>
#include <functional>
#include <map>std::string opers = "+-*/%";enum
{DivZero = 1,ModZero,Unknown
};class Task
{
public:Task(int x, int y, char op): data1_(x), data2_(y), oper_(op){}void Run(){std::map<char, std::function<void()>> CmdOp{{'+', [this](){ result_ = data1_ + data2_; }},{'-', [this](){ result_ = data1_ - data2_; }},{'*', [this](){ result_ = data1_ * data2_; }},{'/', [this](){ if (data2_ == 0) exitcode_ = DivZero; else result_ = data1_ / data2_; }},{'%', [this](){ if (data2_ == 0) exitcode_ = ModZero; else result_ = data1_ % data2_; }}};// auto it = CmdOp.find(oper_);std::map<char, std::function<void()>>::iterator it = CmdOp.find(oper_);if (it != CmdOp.end())it->second(); // 调用lambda函数elseexitcode_ = Unknown; // 如果没有找到操作，设置错误代码}void operator()(){Run();}std::string GetResult() // 这里可以使用stringstream{std::stringstream s;s << data1_ << oper_ << data2_ << "=" << result_ << " [code: " << exitcode_ << "]";std::string r;r = s.str();return r;}std::string GetTask(){std::stringstream s;s << data1_ << oper_ << data2_ << "=?";std::string r;r = s.str();return r;}~Task(){}private:int data1_; // 操作数int data2_; // 操作数char oper_; // 操作符int result_;   // 结果int exitcode_; // 错误码
};

main.cc

#include "blockqueue.hpp"
#include "Task.hpp"
#include <unistd.h>
#include <ctime>std::string TOHEX(pthread_t x)
{char buffer[128];snprintf(buffer, sizeof(buffer), "%p", x);return buffer;
}void *Consumer(void *args)
{BlockQueue<Task> *bq = static_cast<BlockQueue<Task> *>(args);while (true){// 消费Task t = bq->Pop();// t.Run(); // 之间调用t(); // 仿函数std::cout << "处理任务: " << t.GetTask() << " 运算结果是： " << t.GetResult() << " thread id: " << TOHEX(pthread_self()) << std::endl;// sleep(1);}
}void *Productor(void *args)
{int len = opers.size();BlockQueue<Task> *bq = static_cast<BlockQueue<Task> *>(args);while (true){// 生产int data1 = rand() % 10 + 1;int data2 = rand() % 10 + 1;usleep(10);char op = opers[rand() % len];Task t(data1, data2, op);bq->Push(t); // 入队std::cout << "生产了一个任务: " << t.GetTask() << " thread id: " << TOHEX(pthread_self()) << std::endl;sleep(1);}
}int main()
{srand(time(nullptr));// 在堆上上创建对象BlockQueue<Task> *bq = new BlockQueue<Task>();pthread_t c[3], p[5];for (int i = 0; i < 3; i++){pthread_create(c + i, nullptr, Consumer, bq);}for (int i = 0; i < 5; i++){pthread_create(p + i, nullptr, Productor, bq);}for (int i = 0; i < 3; i++){pthread_join(c[i], nullptr);}for (int i = 0; i < 5; i++){pthread_join(p[i], nullptr);}delete bq;return 0;
}

POSIX信号量

• 信号量的本质是一把计数器，那么这把计数器的本质是什么？？

  • 来描述资源数目的，把资源是否就绪放在了临界区之外，申请信号量时，其实就间接的已经在做判断了！

  • –P->原子的->申请资源

  • ++V->原子的->归还资源

• 信号量申请成功了，就一定保证会拥有一部分临界资源吗？

  • 只要信号量申请成功，就一定会获得指定的资源。

  • 申请mutex，只要拿到了锁，就可以获得临界资源，并且不担心被切换。

• 临界资源可以当成整体，可不可以看成一小部分一小部分呢？

  • 结合场景，一般是可以的

信号量：

• –p：1->0 ---- 加锁

• ++v: 0->1 ---- 释放锁

• 这样的叫做二元信号量 == 互斥锁

初始化信号量

#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()

基于环形队列的生产消费模型（代码验证）

• 环形队列中的生产者和消费者什么时候会访问同一个位置？

  • 当这两个同时指向同一个位置的时候，只有满or空的时候（互斥and同步）

  • 其它时候都指向的是不同的位置（并发）

因此，操作的基本原则：

① 空：消费者不能超过生产者 -->生产者先运行

② 满：生产者不能把消费者套一个圈里，继续再往后写入 -->消费者先运行

• 谁来保证这个基本原则呢？

  • 信号量来保证。

• 生产者最关心的是什么资源？

  • 空间

• 消费者最关心的是什么资源？

  • 数据

• 怎么保证，不同的线程，访问的是临界资源中不同的区域呢？

  • 通过程序员编码保证

makefile

myRingQueue:Main.ccg++ -o $@ $^ -std=c++11 -lpthread
.PHONY:clean
clean:rm -rf myRingQueue

RingQueue.hpp

#include <iostream>
#include <pthread.h>
#include <unistd.h>
#include <vector>
#include <semaphore.h>// 默认队列容量
const static int defaultcap = 5;template <class T>
class RingQueue
{
private:void Lock(pthread_mutex_t &mutex){pthread_mutex_lock(&mutex);}void UnLock(pthread_mutex_t &mutex){pthread_mutex_unlock(&mutex);}// -void P(sem_t &sem){sem_wait(&sem);}// +void V(sem_t &sem){sem_post(&sem);}public:RingQueue(int cap = defaultcap): ringqueue_(cap), cap_(cap), c_step_(0), p_step_(0){sem_init(&cdata_sem_, 0, 0);    // 消费者初始值为0sem_init(&pspace_sem_, 0, cap); // 生产者初始值为容量pthread_mutex_init(&c_mutex_, nullptr);pthread_mutex_init(&p_mutex_, nullptr);}// 生产void Push(const T &in){// 先申请空间--P(pspace_sem_); // P-  V+// 加锁Lock(p_mutex_);// 将数据存放ringqueue_[p_step_] = in;p_step_++;p_step_ %= cap_;// 解锁UnLock(p_mutex_);// data资源++V(cdata_sem_);}// 消费void Pop(T *out){// 申请资源--P(cdata_sem_);// 加锁Lock(c_mutex_);*out = ringqueue_[c_step_];c_step_++;c_step_ %= cap_;// 解锁UnLock(c_mutex_);// 释放空间++V(pspace_sem_);}~RingQueue(){sem_destroy(&cdata_sem_);sem_destroy(&pspace_sem_);pthread_mutex_destroy(&c_mutex_);pthread_mutex_destroy(&p_mutex_);}private:std::vector<T> ringqueue_; // 队列int cap_;                  // 队列的容量int c_step_; // 消费者下标int p_step_; // 生产者下标sem_t cdata_sem_;  // 消费者关注的数据资源sem_t pspace_sem_; // 生产者关注的空间资源pthread_mutex_t c_mutex_; // 消费者锁pthread_mutex_t p_mutex_; // 生产者锁
};

Task.hpp

#pragma once
#include <iostream>
#include <string>
#include <sstream>
#include <functional>
#include <map>std::string opers = "+-*/%";enum
{DivZero = 1,ModZero,Unknown
};class Task
{
public:Task(){}Task(int x, int y, char op): data1_(x), data2_(y), oper_(op){}void Run(){std::map<char, std::function<void()>> CmdOp{{'+', [this](){ result_ = data1_ + data2_; }},{'-', [this](){ result_ = data1_ - data2_; }},{'*', [this](){ result_ = data1_ * data2_; }},{'/', [this](){ if (data2_ == 0) exitcode_ = DivZero; else result_ = data1_ / data2_; }},{'%', [this](){ if (data2_ == 0) exitcode_ = ModZero; else result_ = data1_ % data2_; }}};// auto it = CmdOp.find(oper_);std::map<char, std::function<void()>>::iterator it = CmdOp.find(oper_);if (it != CmdOp.end())it->second(); // 调用lambda函数elseexitcode_ = Unknown; // 如果没有找到操作，设置错误代码}void operator()(){Run();}std::string GetResult() // 这里可以使用stringstream{std::stringstream s;s << data1_ << oper_ << data2_ << "=" << result_ << " [code: " << exitcode_ << "]";std::string r;r = s.str();return r;}std::string GetTask(){std::stringstream s;s << data1_ << oper_ << data2_ << "=?";std::string r;r = s.str();return r;}~Task(){}private:int data1_; // 操作数int data2_; // 操作数char oper_; // 操作符int result_;   // 结果int exitcode_; // 错误码
};

Main.cc

#include <iostream>
#include <pthread.h>
#include <ctime>
#include "RingQueue.hpp"
#include "Task.hpp"using namespace std;// 创建的线程数量
#define ProductorNum 5
#define ConsumerNum 5struct ThreadData
{RingQueue<Task> *rq;std::string threadname;
};void *Productor(void *args)
{ThreadData *td = static_cast<ThreadData *>(args);RingQueue<Task> *rq = td->rq;std::string name = td->threadname;int len = opers.size();while (true){// 1. 获取数据int data1 = rand() % 10 + 1;usleep(10);int data2 = rand() % 10 + 1;char op = opers[rand() % len];Task t(data1, data2, op);t();// 2. 生产数据rq->Push(t);cout << "Productor task done, task is : " << t.GetTask() << " who: " << name << endl;sleep(1);}return nullptr;
}
void *Consumer(void *args)
{ThreadData *td = static_cast<ThreadData *>(args);RingQueue<Task> *rq = td->rq;std::string name = td->threadname;while (true){// 消费数据Task t;rq->Pop(&t);t(); // 处理数据cout << "Consumer get task, task is : " << t.GetTask() << " who: " << name << " result: " << t.GetResult() << endl;// sleep(1);}return nullptr;
}int main()
{srand(time(nullptr) ^ getpid());RingQueue<Task> *rq = new RingQueue<Task>(5);pthread_t c[ConsumerNum], p[ProductorNum];for (int i = 0; i < ProductorNum; i++){ThreadData *td = new ThreadData();td->rq = rq;td->threadname = "Productor-" + std::to_string(i);pthread_create(p + i, nullptr, Productor, td);}for (int i = 0; i < ConsumerNum; i++){ThreadData *td = new ThreadData();td->rq = rq;td->threadname = "Consumer-" + std::to_string(i);pthread_create(c + i, nullptr, Consumer, td);}for (int i = 0; i < ProductorNum; i++){pthread_join(p[i], nullptr);}for (int i = 0; i < ConsumerNum; i++){pthread_join(c[i], nullptr);}return 0;
}

• 一个线程查看直观一些：

• 多个线程一起跑，打印就会显示的很乱

线程池

• 一种线程使用模式。线程过多会带来调度开销，进而影响缓存局部性和整体性能。而线程池维护着多个线程，等待着监督管理者分配可并发执行的任务。这避免了在处理短时间任务时创建与销毁线程的代价。线程池不仅能够保证内核的充分利用，还能防止过分调度。可用线程数量应该取决于可用的并发处理器、处理器内核、内存、网络sockets等的数量。

线程池的应用场景：

• ① 需要大量的线程来完成任务，且完成任务的时间比较短。 WEB服务器完成网页请求这样的任务，使用线程池技术是非常合适的。因为单个任务小，而任务数量巨大，你可以想象一个热门网站的点击次数。 但对于长时间的任务，比如一个Telnet连接请求，线程池的优点就不明显了。因为Telnet会话时间比线程的创建时间大多了。

• ② 对性能要求苛刻的应用，比如要求服务器迅速响应客户请求。

• ③ 接受突发性的大量请求，但不至于使服务器因此产生大量线程的应用。突发性大量客户请求，在没有线程池情况下，将产生大量线程，虽然理论上大部分操作系统线程数目最大值不是问题，短时间内产生大量线程可能使内存到达极限，出现错误。

线程池示例：

1. 创建固定数量线程池，循环从任务队列中获取任务对象

2. 获取到任务对象后，执行任务对象中的任务接口

---

• 下列是基于单例模式（懒汉模式）来创建的线程池

makefile

ThreadPool:Main.cppg++ -o $@ $^ -std=c++11 -lpthread
.PHONY:clean
clean:rm -f ThreadPool

ThreadPool.hpp

#pragma once
#include <iostream>
#include <queue>
#include <vector>
#include <pthread.h>
#include <unistd.h>static const int defaultnum = 5;struct ThreadInfo
{pthread_t tid;std::string name;
};template <class T>
class ThreadPool
{
private:void Lock(){pthread_mutex_lock(&mutex_);}void Unlock(){pthread_mutex_unlock(&mutex_);}void WakeUp(){pthread_cond_signal(&cond_);}std::string GetThreadName(pthread_t tid){for (const auto &ti : thread_){if (ti.tid == tid)return ti.name;}return "None";}bool IsQueueEmpty(){return task_.empty();}void ThreadSleep(){pthread_cond_wait(&cond_, &mutex_);}public:static void *HandlerTask(void *args){ThreadPool<T> *tp = static_cast<ThreadPool<T> *>(args);std::string name = tp->GetThreadName(pthread_self()); // 获取名字while (true){// 加锁tp->Lock();// 如果队列为空就将线程放到条件变量下等待while (tp->IsQueueEmpty()){tp->ThreadSleep();}// 取出节点T t = tp->Pop();// 解锁tp->Unlock();// 执行任务t();std::cout << name << " run, " << "result: " << t.GetResult() << std::endl;}}void Start(){for (int i = 0; i < thread_.size(); i++){thread_[i].name = "Thread-" + std::to_string(i);pthread_create(&(thread_[i].tid), nullptr, HandlerTask, this);}}void Push(T &t){Lock(); // 先加锁task_.push(t);WakeUp(); // 唤醒Unlock(); // 解锁}T Pop(){T t = task_.front();task_.pop();return t;}// 单例模式创建对象static ThreadPool<T> *GetInstance(){// 其他线程就不会每次执行加锁和解锁了~if (nullptr == tp_){// 加锁pthread_mutex_lock(&lock_);if (nullptr == tp_){std::cout << "log: singleton create done first!" << std::endl;tp_ = new ThreadPool<T>();}// 解锁pthread_mutex_unlock(&lock_);}return tp_;}public:ThreadPool(int num = defaultnum): thread_(num){pthread_mutex_init(&mutex_, nullptr);pthread_cond_init(&cond_, nullptr);}~ThreadPool(){pthread_mutex_destroy(&mutex_);pthread_cond_destroy(&cond_);}private:std::vector<ThreadInfo> thread_; // 任务线程std::queue<T> task_;             // 任务队列pthread_mutex_t mutex_; // 锁pthread_cond_t cond_;   // 条件变量// 单例模式，只能创建一个对象static ThreadPool<T> *tp_;static pthread_mutex_t lock_;
};// 类外进行初始化
template <class T>
ThreadPool<T> *ThreadPool<T>::tp_ = nullptr;template <class T>
pthread_mutex_t ThreadPool<T>::lock_ = PTHREAD_MUTEX_INITIALIZER;

Main.cpp

#include <iostream>
#include <ctime>
#include "Task.hpp"
#include "ThreadPool.hpp"int main()
{// 创建线程ThreadPool<Task> *tp = new ThreadPool<Task>(5);// 启动线程ThreadPool<Task>::GetInstance()->Start();srand(time(nullptr) ^ getpid());while (true){// 1. 构建任务int x = rand() % 10 + 1;usleep(10);int y = rand() % 10 + 1;char op = opers[rand() % opers.size()];Task t(x, y, op);// 2. 交给线程池进行处理ThreadPool<Task>::GetInstance()->Push(t);std::cout << "main thread make task: " << t.GetTask() << std::endl;sleep(1);}return 0;
}

• 这样就创建起来跑起来了~~

简单封装一下pthread库

#include <iostream>
#include <string>
#include <ctime>
#include <pthread.h>typedef void (*callback_t)();
static int num = 1;class Thread
{
private:void Entery(){cb_();}static void *Routine(void *args){Thread *thread = static_cast<Thread *>(args);thread->Entery();return nullptr;}public:Thread(callback_t cb): tid_(0), name_(""), start_timestamp_(0), isrunning_(false), cb_(cb){}void Run(){name_ = "thread-" + std::to_string(num++);start_timestamp_ = time(nullptr);isrunning_ = true;pthread_create(&tid_, nullptr, Routine, this);}void Join(){pthread_join(tid_, nullptr);isrunning_ = false;}std::string Name(){return name_;}uint64_t StartTimestamp(){return start_timestamp_;}bool IsRunning(){return isrunning_;}private:pthread_t tid_;std::string name_;uint64_t start_timestamp_;bool isrunning_;// 回调函数callback_t cb_;
};

STL、智能指针和线程安全

STL中的容器不是线程安全的

原因：

• STL 的设计初衷是将性能挖掘到极致, 而一旦涉及到加锁保证线程安全, 会对性能造成巨大的影响。

• 而且对于不同的容器, 加锁方式的不同, 性能可能也不同(例如hash表的锁表和锁桶)。

• 因此 STL 默认不是线程安全. 如果需要在多线程环境下使用, 往往需要调用者自行保证线程安全。

智能指针是否是线程安全的

• 对于 unique_ptr, 由于只是在当前代码块范围内生效, 因此不涉及线程安全问题.

• 对于 shared_ptr, 多个对象需要共用一个引用计数变量, 所以会存在线程安全问题. 但是标准库实现的时候考虑到了这个问题, 基于原子操作(CAS)的方式保证 shared_ptr 能够高效, 原子的操作引用计数。

常见的锁

• 悲观锁：在每次取数据时，总是担心数据会被其他线程修改，所以会在取数据前先加锁（读锁，写锁，行锁等），当其他线程想要访问数据时，被阻塞挂起。

• 乐观锁：每次取数据时候，总是乐观的认为数据不会被其他线程修改，因此不上锁。但是在更新数据前，会判断其他数据在更新前有没有对数据进行修改。主要采用两种方式：版本号机制和CAS操作。

• CAS操作：当需要更新数据时，判断当前内存值和之前取得的值是否相等。如果相等则用新值更新。若不等则失败，失败则重试，一般是一个自旋的过程，即不断重试。

• 挂起等待锁：当某个线程没有申请到锁的时候，此时该线程会被挂起，即加入到等待队列等待。当锁被释放的时候，就会被唤醒，重新竞争锁。当临界区运行的时间较长时，我们一般使用挂起等待锁。我们先让线程PCB加入到等待队列中等待，等锁被释放时，再重新申请锁。

之前所学的互斥锁就是挂起等待锁

• 自旋锁：当某个线程没有申请到锁的时候，该线程不会被挂起，而是每隔一段时间检测锁是否被释放。如果锁被释放了，那就竞争锁；如果没有释放，过一会儿再来检测。如果这里使用挂起等待锁，可能线程刚加入等待队列，锁就被释放了，因此，当临界区运行的时间较短时，我们一般使用自旋锁。

pthread_spin_lock();

• 自旋锁只需要把mutex变成spin。

读者写者问题

• 在编写多线程的时候，有一种情况是十分常见的。那就是，有些公共数据修改的机会比较少。相比较改写，它们读的机会反而高的多。通常而言，在读的过程中，往往伴随着查找的操作，中间耗时很长。给这种代码段加锁，会极大地降低我们程序的效率。那么有没有一种方法，可以专门处理这种多读少写的情况呢？ 有，那就是读写锁。

• 3种关系：写者和写者（互斥），读者和读者（没有关系），读者和写者（互斥关系）

• 2种角色：读者、写者

• 1个交易场所：读写场所

---

• 读者写者 vs 生产者消费者
  • 本质区别：消费者会把数据拿走，而读者不会

---

初始化：

int pthread_rwlock_init(pthread_rwlock_t *restrict rwlock, const pthread_rwlockattr_t *restrict attr);

销毁：

int pthread_rwlock_destroy(pthread_rwlock_t *rwlock);

加锁和解锁：

int pthread_rwlock_rdlock(pthread_rwlock_t *rwlock);  // 读者加锁
int pthread_rwlock_wrlock(pthread_rwlock_t *rwlock); // 写者加锁int pthread_rwlock_unlock(pthread_rwlock_t *rwlock); // 读写者解锁

• 设置读写优先：

  • 分为读者优先和写者优先。

• 读者写者进行操作的时候，读者非常多，频率特别高，写者比较少，频率不高

• 存在写者饥饿的情况

代码实现的较为简单的读者写者问题

#include <iostream>
#include <unistd.h>
#include <pthread.h>int board = 0;pthread_rwlock_t rw;using namespace std;void *reader(void *args)
{const char *name = static_cast<const char*>(args);cout << "run..." << endl;while(true){pthread_rwlock_rdlock(&rw);cout << "reader read : " << board << "tid: " << pthread_self() << endl;sleep(10);pthread_rwlock_unlock(&rw);}
}void *writer(void *args)
{const char *name = static_cast<const char*>(args);sleep(1);while(true){pthread_rwlock_wrlock(&rw);board++;cout << "I am writer" << endl;sleep(10);pthread_rwlock_unlock(&rw);}
}int main()
{pthread_rwlock_init(&rw, nullptr);pthread_t r1,r2,r3,r4,r5,r6, w;pthread_create(&r1, nullptr, reader, (void*)"reader");pthread_create(&r2, nullptr, reader, (void*)"reader");pthread_create(&r3, nullptr, reader, (void*)"reader");pthread_create(&r4, nullptr, reader, (void*)"reader");pthread_create(&r5, nullptr, reader, (void*)"reader");pthread_create(&r6, nullptr, reader, (void*)"reader");pthread_create(&w, nullptr, writer, (void*)"writer");pthread_join(r1, nullptr);pthread_join(r2, nullptr);pthread_join(r3, nullptr);pthread_join(r4, nullptr);pthread_join(r5, nullptr);pthread_join(r6, nullptr);pthread_join(w, nullptr);pthread_rwlock_destroy(&rw);return 0;
}

struct rwlock_t
{int readers = 0;int who;mutex_t mutex;
}

读者：

ptjread_rwlock_rdlock()
lock(mutex)
readers++
(unlock)mutex

read操作：

lock(mutex)
readers--;
unlock(mutex) 

写者：

pthread_rwlock_wrlock()
lock(mutex)
while(readers > 0) 释放锁, wait

write操作：

unlock(mutex)

Linux系统编程到此结束end…