Linux：多线程中的互斥与同步

多线程

线程互斥
- 互斥锁
- 互斥锁实现的原理
- - 封装原生线程库
  - 封装互斥锁
- 死锁
- - 避免死锁的四种方法
线程同步
- 条件变量

线程互斥

在多线程中，如果存在有一个全局变量，那么这个全局变量会被所有执行流所共享。但是，资源共享就会存在一种问题：并发访问

什么是并发访问呢？

好比学校的教室，教室属于共享的资源。如果没有安排课的话，那么任何人都可以使用这间教室。假设某一天刚好就是本班老师要补课占用这间教室，没有提前跟教务处报备。恰好此时此刻又有其他班想要用此间教室，在大家都没有商量的情况下，都在这间教室上课，这不得乱套了。并发就相当于两个班同时使用这间教室。

下面来介绍几个概念：

互斥：多个进程或线程在访问共享资源时不会发生冲突
临界资源：一次仅允许一个进程或线程访问的资源=
临界区：进程中访问临界资源的那段代码
原子性：一个事务或操作的所有步骤作为一个整体执行，这些步骤要么全部执行，要么全部不执行

临界资源于临界区是属于匹配的关系

回到上面提到的，共享资源的存在，没有做特殊处理的话，会引发并发访问，使得多线程造成数据不一致的问题。

下面来举个例子：实现一个线程池，让这些线程去抢票（票数 1000 是共享资源）

#include <pthread.h>
#include <iostream>
#include <cstdio>
#include <unistd.h>
#include <string>

int tickets = 1000;

void *threadRoutine(void *arg)
{
    string name = static_cast<char*>(arg);//类型转换+构造

    while(true)
    {
        if(tickets > 0)
        {
            //模仿抢票时间
            usleep(2000);//usleep：微秒(秒、毫秒、微秒、纳秒)
            cout << name << " get a ticket: " << tickets-- << endl; 
        }
        else
        {
            break;
        }
    }
}

int main()
{
    //创建4个子线程
    pthread_t t[4];
    int n = sizeof(t) / sizeof(t[0]);
    for(int i = 0; i < n; i++)
    {
        char* tname = new char[64];
        snprintf(tname, 64, "thread -> %d", i + 1);
        pthread_create(t + i, nullptr, threadRoutine, tname);
    }

    //等待线程
    for(int i = 0; i < n; i++)
    {
        pthread_join(t[i], nullptr);
    }
    return 0;
}

运行查看效果：
在这里插入图片描述
当然中间的抢票过程不重要，重要的是结果。从上面的结果来看，当票数为0的时候还有线程继续执行抢票的操作，导致票变成负数。这个合理吗？很不合理。这就是并发带来的坏处。

为了避免并发访问造成的数据不一致，需要对共享资源做保护处理，被保护的资源也被称为 临界资源。任何一个线程都有属于自己的代码会去访问临界资源，这些代码也可以称为 临界区；同样的，在线程中没有访问到临界资源的代码被称为 非临界区。

可以说 临界资源 是衡量 共享资源 的；而 临界区 是衡量 线程代码的！

如何对共享资源进行保护呢？可以通过加锁的方式保护共享资源

互斥锁

下面来介绍几个关于互斥锁的API接口，使用互斥锁需要包含头文件：#include <pthread.h>

一般的，要定义一个互斥锁需要用到数据类型：pthread_mutex_t

pthread_mutex_t mutex; //定义一个互斥锁对象

初始化互斥锁

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

参数介绍：

mutex：指向初始化的互斥锁对象的指针
attr：设置互斥锁的属性
返回值：初始化成功返回0，失败错误码被设置

销毁互斥锁

 int pthread_mutex_destroy(pthread_mutex_t *mutex);

参数介绍：

mutex：指向想要销毁的互斥锁
返回值：销毁成功返回0，失败错误码被设置

关于初始化互斥锁和销毁互斥锁这里需要注意一点：

如果定义了一个全局的或者静态的互斥锁对象，可以通过使用宏:PTHREAD_MUTEX_INITIALIZER来初始化这个互斥锁对象，并且这个互斥对象是不需要手动去销毁的

前提必须是 全局的 或是 静态的 互斥锁对象

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

上锁

int pthread_mutex_lock(pthread_mutex_t *mutex);

解锁

int pthread_mutex_unlock(pthread_mutex_t *mutex);

通过上面的互斥锁的提供的API，将上面的抢票代码稍作修改，将上面的互斥锁同线程名封装成 TDate 类，让多个线程用到同一个锁：

#include <pthread.h>
#include <iostream>
#include <cstdio>
#include <cstring>
#include <unistd.h>
#include <string>

using namespace std;

class TDate
{
public:
    //构造
    TDate(const char* tname, pthread_mutex_t *mutex)
        :_tname(tname),
        _mutex(mutex)
        {}

    ~TDate(){}

public:
    string _tname;
    pthread_mutex_t *_mutex;
};

int tickets = 1000;

void *threadRoutine(void *arg)
{
    TDate* td = static_cast<TDate*>(arg);//类型转换+构造

    while(true)
    {
        //将票数（共享资源）,上锁保护
        pthread_mutex_lock(td->_mutex);
        if(tickets > 0)
        {
            //模仿抢票时间
            usleep(2000);//usleep：微秒(秒、毫秒、微秒、纳秒)
            cout << td->_tname << " get a ticket: " << tickets-- << endl; 

            //抢票成功，执行解锁操作
            pthread_mutex_unlock(td->_mutex);
        }
        else
        {
            pthread_mutex_unlock(td->_mutex);   //先解锁再跳出循环
            break;
        }
    }
}

int main()
{
    //定义互斥锁对象
    pthread_mutex_t mutex;
    //初始化互斥锁
    pthread_mutex_init(&mutex, nullptr);

    //创建4个子线程
    pthread_t t[4];
    int n = sizeof(t) / sizeof(t[0]);
    for(int i = 0; i < n; i++)
    {
        char tname[64] = { 0 };
        snprintf(tname, sizeof(char)*64, "thread -> %d", i + 1);
        TDate* td = new TDate(tname, &mutex);

        pthread_create(t + i, nullptr, threadRoutine, td);
    }
    //等待线程
    for(int i = 0; i < n; i++)
    {
        pthread_join(t[i], nullptr);
    }
    //销毁互斥锁
    pthread_mutex_destroy(&mutex);

    return 0;
}

再来看看运行结果：
在这里插入图片描述

上锁其实就是：当一个线程访问一块临界区时，将其他要访问这块临界区的线程阻塞起来；解锁就是：将临界资源重新开放，让所有的线程可以重新访问到临界资源。通过上锁、解锁的操作，就可以很好的解决并发的问题。

对于互斥锁，需要注意以下细节：

凡是访问同一个临界资源的线程，都要进行加锁保护，而且必须加同一把锁，不能有例外
加锁本质是给 临界区 加锁，加锁的粒度尽量要细一些（加锁的代码不宜太多）
由于线程都必须看到同一把锁，锁本身就是公共资源，对此加锁和解锁操作就是原子性的
在临界区线程是有可能被切换的，但是切换线程不影响临界区的资源被修改

这是因为锁只有一把，由于锁提前被切走的线程先申请了，对于其他线程是不能再次申请锁资源。至此，其他线程无法对临界区的资源做任何修改

互斥锁实现的原理

在计算机体系结构中存在两个指令：swap 和 exechange，这两个指令作用是把寄存器和内存单元的数据做交换

例如：

加锁的过程是原子性的。如何去理解这句话呢？

下面来看一段伪代码：
在这里插入图片描述
上面的伪代码中，lock 相当于 pthread_mutex_lock()；unlock 就相当于 pthread_mutex_unloc()。为了方便描述，我们将定义一个互斥锁对象：pthread_mutex_t mutex，这里的 mutex 我们可以看成就定义了一个大于0的整数。

在多线程中，上面加锁和解锁的操作，是由线程来调用的，谁先来谁先调用。

由于互斥锁是共享资源，定义了一把锁，就会在内存中开辟对应的空间，这里的mutex内容假设为1：
在这里插入图片描述

由于寄存器硬件只有一套，寄存器内部的数据是每个线程都有的（就是线程数据内容有多份，但是不能说寄存器有多个）

寄存器个数！= 寄存器内容（线程的上下文）

好比图书馆内，有很多的公共座位。当你去自习学习，带上自己的学习工具。学习工具是属于个人的，座位是公共的。当每次学习时，别人是不能占有你的学习工具。当你学习完要离开，你只能带走你的学习资料，但是板凳座椅还是在那可以被其他人使用。这里的学习工具就是寄存器内容，座椅板凳就是寄存器。

说了这么多铺垫，回过来看看前面的伪代码：

在这里插入图片描述

假设存在两个线程：threadA、threadB。线程A比较快，先申请了锁资源，调用了 lock API。

先执行第一条伪代码：movb $0, %al，线程A向寄存器 al 中写入数据 0
在这里插入图片描述

接下来走到第二条伪代码：xchgb %al，mutex （xchgb就是开始提到的 exchange 指令），将内存中的 mutex 值 1，交换到 CPU 中的 al 寄存器中。

在这里插入图片描述

由于 mutex 是共享资源，线程A 将 mutex 互斥锁对象的共享资源交换到寄存器中，其实就是将共享资源数据交换到自己的私有上下文中。可以说：交换的本质就是加锁，由于 exchange 指令是一条汇编，所以说 加锁是原子性的

下面执行第三步：判断寄存器 al 内部的值
在这里插入图片描述

如果 al 内部的值大于0，那么直接结束 return 返回。

但是，有没有一种可能，就是此时正要执行第三步的时候，线程A时间片到了，直接切换到线程B 运行。前面提到过，就是当一个线程切换到另一个线程的时候，寄存器内容是要被切换的线程一起带走的，这就导致共享资源的 mutex 内部原来的 1 变成 0 。

接下来，线程B来了哈，执行加锁的操作。

第一步先将 al 寄存器值设置为0；第二步将寄存器与内存中的 mutex 内容进行交换，这下交换是什么，交换的是 0 啊。
在这里插入图片描述
mutex 内部的值是 0 ，原因是线程A被切走，线程A 没有回来。此刻，线程B 进行执行后续的代码，寄存器 al 内部的值为0，阻塞挂起。不管后续来多少的线程，都会跟线程B一样，被阻塞着。直到线程A被调度回来，线程A会将原有的上下文带回，并且进行执行被调度走后的代码。

加锁明白了，解锁就更加简单了：
在这里插入图片描述

在诸多线程中，只有线程A没有被阻塞。线程A执行完对应的临界区代码后，需要解锁操作。将内存中的 mutex 值交换回1。此时，其他线程再继续执行的时候内存中的 mutex 值就不是0了，继续可以向后执行并发操作。

加锁和解锁的代码是线程在调用，说白了就是谁用锁谁调用。加锁与解锁其实就是一种，让没有调用互斥锁的线程通不通过的策略

封装原生线程库

模拟封装一个简化版的C++线程库：

实现 Thread 类，类中的成员函数调用原生线程库的API。具体实现的内容如下：

#include <iostream>
#include <pthread.h>
#include <string>
#include <cstdlib>

class Thread
{
public:
    // 定义枚举类型
    typedef enum
    {
        NEW = 0,
        RUNNING,
        EXITED
    } ThreadStatus;

    typedef void* (*func_t)(void*); // 函数指针

public:
    //构造
    Thread(int num, func_t func, void* args)
        : _tid(0),
          _status(NEW),
          _func(func),
          _args(args)
    {
        char name[128];
        snprintf(name, sizeof(name), "thread->%d", num);
        _name = name;
    }

    // 获取线程状态
    int status()
    {
        return _status;
    }
    // 获取线程名称
    const std::string &threadname()
    {
        return _name;
    }
    // 获取线程tid：线程运行才返回，否则返回0
    const std::string threadid()
    {
        if (_status == RUNNING)
            return stringID();
        else
            return "0";
    }

    // 线程id转换为16进制，以字符串输出
    const std::string stringID()
    {
        char id[64];
        snprintf(id, sizeof(id), "0x%x", _tid);
        return id;
    }
    //利用静态成员函数,防止默认参数（this）生成
    static void* runHelper(void* args)
    {
        Thread* ts = static_cast<Thread*>(args);//类型转换
        (*ts)(); //函数对象

        return nullptr;
    }

    void operator()() //仿函数
    {
        if(_func != nullptr) _func(_args);
    }

    //线程执行
    void run()
    {
        // 创建线程
        // int n = pthread_create(&_tid, nullptr, runHelper, nullptr);
        int n = pthread_create(&_tid, nullptr, runHelper, this);//传入this，Thread本身

        if(n != 0) exit(1);    
        // 更改线程状态
        _status = RUNNING;
    }
    // 等待线程
    void join()
    {
        int n = pthread_join(_tid, nullptr);
        if (n != 0) 
        {
            std::cerr << "main thread join thread " << _name << " error" << std::endl;
            return;
        }
        _status = EXITED;
    }

    ~Thread() {}
private:
    pthread_t _tid;       // 线程id
    std::string _name;    // 线程名
    func_t _func;         // 线程未来要执行的回调函数
    void* _args;           //回调函数的参数
    ThreadStatus _status; // 线程所处的状态
};

封装互斥锁

实现两个类：Mutex类和 lockGuard类

Mutex 类中的成员函数主要用于调用原生线程库中的互斥锁API；而lockGuard 主要实现构造与析构，其类内部包含 Mutex 对象的成员。

#include <iostream>
#include <pthread.h>
#include <string>

class Mutex
{
public:
    Mutex(pthread_mutex_t *mutex)
        : _mutex(mutex)
    {}

    void lock()
    {
        // 上锁
        pthread_mutex_lock(_mutex);
    }

    void unlock()
    {
        // 解锁
        pthread_mutex_unlock(_mutex);
    }

    ~Mutex() {}

private:
    pthread_mutex_t *_mutex;
};

class lockGuard
{
public:
    //利用RAII思想
    lockGuard(pthread_mutex_t *mutex)
        : _mutex(mutex)
    {
        _mutex.lock();
    }

    ~lockGuard()
    {
        _mutex.unlock();
    }

private:
    Mutex _mutex;
};

lockGuard 类对象利用RAII思想，只需要实例化出lockGuard 对象后，使共享资源保护起来形成临界资源。不需要手动去上锁解锁操作

示例：

#include "lockGuard.hpp"
#include "Thread.hpp"

int tickets = 1000;
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER; //定义全局的互斥锁

void *threadRoutine(void *arg)
{
    string message  = static_cast<const char*>(arg);

    while(true)
    {
        lockGuard lockguard(&mutex); //利用RAII思想
        
        if(tickets > 0)
        {
            usleep(2000);//usleep：微秒
            cout << message << " get a ticket: " << tickets-- << endl;
        }
        else
        {
            break;
        }
    }
}

int main()
{
    Thread t1(1, threadRoutine, (void*)"hello world");
    Thread t2(2, threadRoutine, (void*)"hello world");
    Thread t3(3, threadRoutine, (void*)"hello world");
    Thread t4(4, threadRoutine, (void*)"hello world");

    t1.run();
    t2.run();
    t3.run();
    t4.run();

    t1.join();
    t2.join();
    t3.join();
    t4.join();

    return 0;
}

运行结果如下：

在这里插入图片描述

死锁

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

举个简单的例子：

有一天，有两个小朋友（小明和小强）都去小卖部买棒棒。这两个小朋友身上都有 5 毛钱。来到小卖部后，老板说：一个棒棒糖要 1 元。此时，小明说：小强要不你把你的 5 毛给我，我就可以买这个棒棒糖吃啦！小强一听肯定不行，说到：那你为什么不能给我你的5毛钱，我买来吃呢？谁也不让谁的状态，也就是死锁状态。举例的小明和小强就是两个线程、5 毛钱就是锁、棒棒糖就是临界资源、老板就是操作系统。

产生死锁的四个必要条件：

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

不剥夺的条件就是上面举例的，小明和小强不能为买棒棒糖的事情去抢对方的钱。

引入多线程就是为了高效的工作，但是多线程就会产生并发问题；为了解决并发问题，就引入了互斥锁，但是互斥锁的存在就会导致死锁的产生；对此，引入了一个解决问题的办法，往往就会滋生另一个问题的产生。

那么如何避免死锁呢？

避免死锁的四种方法

避免死锁的核心思想就是：破坏产生死锁的4个条件

不加锁：不会产生互斥条件
主动释放锁：不会产生请求与保护条件（不争锁资源）
按照顺序申请锁
控制线程统一释放锁（剥夺锁资源）

以代码的方式来解释一下第四点：

#include <pthread.h>
#include <iostream>
#include <cstdio>
#include <cstring>
#include <unistd.h>
#include <string>

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

void *threadRoutine(void *arg)
{
    cout << "I am a new thread " << endl;

    pthread_mutex_lock(&mutex);
    cout << "I got a mutex!" << endl;

    // 再次申请锁会发生阻塞
    pthread_mutex_lock(&mutex);
    cout << "I alive again" << endl;

    return nullptr;
}

int main()
{
    pthread_t t1;
    int n = pthread_create(&t1, nullptr, threadRoutine, nullptr);
    if (n != 0)
        exit(1);
    sleep(3);

    // 主线程
    cout << "main thread run begin" << endl;
    pthread_mutex_unlock(&mutex); //主线程释放锁
    cout << "main thread unlock..." << endl;

    sleep(3);
    return 0;
}

来看看运行结果：
在这里插入图片描述

不同线程申请锁资源是可以被其他线程释放的。为了避免死锁，可以将锁资源进行剥夺处理，也就是控制线程统一释放锁。

线程同步

一个线程访问临界区过程一般是：申请锁，访问资源，再是释放锁。

如果在一个多线程程序中，一个线程一直就是执行这样的操作：优先申请了锁，在访问完临界区代码后，释放锁。在释放锁后，即刻又申请锁（中间没有间隔时间），访问临界资源，释放锁。循环反复，就会造成一个结果每次都只有这个线程在访问这个临界区，其他线程一直处于阻塞状态。这样的现象被称为：饥饿

互斥规则也是为了避免并发问题的产生，保证了数据的正确性。但是，互斥的规则也有不合理性，犹如上面提到的。

对此，需要对原有的规则进行修改：

同步：在保证数据的安全性前提下，要让线程能够按照某种特定的顺序访问临界资源

线程同步是为了解决多线程中的饥饿问题，让多线程进行协同工作

条件变量

条件变量：在多线程中用于实现线程间同步的一种机制

条件变量主要用于实现等待到唤醒的逻辑；条件变量本身不是锁，它通常是配合互斥锁一起使用。

下面来介绍一些关于环境变量的接口：使用条件变量需要包含头文件 #include <pthread.h>

条件变量的初始化

条件变量的初始化有两种方式，跟互斥锁差不多

第一种：通过宏的方式初始化全局的条件变量，

 pthread_cond_t cond = PTHREAD_COND_INITIALIZER;

第二种：通过条件变量的 pthread_cond_init API 来初始化

int pthread_cond_init(pthread_cond_t *restrict cond,
             		 const pthread_condattr_t *restrict attr);

销毁条件变量

int pthread_cond_destroy(pthread_cond_t *cond);

让线程处于等待状态

int pthread_cond_wait(pthread_cond_t *restrict cond,
              		 pthread_mutex_t *restrict mutex);

唤醒线程

唤醒线程的方式有两种：

将处于阻塞队列的线程一个一个唤醒

int pthread_cond_signal(pthread_cond_t *cond);

直接唤醒全部的线程

int pthread_cond_broadcast(pthread_cond_t *cond);

下面来举个示例代码：

#include <iostream>
#include <pthread.h>
#include <unistd.h>

using namespace std;

#define NUM 5

//初始化互斥锁
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

void* active(void* args)
{
    string name = static_cast<const char*>(args);

    while(true)
    {
        //上锁
        pthread_mutex_lock(&mutex);
        cout << name << "活动" << endl;
        //解锁
        pthread_mutex_unlock(&mutex);
    }
}

int main()
{
    //创建线程池
    pthread_t tids[NUM];
    for(int i = 0; i < NUM; i++)
    {
        char* name = new char[64];
        snprintf(name, sizeof(name), "thread->%d", i + 1);
        pthread_create(tids + i, nullptr, active, name);
    }


    //等待线程池
    for(int i = 0; i < NUM; i++)
    {
        pthread_join(tids[i], nullptr);
    }

    return 0;
}

创建一个线程池，其中包含5个线程，每个线程都会去回调 active 函数。其中，对active的打印代码进行上锁与解锁操作，在这里直接设置了死循环。设置死循环是为了模拟其中一个线程频繁对锁资源的申请，使其他线程处于饥饿的状态。

来看看执行效果：
在这里插入图片描述

下面来修改代码，利用条件变量 API，当线程申请锁后，让其处于等待状态。在主线程中，再将一个个线程唤醒：

#include <iostream>
#include <pthread.h>
#include <unistd.h>

using namespace std;

#define NUM 5

//初始化条件变量
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
//初始化互斥锁
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

void* active(void* args)
{
    string name = static_cast<const char*>(args);

    while(true)
    {
        //上锁
        pthread_mutex_lock(&mutex);
        //利用条件变量使线程处于阻塞状态
        pthread_cond_wait(&cond, &mutex);

        cout << name << "活动" << endl;

        //解锁
        pthread_mutex_unlock(&mutex);
    }
}

int main()
{
    //创建线程池
    pthread_t tids[NUM];
    for(int i = 0; i < NUM; i++)
    {
        char* name = new char[32];
        snprintf(name, 32, "thread->%d", i + 1);
        pthread_create(tids + i, nullptr, active, name);
    }

    sleep(2);

    //主进程
    while(true)
    {
        cout << "main thread wakeup thread..." << endl;
        pthread_cond_signal(&cond); //唤醒单个线程
        sleep(1);
    }

    //等待线程池
    for(int i = 0; i < NUM; i++)
    {
        pthread_join(tids[i], nullptr);
    }

    return 0;
}