【Linux】多线程（中）

一、线程互斥

1.1 互斥概念

在前面讲System V信号量的时候

【Linux】进程间通信——System V消息队列和信号量_linux semaphore信号量进程间通信-CSDN博客https://blog.csdn.net/Eristic0618/article/details/142635584?spm=1001.2014.3001.5501我们已经接触过互斥这一概念了，这里我们再重温一下相关的概念

• 临界资源：多线程执行流共享的资源
• 临界区：访问临界资源的代码
• 互斥：任何时刻，有且只有一个执行流进入临界区访问临界资源
• 原子性：不会被任何调度机制打断的操作，要么未开始要么已完成

1.2 互斥量mutex

对于线程内部创建的局部变量，其存储在线程私有的栈空间内，因此无法被其他线程所访问

但有时线程还需要访问全局变量等共享资源，如果多个线程同时访问这些共享资源，且没有任何的保护机制，就可能导致问题的发生

例如我们模拟一个抢票的场景：

#include <iostream>
#include <sys/types.h>
#include <unistd.h>
#include <pthread.h>

using namespace std;

int ticket = 100; //100张票

void* threadRoutine(void* arg) //新线程的例程
{
    string name = (char*)arg;
    while(true)
    {
        if(ticket > 0)
        {
            usleep(1000); //模拟抢票动作
            cout << name << " get ticket:" << ticket << endl;
            ticket--;
        }
        else
            break;
    }
    return nullptr;
}

int main()
{
    pthread_t t1, t2, t3, t4;
    pthread_create(&t1, nullptr, threadRoutine, (void*)"Thread 1");
    pthread_create(&t2, nullptr, threadRoutine, (void*)"Thread 2");
    pthread_create(&t3, nullptr, threadRoutine, (void*)"Thread 3");
    pthread_create(&t4, nullptr, threadRoutine, (void*)"Thread 4");

    pthread_join(t1, nullptr);
    pthread_join(t2, nullptr);
    pthread_join(t3, nullptr);
    pthread_join(t4, nullptr);
    return 0;
}

运行结果：

可以看到我们的票数到最后已经变为了一个非法的值，说明在不加保护的场景下，多线程竞争式的访问临界资源可能导致问题。为什么？

• 线程在if判断为真后进入临界区，这个过程中代码可以并发的切换到其他的进程
• usleep模拟我们的抢票过程，即便只有1000微秒，对于CPU而言也是一个十分漫长的时间了。在这个过程中可能有很多个线程也会进入到临界区中
• 减少票数的动作本身也不是一个原子操作，看似只有一条代码，实际对应了三条汇编指令

多个因素导致了多个线程并发进行“抢票”动作时，票数最后变为了负数

要解决这个问题，就必须保证多个线程互斥的进行抢票动作，即同一时刻有且只能有一个线程进入代码临界区访问临界资源。如何做到？我们需要一把“锁”，即互斥量mutex

临界区就像一个小房间，一开始房间的门没有锁，于是大家都一拥而上进入房间。有了互斥量，房门就上了锁，并且同一时刻只允许一个线程能够开锁，线程从房间出来后就要把锁的钥匙归还

1.3 互斥量相关API

（1）初始化互斥量

我们可以通过两种方式初始化互斥量：

• 静态初始化

pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;

使用名为PTHREAD_MUTEX_INITIALIZER的宏，在创建互斥量的同时对其进行初始化

• 动态初始化

使用pthread_mutex_init函数对互斥量进行初始化

#include <pthread.h>

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

其中mutex为要初始化的互斥量，attr设置为nullptr即可，代表使用默认的互斥锁属性

（2）销毁互斥量

#include <pthread.h>

int pthread_mutex_destroy(pthread_mutex_t *mutex);

pthread_mutex_destroy函数用于销毁一个互斥量，其中：

• 使用静态初始化方式的互斥量不需要销毁
• 不要销毁一个已经加锁的互斥量
• 已经销毁的互斥量，要确保不会再有线程对该互斥量加锁

（3）互斥量加锁和解锁

#include <pthread.h>

int pthread_mutex_lock(pthread_mutex_t *mutex);
int pthread_mutex_unlock(pthread_mutex_t *mutex);

调用pthread_mutex_lock函数的线程将会尝试对目标互斥量加锁

• 此时互斥量处于未锁状态，那么线程能成功锁定该互斥量
• 如果互斥量已被其他线程锁定或线程竞争互斥量失败，那么线程就会陷入阻塞并等待互斥量解锁

持有锁的线程调用pthread_mutex_unlock函数后可以对互斥量解锁

将互斥锁引入我们上面的售票系统后：

#include <iostream>
#include <sys/types.h>
#include <unistd.h>
#include <pthread.h>

using namespace std;

int ticket = 100; //100张票
pthread_mutex_t lock; //互斥锁

void* threadRoutine(void* arg) //新线程的例程
{
    string name = (char*)arg;
    while(true)
    {
        pthread_mutex_lock(&lock); //互斥量加锁
        if (ticket > 0)
        {
            usleep(1000); //模拟抢票动作
            cout << name << " get ticket:" << ticket << endl;
            ticket--;
            pthread_mutex_unlock(&lock); //互斥量解锁
        }
        else
        {
            pthread_mutex_unlock(&lock); //互斥量解锁
            break;
        }
    }
    return nullptr;
}

int main()
{
    pthread_mutex_init(&lock, nullptr); //初始化互斥锁

    pthread_t t1, t2, t3, t4;
    pthread_create(&t1, nullptr, threadRoutine, (void*)"Thread 1");
    pthread_create(&t2, nullptr, threadRoutine, (void*)"Thread 2");
    pthread_create(&t3, nullptr, threadRoutine, (void*)"Thread 3");
    pthread_create(&t4, nullptr, threadRoutine, (void*)"Thread 4");

    pthread_join(t1, nullptr);
    pthread_join(t2, nullptr);
    pthread_join(t3, nullptr);
    pthread_join(t4, nullptr);

    pthread_mutex_destroy(&lock); 
    return 0;
}

运行结果：

引入互斥量后，会发现只有一个线程不停执行抢票操作，原因：

解锁后的线程离互斥锁”最近“，其竞争能力相比其他线程更强，因此能够更快对互斥量上锁，导致其他线程一直处于阻塞状态

在实际情况下，我们抢完票后也不会立即抢下一张票，在这过程中需要一个时间窗口。因此我们在线程解锁后可以让其睡眠一段时间，模拟抢完票后的动作

#include <iostream>
#include <sys/types.h>
#include <unistd.h>
#include <pthread.h>

using namespace std;

int ticket = 100; //100张票
pthread_mutex_t lock; //互斥锁

void* threadRoutine(void* arg) //新线程的例程
{
    string name = (char*)arg;
    while(true)
    {
        pthread_mutex_lock(&lock); //互斥量加锁
        if (ticket > 0)
        {
            usleep(1000); //模拟抢票动作
            cout << name << " get ticket:" << ticket << endl;
            ticket--;
            pthread_mutex_unlock(&lock); //互斥量解锁
        }
        else
        {
            pthread_mutex_unlock(&lock); //互斥量解锁
            break;
        }
        usleep(10); //模拟抢完票后的动作
    }
    return nullptr;
}

int main()
{
    pthread_mutex_init(&lock, nullptr); //初始化互斥锁

    pthread_t t1, t2, t3, t4;
    pthread_create(&t1, nullptr, threadRoutine, (void*)"Thread 1");
    pthread_create(&t2, nullptr, threadRoutine, (void*)"Thread 2");
    pthread_create(&t3, nullptr, threadRoutine, (void*)"Thread 3");
    pthread_create(&t4, nullptr, threadRoutine, (void*)"Thread 4");

    pthread_join(t1, nullptr);
    pthread_join(t2, nullptr);
    pthread_join(t3, nullptr);
    pthread_join(t4, nullptr);

    pthread_mutex_destroy(&lock); 
    return 0;
}

运行结果：

对于这种纯互斥环境，如果对锁的分配不够合理，就可能导致其他线程的饥饿问题

在我们的视角看来，线程就像排着队申请互斥量，解锁后的线程不能让它能够立刻申请锁，而是排到队列尾部等待。让多个线程按照一定的顺序来获取互斥量，这也是一种同步。

互斥锁要保护临界资源的安全，首先要确保自己是线程安全的。锁本身也是共享资源，因此加锁和解锁本身就必须得是原子性的操作。

并且当线程持有锁在访问临界区时如果被切换走，也必须要保证其他线程无法进入临界区。这些功能是如何做到的？

1.4 互斥量原理

互斥量加锁和解锁的过程都需要多条汇编语句，如何保证其原子性？

首先看下互斥量加锁的伪代码

lock:
    movb $0, %al
    xchgb %al, mutex
    if(al寄存器的内容 > 0)
        return 0;
    else
        线程挂起等待;
    goto lock;

我们可以把互斥量看作一个整型变量，为1时代表未加锁状态，为0时代表已经被加锁

初始互斥量的值为1，首先线程执行第一条汇编语句时把al寄存器的值变为0

第二步是重点：将互斥量mutex中的1与al寄存器中的0交换。哪个线程先执行到这条汇编语句，哪个线程才真正的对互斥量上了锁

如果某个线程被切换时寄存器al已经和互斥锁交换拿到了这个1，则会将al寄存器中的数据交换到自己的硬件上下文中带走，这个唯一的1就被该线程所私有。线程被切换走后al寄存器和锁中都没有1，这个1就像一把唯一的钥匙，再没有第二个线程能够获得这个1了。

后续就是通过判断al寄存器的值是否为1来判断线程是否竞争锁成功，如果为1则返回0代表加锁成功，不为1则加锁失败挂起等待

所以，虽然加锁的过程有多条汇编语句，但只有第二条决定一个线程能否申请到锁，因此是原子的

然后是互斥量解锁：

unlock:
    movb $1, mutex
    唤醒被挂起的线程;
    return 0;

要对一个互斥量解锁，只需要再把mutex的值变为1即可

虽然原本持有锁的线程的al寄存器值仍为1，但当该线程再次申请锁时会首先将al寄存器的值变为0，因此不需要在解锁时额外对al寄存器进行处理

1.5 重入和线程安全

关于重入和线程安全的概念：

• 线程安全：多个线程并发执行同一段代码时，不会出现数据不一致问题。多线程访问全局变量或静态变量且没有保护机制的情况就是线程不安全的情况
• 重入：当前执行流未调用完某个函数时，有其他执行流再次进入该函数就称为重入。一个函数在被重入的情况下运行结果不会出现问题就称为可重入函数，否则是不可重入函数

常见线程不安全的情况：

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

常见不可重入的情况：

• 调用了malloc/free函数
• 调用了标准I/O库函数
• 可重入函数内部使用了静态的数据结构

可重入与线程安全相关概念：

• 可重入函数是线程安全函数的一种，一个函数是可重入的，那么也是线程安全的
• 不可重入函数不能被多个线程同时调用，否则可能引发线程安全问题

二、死锁

2.1 概念

死锁通常是因为两个或两个以上的进程在执行过程中，因争夺资源而造成的一种互相等待的现象

像这样，每辆车都在等待对方先走，自己才能继续向前移动，最后导致所有的车都走不了

2.2 造成死锁的必要条件

系统中的资源分为两种：

可剥夺性资源：线程在获取资源后，该资源可以被其他进程或者系统剥夺，例如CPU或主存

不可剥夺性资源：系统将该类型的资源分配给某个进程后，不能强行的回收

只有不可剥夺性资源才会造成死锁，当系统中的不可剥夺性资源的数量无法满足多个线程的需求，线程在争夺这些资源时出现阻塞现象，造成死锁

• 互斥条件：任何时间下一个资源只能被一个执行流调用（一个车道只能走一辆车）
• 请求与保持条件：一个执行流因请求资源时阻塞等待，不会释放自己已有的资源（车被堵住后不会倒车）
• 不剥夺条件：线程不能强行剥夺其他线程已获取的资源（不能撞车）
• 循环等待条件：若干执行流之间形成一种头尾相接的循环等待资源的关系（如上图）

2.3 死锁的处理方式

• 破坏死锁的四个必要条件，例如将资源换成允许共享使用的资源
• 让线程以安全的序列推进
• 避免死锁的算法，如银行家算法
• 死锁检测算法

三、线程同步

2.1 概念

线程同步：在保证数据安全的前提下，让多个线程按照某种特定的顺序访问临界资源，从而有效避免线程的饥饿问题

多个线程互斥访问某个变量时，可能有些情况下变量的条件不满足线程的调用条件，需要等待变量符合条件后才能继续按顺序同步访问该变量。

例如超市没货了，消费者们就需要按顺序等待补货，当有货后超市再依次通知消费者

对于上述情况，就需要用到条件变量

2.2 条件变量相关API

条件变量实际上就是等待队列+通知机制，线程在等待某个资源时被依次加入等待队列中，当资源就绪时通知机制就依次唤醒队列中正在等待的线程

（1）初始化条件变量

• 静态

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设置为nullptr即可

（2）销毁条件变量

#include <pthread.h>

int pthread_cond_destroy(pthread_cond_t *cond);

（3）等待资源就绪

#include <pthread.h>

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

pthread_cond_wait函数用于将当前线程加入条件变量的等待队列中。其中cond为待加入的条件变量，mutex是互斥锁

通过pthread_cond_wait函数的参数我们可以了解到，条件变量必须和锁一起使用，为什么？

在条件变量的使用场景中，往往涉及到多线程访问临界资源的情况，因此我们需要使用互斥锁保证线程安全。

因此就可能出现这样一种情况：

当某个线程持有锁要访问共享资源时，资源未就绪条件不满足，线程需要加入条件变量中等待（但是线程还拿着锁呢！），也就是说在pthread_cond_wait函数中，我们一定需要对线程进行解锁！

这就是为什么参数中包含互斥锁

当临界资源未就绪时让线程等待，因此我们在判断是否需要等待前一定要对临界资源进行判断，而判断也是访问临界资源，要保证线程安全，一定要在加锁之后进行判断

（4）唤醒等待线程

#include <pthread.h>

int pthread_cond_broadcast(pthread_cond_t *cond);
int pthread_cond_signal(pthread_cond_t *cond);

其中，pthread_cond_signal用于唤醒一个线程，pthread_cond_broadcast用于唤醒所有线程

互斥锁和条件变量的经典应用：生产者消费者模型，在下一篇中会提到，这里我们先快速实现一个简单的生产消费模型

#include <iostream>
#include <sys/types.h>
#include <unistd.h>
#include <pthread.h>

using namespace std;

#define MAX 5 // 货物最大容量

int goods = 0;       // 货物
pthread_mutex_t lock; // 互斥锁
pthread_cond_t cond;  // 条件变量

void *consumerRoutine(void *arg) // 消费者
{
    while (true)
    {
        pthread_mutex_lock(&lock);           // 加锁
        if (goods == 0)                      // 无货时
            pthread_cond_wait(&cond, &lock); // 等待
        goods--;                             // 消费
        cout << "consuming a goods" << endl;
        pthread_cond_signal(&cond);  // 唤醒等待线程
        pthread_mutex_unlock(&lock); // 解锁
    }
    return nullptr;
}

void *producerRoutine(void *arg) // 生产者
{
    while (true)
    {
        pthread_mutex_lock(&lock);           // 加锁
        if (goods == MAX)                    // 容量已满
            pthread_cond_wait(&cond, &lock); // 等待
        goods++;                             // 生产
        cout << "produce a goods" << endl;
        pthread_cond_signal(&cond);  // 唤醒等待线程
        pthread_mutex_unlock(&lock); // 解锁
    }
    return nullptr;
}

int main()
{
    pthread_mutex_init(&lock, nullptr); // 初始化互斥锁
    pthread_cond_init(&cond, nullptr);  // 初始化条件变量

    pthread_t c, p;
    pthread_create(&c, nullptr, consumerRoutine, nullptr);
    pthread_create(&p, nullptr, producerRoutine, nullptr);

    pthread_join(c, nullptr);
    pthread_join(p, nullptr);

    pthread_mutex_destroy(&lock);
    pthread_cond_destroy(&cond);
    return 0;
}

运行结果：

完.