多线程的同步与互斥

始终学不会

发布于 2023-10-17 08:25:37

1970

发布于 2023-10-17 08:25:37

文章被收录于专栏：小杰的学习本小杰的学习本

线程安全问题

首先来看一段代码，该代码是一个多线程抢票的逻辑

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

using namespace std;

//票是共享资源，搞多个线程来抢票

int tickets=1000;

void *gettickets(void * args)
{
    string username=static_cast<const char*>(args);
    //在这里抢票，逻辑是先判断是否有票，有票就直接开抢
    while(true)
    {
        if(tickets>0)
        {
            usleep(1234);//让线程休眠一段时间来模拟抢票消耗的时间
            cout<<username<<"正在抢票，当前票数："<<tickets<<endl;

            tickets--;

        }
        else
        {
            break;//没有余票，直接结束
        }
    }
}

int main()
{
    //创建多个线程来运行抢票逻辑
    pthread_t t1,t2,t3,t4;
    pthread_create(&t1,nullptr,gettickets,(void*)"thread 1");
    pthread_create(&t2,nullptr,gettickets,(void*)"thread 2");
    pthread_create(&t3,nullptr,gettickets,(void*)"thread 3");
    pthread_create(&t4,nullptr,gettickets,(void*)"thread 4");

    //线程执行完毕还要回收
    pthread_join(t1,nullptr);
    pthread_join(t2,nullptr);
    pthread_join(t3,nullptr);
    pthread_join(t4,nullptr);

    return 0;
}

发现线程的抢票行为使tickets变为负数了，我们明明做了判断，票数大于零才进入抢票逻辑，居然还会出现负数；引发这个问题的主要原因是数据的修改并非是原子性的，修改一个数据需要三条汇编指令：1.将数据从内存中加载到寄存器 2.在寄存器中让CPU进行算术或逻辑运算 3.将修改过的数据写回到内存中；如果在第三步之前，CPU将这个线程给切换了，那么就可能导致：明明这个数据已经被修改了一次，但还未来的及写回就被切换到下一个线程，此时新来的线程获取到的就是旧的未被修改的数据；

等到线程1再度被唤醒时，它需要完成之前未完成的动作，它会将未来的及写回的数据再次写回，此时内存中的票数又变成了999

. 从上述的情况可以得到一个结论：多线程在访问共享资源的时候是不安全的，这主要是因为多线程之间的并发执行的且访问资源的动作是非原子性的（单纯的++或者–都不是原子的）

为了解决这个问题，就提出了互斥锁；互斥锁可以让多个线程串行的访问资源（即有一个线程在访问资源时，其他线程只能等待），它也可以使得访问资源的动作变成原子性的；

在介绍锁之前补充一些概念：

原子性：要么不做，要么做完，它不会被调度机制打断，简单的理解就是：它的汇编指令只有一条临界资源：被共享的资源都可以叫做临界资源临界区：访问临界资源的代码段就是临界区互斥：任何时刻，互斥保证有且只有一个执行流进入临界区，访问临界资源，通常对临界资源起保护作用。

多线程互斥

互斥量mutex

大部分情况，线程使用的数据都是局部变量，变量的地址空间在线程栈空间内，这种情况，变量归属单个线程，其他线程无法获得这种变量。

但有时候，很多变量都需要在线程间共享，这样的变量称为共享变量，可以通过数据的共享，完成线程之间的交互。

多个线程并发的操作共享变量，会带来一些问题，这在上述的线程安全问题上已经体现了

要解决多线程并发访问临界资源带来的问题，需要做到三点：

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

其实就是加一把互斥锁，这个锁就是mutex，一个线程在持有锁的期间，其他的线程只能挂起等待；

下面介绍其常用的接口（因为接口属于pthread库，所以makefile中仍然需要包含该库）：

#include <pthread.h>//头文件

// 初始化
int pthread_mutex_init(pthread_mutex_t *restrict mutex,const pthread_mutexattr_t *restrict attr);
// 销毁
int pthread_mutex_destroy(pthread_mutex_t *mutex);

// 作为全局变量时的初始化方式，此时的锁不需要使用init初始化也不必用destory销毁
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

//加锁，如果此时没有锁则阻塞等待，直到获取到锁
int pthread_mutex_lock(pthread_mutex_t *mutex);

//如果加锁成功，直接持有锁，加锁不成功，此时立马出错返回（试着加锁，非阻塞获取方式）
int pthread_mutex_trylock(pthread_mutex_t *mutex);

//解锁
int pthread_mutex_unlock(pthread_mutex_t *mutex);

//上述所有的接口都是成功返回0，失败返回错误码

互斥锁的使用

将锁设置为全局变量，在临界区的最开始加锁，出临界区之间要记得解锁（否则其他线程就只能一直处于阻塞等待锁的过程）

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

using namespace std;

//票是共享资源，搞多个线程来抢票
pthread_mutex_t lock=PTHREAD_MUTEX_INITIALIZER;

int tickets=1000;

void *gettickets(void * args)
{
    string username=static_cast<const char*>(args);
    //在这里抢票，逻辑是先判断是否有票，有票就直接开抢
    while(true)
    {
        pthread_mutex_lock(&lock);
        if(tickets>0)
        {
            usleep(1234);//让线程休眠一段时间来模拟抢票消耗的时间
            cout<<username<<"正在抢票，当前票数："<<tickets<<endl;

            tickets--;
            //出了临界区需要解锁，否则其他线程无法使用
            pthread_mutex_unlock(&lock);
        }
        else
        {
            pthread_mutex_unlock(&lock);
            break;//没有余票，直接结束
        }
    }
}

int main()
{
    //创建多个线程来运行抢票逻辑
    pthread_t t1,t2,t3,t4;
    pthread_create(&t1,nullptr,gettickets,(void*)"thread 1");
    pthread_create(&t2,nullptr,gettickets,(void*)"thread 2");
    pthread_create(&t3,nullptr,gettickets,(void*)"thread 3");
    pthread_create(&t4,nullptr,gettickets,(void*)"thread 4");

    //线程执行完毕还要回收
    pthread_join(t1,nullptr);
    pthread_join(t2,nullptr);
    pthread_join(t3,nullptr);
    pthread_join(t4,nullptr);

    return 0;
}

此时就不会出现抢票抢到负数的情况了

当然也可以使用局部锁，为了让多个线程看到同一把锁，我们可以创建一个结构体，将这个结构体传给线程

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

using namespace std;

//当成结构体来用，里面存放的是线程名称和锁
class BuyTicket
{
public:
    BuyTicket(const string &threadname,pthread_mutex_t *mutex_p)//一般来说传参输入型是：const&，输出型是*，输入输出型是&
    :threadname_(threadname)
    ,mutex_p_(mutex_p)
    {}

public:
    string threadname_;
    pthread_mutex_t*mutex_p_;

};


int tickets=1000;

void *gettickets(void * args)
{
    BuyTicket*td=static_cast<BuyTicket*>(args);
    while(true)
    {
        pthread_mutex_lock(td->mutex_p_);
        if(tickets>0)
        {
            usleep(1234);//让线程休眠一段时间来模拟抢票消耗的时间
            cout<<td->threadname_<<"正在抢票，当前票数："<<tickets<<endl;

            tickets--;
            //出了临界区需要解锁，否则其他线程无法使用
            pthread_mutex_unlock(td->mutex_p_);
        }
        else
        {
            pthread_mutex_unlock(td->mutex_p_);
            break;//没有余票，直接结束
        }
    }
}

int main()
{

    //创建局部锁并初始化
    pthread_mutex_t lock;
    pthread_mutex_init(&lock,nullptr);

    //创建数组,表示线程ID
    vector<pthread_t> tids(4);

    //创建四个线程,并将结构体传给线程，所以要先初始化结构体
    for(int i=0;i<4;i++)
    {
        char buffer[64];
        snprintf(buffer,64,"thread %d",i+1);
        BuyTicket*td=new BuyTicket(buffer,&lock);
        pthread_create(&tids[i],nullptr,gettickets,td);//td是传给gettickets的实参
    }
    
    //回收线程
    for(const auto &tid:tids)
    {
        pthread_join(tid,nullptr);
    }

    //用完锁以后要将锁销毁
    pthread_mutex_destroy(&lock);
    return 0;
}

这种写法相比上一种要更麻烦一些，在这里我想对pthread_create函数再做一些讲解

pthread_create(pthread_t *thread, const pthread_attr_t *attr,
               void *(*start_routine) (void *), void *arg)
（1）thread：事先创建好的pthread_t类型的参数。成功时thread指向的内存单元被设置为新创建线程的线程ID。

（2）attr：用于定制各种不同的线程属性,通常直接设为NULL。

（3）start_routine：新创建线程从此函数开始运行。无参数是arg设为NULL即可。

（4）arg：是传给start_routine函数的实参，如果参数大于一个就需要用结构体来传参

首先加锁给我们的直观现象就是程序的运行速度变慢了，这是因为多线程从并发运行变成了串行，且还要加解锁；此外我们发现以上的两种写法，程序在运行时都只有一个线程在抢票，这是因为锁只规定需要互斥访问，谁持有锁谁就占有该资源；解决这个问题的办法也很简单，只需要让该线程陷入休眠即可，在现实中我们抢完票还需要付款，付款的时候线程已经退出临界区了，这里用休眠来代替：

理解锁

为了保证让多个线程串行的访问临界资源，所以必须多个线程之间只能有一把锁，并且这把锁要对所有线程都可见；也就是说锁也是一种共享资源，那么谁又来保护锁呢？

pthread_mutex_lock，pthread_mutex_unlock加锁和解锁的过程必须是安全的，且加锁的过程是原子性的。谁持有锁，谁就能进入临界区，如果某个线程申请锁，但是此时并没有锁，该线程就会阻塞式等待的加锁，所以说使用pthread_mutex_lock加锁是原子性的

在接口介绍时有一个trylock接口，该接口就是非阻塞式申请锁

线程申请到锁，就可以继续往下执行；此时其他没有申请到锁的线程就要阻塞等待，直到它们申请到锁；

一个线程在加锁期间，如果时间片到了也是可以被CPU切换的，绝对可以！但持有锁的线程在被切换的时候是抱着锁走的，其他线程仍旧无法申请到锁，所以对于其他线程而言只有两种状态：1.加锁前 2.释放锁后；站在其他线程的角度来看，持有锁的过程是原子的

我们在使用锁的时候，要尽量保证临界区的粒度要小（代码量小）；加锁是程序员行为，如果要对公共资源加锁那么每个线程都要加锁

加锁如何做到原子性

为了实现互斥锁操作,大多数体系结构都提供了swap或exchange指令，该指令的作用是把寄存器和内存单元的数据相交换,由于只有一条指令,保证了原子性。即使是多处理器平台,访问内存的总线周期也有先后，一个处理器上的交换指令执行时另一个处理器的交换指令只能等待总线周期。下面来看一下lock和unlock的伪代码：

%al代表了一个寄存器，xchgb就是exchange指令，用于数据交换；mov可以理解为赋值加锁过程：

. 寄存器中的数据属于线程的上下文，在线程切换时是要呗带走的，所以线程被切换的时候是带着线程走的解锁的过程就是将mutex中的数据重新置为1，所以一个线程加锁，另一个线程是可以将其解锁的，只是我们的代码不会这样写；

对mutex做封装

为了使用方便，可以对mutex做封装

//Mutex.hpp

#pragma once
#include<iostream>
#include<pthread.h>

//对锁做简单的封装，搞两个类，一个类是Mutex，另一个是加锁的类
class Mutex
{
public:
    Mutex(pthread_mutex_t*mutex_p=nullptr):mutex_p_(mutex_p)
    {}

    //加锁解锁
    void lock()
    {
        if(mutex_p_)pthread_mutex_lock(mutex_p_);
    }

    void unlock()
    {
        if(mutex_p_)pthread_mutex_unlock(mutex_p_);
    }

    ~Mutex()
    {}
private:
    pthread_mutex_t*mutex_p_;
};

//构造一个锁类，该类的构造是加锁，析构就是解锁
class LockGuard
{
public:
    LockGuard(pthread_mutex_t *mutex):mutex_(mutex)
    {
        mutex_.lock();
    }

    ~LockGuard()
    {
        mutex_.unlock();
    }

private:
    Mutex mutex_;
};

此时抢票的代码可以修改成以下的模样，只需要将锁作为参数传给类用以构造即可，不必再手动调用接口，且解锁过程就不需要我们显示的去调用；

可重入与线程安全

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

如果函数可重入，那么线程一定安全；线程安全，函数不一定可重入

常见的线程安全的情况

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

常见不可重入的情况

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

死锁

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

已经持有锁的线程再去申请锁也是一种死锁，死锁产生有四个必要条件：

1.互斥：一个共享资源每次被一个执行流使用 2.请求与保持：一个执行流因请求资源而阻塞，对已有资源保持不放 3.不剥夺：一个执行流获得的资源在未使用完之前，不能强行剥夺 4.环路等待条件：执行流间形成环路问题，循环等待资源

📕为什么会有死锁？

首先肯定是因为我们使用了锁->使用锁是为了保护线程安全->因为多线程在访问共享资源时有数据不一致问题->多线程的大部分资源是共享的->在解决问题的时候又带来了新的问题：死锁

如何解决死锁？

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

检测死锁的方法：1.银行家算法 2.死锁检测算法

线程同步

假设学校有一个条件极好的VIP自习室，这个自习室一次只能一个人使用并且规定是来的最早的人使用；我为了体验这个自习室，凌晨三点的时候我就奔向了自习室，当我在里面呆到七点多的时候我想去上个厕所，为了防止在我上厕所期间别人占用该自习室，我将自习室的门反锁并且带走了钥匙；又在自习室里待了几个小时候，我觉得待不住了，我准备离开，我刚将钥匙挂回去，我突然觉得好不容易占到这个自习室不能就这样离去，于是我又拿到钥匙（因为我离钥匙最近），开门以后待了没一分钟我又出来了，刚把钥匙挂好，我又觉得不能就这样算了；于是我一直重复着开门关门拿钥匙放钥匙的动作；虽然没有违反规定，但导致其他同学一直无法使用该自习室，学校的目的无法达到

在上述的例子中，因为我始终离钥匙最近，竞争力最强，所以始终是我获取到资源，且我重复获取资源的过程并没有违反任何规定；但这样并没有使学校达到提升大家学习的目的，也就是说我一直占着资源做着无意义的动作，虽然不违反规定，但是造成了其他线程的饥饿问题；为了解决这个问题就提出了线程同步：

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

条件变量

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

条件变量函数接口

#include <pthread.h>//与互斥锁有些类似

//初始化
int pthread_cond_init(pthread_cond_t *restrict cond,const pthread_condattr_t *restrict attr);
//cond是要初始化的条件

pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
//销毁
int pthread_cond_destroy(pthread_cond_t *cond);

//特定时间阻塞等待
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);

// 唤醒一批线程
int pthread_cond_broadcast(pthread_cond_t *cond);

// 唤醒一个线程
int pthread_cond_signal(pthread_cond_t *cond);

理解条件变量

校招时，公司都会在大学城附近的酒店包一层楼用于面试，来了很多面试的人，但hr是少量的，为了规范来面试的人，于是公司的管理层就规定，没有面试的人需要在某个地方排队等待

排队等待的地方就是条件变量，来面试的人就是线程；当条件不满足的时候，线程必须要到定义好的条件变量上去等，条件变量包含一个等待队列，当线程不满足条件时，就链接在这个等待队列上进行等待，当条件满足了，再去等待队列上唤醒

条件变量的使用

一次唤醒一个线程：

int tickets=1000;
pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
void *gettickets(void *args)
{
    string username = static_cast<const char *>(args);
    // 在这里抢票，逻辑是先判断是否有票，有票就直接开抢

    // BuyTicket*td=static_cast<BuyTicket*>(args);
    while (true)
    {
        pthread_mutex_lock(&lock);
        pthread_cond_wait(&cond,&lock);
        {
            if (tickets > 0)
            {
                usleep(1234); // 让线程休眠一段时间来模拟抢票消耗的时间
                cout << username << "正在抢票，当前票数：" << tickets << endl;
           		tickets--;  
                pthread_mutex_unlock(&lock);
            }
            else
            {
                pthread_mutex_unlock(&lock);
                break; // 没有余票，直接结束
            }
        }
        usleep(1234); // 用休眠来代替抢到票后的其他行为
    }
}

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

    while(true)
    {
        sleep(1);
        pthread_cond_signal(&cond); //一次唤醒一个线程
        cout<<"main thread wake up one...."<<endl;

    }
    // 线程执行完毕还要回收
    pthread_join(t1, nullptr);
    pthread_join(t2, nullptr);
    pthread_join(t3, nullptr);
    pthread_join(t4, nullptr);

    return 0;
}

此外也可一次性唤醒多个线程：

while(true)
    {
        sleep(1);
         pthread_cond_broadcast(&cond);//只要使用这个接口即可，别的代码都不必修改
        cout<<"main thread wake up one...."<<endl;

    }