掌握线程安全之道：线程互斥与同步技术解析

绝活蛋炒饭

发布于 2024-12-29 08:04:18

3280

文章被收录于专栏：绝活编程学习绝活编程学习

1. 进程间线程互斥的相关背景

临界资源：多线程执行流共享的资源就叫做临界资源

临界区：每个线程内部，访问临界资源的代码，就叫做临界区

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

临界资源和临界区

进程之间如果要进行通信我们需要先创建第三方资源，让不同的进程看到同一份资源，由于这份第三方资源可以由操作系统中的不同模块提供，于是进程间通信的方式有很多种。进程间通信中的第三方资源就叫做临界资源，访问第三方资源的代码就叫做临界区。

而多线程的大部分资源都是共享的，线程之间进行通信不需要费那么大的劲去创建第三方资源。

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

using namespace std;
int nums = 5;
int tickets = 10000;

void *pthreadrun(void *args)
{
    int mytickets = 0;
   char * name= static_cast<char*>(args);
    while (tickets >0)
    {
         cout << " I am pthread- " << name<< " : "<<tickets<<endl;
        tickets--;

        mytickets++;
        //sleep(1);
    }
    cout << " My all tickets nums : " << mytickets << endl;
    return nullptr;
}

int main()
{

    pthread_t tid;
    vector<pthread_t> pthreadnums;

    

    for (int i = 0; i < nums; i++)
    {
        
        char *td = new char[64];
        
        snprintf(td, 64, "Thread-%d", i + 1);
        
        pthread_create(&tid, nullptr, pthreadrun, (void*)td);
        pthreadnums.emplace_back(tid);
        //sleep(1);
    }

    for (auto &pb : pthreadnums)
    {
        pthread_join(pb, nullptr);
    }
    return 0;
}

此时，我们对于全局变量tickets的访问

此时我们相当于实现了主线程和新线程之间的通信，其中全局变量tickets就叫做临界资源，因为它被多个执行流共享，而新线程线程中的tickets--就叫做临界区，因为这些代码对临界资源进行了访问。

互斥和原子性

在多线程情况下，如果这多个执行流都自顾自的对临界资源进行操作，那么此时就可能导致数据不一致的问题。解决该问题的方案就叫做互斥，互斥的作用就是，保证在任何时候有且只有一个执行流进入临界区对临界资源进行访问。

原子性指的是不可被分割的操作，该操作不会被任何调度机制打断，该操作只有两态，要么完成，要么未完成。

例如，上面我们模拟实现一个抢票系统，我们将记录票的剩余张数的变量定义为全局变量，主线程创建四个新线程，让这五个新线程进行抢票，当票被抢完后这五个线程自动退出。

运行结果显然不符合我们的预期，因为其中出现了余票数为负数的情况。

该代码中记录剩余票数的变量tickets就是临界资源，因为它被多个执行流同时访问，而判断tickets是否大于0、打印剩余票数以及--tickets这些代码就是临界区，因为这些代码对临界资源进行了访问。

剩余票数出现负数的原因：

if语句判断条件为真以后，代码可以并发的切换到其他线程。
usleep用于模拟漫长业务的过程，在这个漫长的业务过程中，可能有很多个线程会进入该代码段。
--ticket操作本身就不是一个原子操作。

为什么--ticket不是原子操作？

我们对一个变量进行--，我们实际需要进行以下三个步骤：

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

--操作对应的汇编代码如下：

既然--操作需要三个步骤才能完成，那么就有可能当thread1刚把tickets的值读进CPU就被切走了，也就是从CPU上剥离下来，假设此时thread1读取到的值就是1，而当thread1被切走时，寄存器中的1叫做thread1的上下文信息，因此需要被保存起来，之后thread1就被挂起了。

假设此时thread2被调度了，由于thread1只进行了--操作的第一步，因此thread2此时看到tickets的值还是，这样pthread-2也能成功进入访问临界数据。那么，进程3,4都这样呢？

然后，当线程1重新被调度，然后将下面的代码都执行完了，并且把计算结果0写进了内存。

但是，由于判断已经判断过了，此时的进程2/3/4都会把从内存中取数据，然后--，再重新加载进入内存。这样就发生了抢票抢到了负数的情况。

2 互斥量 mutex

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

要解决上述抢票系统的问题，需要做到三点：

代码必须有互斥行为：当代码进入临界区执行时，不允许其他线程进入该临界区。
如果多个线程同时要求执行临界区的代码，并且此时临界区没有线程在执行，那么只能允许一个线程进入该临界区。
如果线程不在临界区中执行，那么该线程不能阻止其他线程进入临界区。
要做到这三点，本质上就是需要一把锁，Linux上提供的这把锁叫互斥量。

3 实现互斥的接口

3.1 初始化互斥锁（mutex）

一共有动态分配和静态分配两种方式：

pthread_mutex_init 函数是 POSIX 线程（pthread）库中用于初始化互斥锁（mutex）的函数。在多线程编程中，互斥锁用于保护共享资源，确保同一时间只有一个线程可以访问该资源，从而避免数据竞争和不一致的问题。

函数原型如下：

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

参数说明：

pthread_mutex_t *mutex：指向需要初始化的互斥锁的指针。这个互斥锁在使用前必须被正确初始化。
const pthread_mutexattr_t *attr：指向互斥锁属性的指针。这个参数是可选的，可以设置为 NULL，表示使用默认的互斥锁属性。如果需要特定的属性（如递归锁、错误检测等），则需要先设置属性对象，然后将其传递给此函数。

返回值：

成功时，pthread_mutex_init 返回 0。
失败时，返回一个错误码。常见的错误码包括 EINVAL（表示参数无效，比如 mutex 指针为 NULL），ENOMEM（表示系统内存不足，无法分配互斥锁所需的资源），以及 EBUSY（在尝试重新初始化一个已经初始化的互斥锁时可能会遇到，但这种情况在标准的互斥锁上不会发生，因为标准互斥锁不允许被重新初始化）。

调用pthread_mutex_init函数初始化互斥量叫做动态分配，除此之外，我们还可以用下面这种方式初始化互斥量，该方式叫做静态分配：

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

3.2 使用互斥锁

pthread_mutex_lock 是 POSIX 线程（pthread）库中的一个函数，用于锁定一个互斥锁（mutex）。互斥锁是一种同步机制，用于防止多个线程同时访问共享资源，从而避免数据竞争和不一致的情况。

当一个线程调用 pthread_mutex_lock 并成功锁定互斥锁时，该线程可以继续执行与互斥锁保护的共享资源相关的代码。如果互斥锁已经被另一个线程锁定，那么调用 pthread_mutex_lock 的线程将被阻塞，直到互斥锁被释放（即，直到另一个线程调用 pthread_mutex_unlock）。

int pthread_mutex_lock(pthread_mutex_t *mutex);

3.3解锁互斥锁

pthread_mutex_unlock 是 POSIX 线程（pthread）库中的一个函数，用于释放一个之前被锁定的互斥锁（mutex）。当线程完成对共享资源的访问后，应该调用此函数来释放互斥锁，以便其他等待该锁的线程可以继续执行。

int pthread_mutex_lock(pthread_mutex_t *mutex);

3.4 销毁互斥量

pthread_mutex_destroy 函数是 POSIX 线程（pthread）库中用于销毁互斥锁（mutex）的函数。在多线程编程中，当一个互斥锁不再需要时，应该使用 pthread_mutex_destroy 函数来销毁它，以释放系统资源。

函数原型如下：

int pthread_mutex_destroy(pthread_mutex_t *mutex);

参数说明：

pthread_mutex_t *mutex：指向需要销毁的互斥锁的指针。这个互斥锁必须是已经初始化过的，并且当前没有被任何线程锁定。

返回值：

成功时，pthread_mutex_destroy 返回 0。
失败时，返回一个错误码。常见的错误码包括 EBUSY（表示互斥锁当前被某个线程锁定，因此无法销毁），EINVAL（表示参数无效，比如 mutex 指针为 NULL 或指向一个未初始化的互斥锁），以及 ENOSYS（表示系统不支持该函数，这种情况在现代操作系统中很少出现）。

使用注意事项：

在销毁互斥锁之前，必须确保没有任何线程在等待该锁，也没有线程持有该锁。如果尝试销毁一个被锁定或被等待的互斥锁，pthread_mutex_destroy 将返回 EBUSY 错误码。
互斥锁一旦被销毁，就不能再被使用（包括重新初始化）。如果尝试对已经销毁的互斥锁进行任何操作（如锁定、解锁或销毁），行为是未定义的。
销毁互斥锁是释放系统资源的好做法，特别是在长时间运行的应用程序或需要频繁创建和销毁互斥锁的场景中。
在多线程环境中，必须小心协调对 pthread_mutex_destroy 的调用，以确保在销毁互斥锁之前没有其他线程正在使用它。

3.5 互斥量引入抢票系统对于抢票过量问题的修正

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

using namespace std;
int nums = 5;
int tickets = 10000;

pthread_mutex_t my_mutex; // 这个就类似于之前信号部分的sigset_t

void *pthreadrun(void *args)
{
    int mytickets = 0;
    char *name = static_cast<char *>(args);

    while (1)
    {
        // pthread_mutex_init(&my_mutex,nullptr);
        pthread_mutex_lock(&my_mutex);

        if (tickets > 0)
        {

            usleep(1000);
            cout << " I am pthread- " << name << " : " << tickets << endl;
            tickets--;

            mytickets++;
            // sleep(1);
            pthread_mutex_unlock(&my_mutex);
        }
        else
        {
             pthread_mutex_unlock(&my_mutex);
            cout << " My all tickets nums : " << mytickets << endl;
            break;
        }
    }
    return nullptr;
}
int main()
{

    pthread_t tid;
    vector<pthread_t> pthreadnums;

    for (int i = 0; i < nums; i++)
    {

        char *td = new char[64];

        snprintf(td, 64, "Thread-%d", i + 1);

        pthread_create(&tid, nullptr, pthreadrun, (void *)td);
        pthreadnums.emplace_back(tid);
        // sleep(1);
    }

    for (auto &pb : pthreadnums)
    {
        pthread_join(pb, nullptr);
    }
    return 0;
}

4 对于互斥量底层实现的探究

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

此时我们就可以看一下申请锁的过程了。

一开始是先将寄存器al清空，清空完之后，是可以发生进程调度的，也就是这时候可以有很多进程可以把自己寄存器的数据情况（进程上下文对应的位置）。

然后，让寄存器与内存上的数据mutex交换数据，在交换数据的时候，由于这个操作只有一条汇编代码，所以这个操作是原子的，这个交换不会被打断，也就是此时mutex上的值一定会被传递给进程。

这样申请到了，到了mutex，那么其他的线程申请到的mutex也就是0，那只能挂起等待了，而申请到了1的mutex的线程也就可以去访问临界区了。

注意：

在申请锁时本质上就是哪一个线程先执行了交换指令，那么该线程就申请锁成功，因为此时该线程的al寄存器中的值就是1了。而交换指令就只是一条汇编指令，一个线程要么执行了交换指令，要么没有执行交换指令，所以申请锁的过程是原子的。
在线程释放锁时没有将当前线程al寄存器中的值清0，这不会造成影响，因为每次线程在申请锁时都会先将自己al寄存器中的值清0，再执行交换指令。
CPU内的寄存器不是被所有的线程共享的，每个线程都有自己的一组寄存器，但内存中的数据是各个线程共享的。申请锁实际就是，把内存中的mutex通过交换指令，原子性的交换到自己的al寄存器中。

5 线程同步问题

同步概念与竞态条件：

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

首先需要明确的是，单纯的加锁是会存在某些问题的，如果个别线程的竞争力特别强，每次都能够申请到锁，但申请到锁之后什么也不做，所以在我们看来这个线程就一直在申请锁和释放锁，这就可能导致其他线程长时间竞争不到锁，引起饥饿问题。
单纯的加锁是没有错的，它能够保证在同一时间只有一个线程进入临界区，但它没有高效的让每一个线程使用这份临界资源。
现在我们增加一个规则，当一个线程释放锁后，这个线程不能立马再次申请锁，该线程必须排到这个锁的资源等待队列的最后。
增加这个规则之后，下一个获取到锁的资源的线程就一定是在资源等待队列首部的线程，如果有十个线程，此时我们就能够让这十个线程按照某种次序进行临界资源的访问。
例如，现在有两个线程访问一块临界区，一个线程往临界区写入数据，另一个线程从临界区读取数据，但负责数据写入的线程的竞争力特别强，该线程每次都能竞争到锁，那么此时该线程就一直在执行写入操作，直到临界区被写满，此后该线程就一直在进行申请锁和释放锁。而负责数据读取的线程由于竞争力太弱，每次都申请不到锁，因此无法进行数据的读取，引入同步后该问题就能很好的解决。

5.1 条件变量

条件变量是利用线程间共享的全局变量进行同步的一种机制，条件变量是用来描述某种资源是否就绪的一种数据化描述。

条件变量主要包括两个动作：

一个线程等待条件变量的条件成立而被挂起。
另一个线程使条件成立后唤醒等待的线程。

条件变量通常需要配合互斥锁一起使用。

5.2 条件变量初始化(动静态两种方式)

pthread_cond_init 是 POSIX 线程（pthread）库中用于初始化条件变量（condition variable）的函数。条件变量是线程同步的一种机制，它允许线程在某些条件满足时被唤醒，从而继续执行。这主要用于多线程编程中，以确保线程之间的正确协作和数据一致性。

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

参数说明

pthread_cond_t *cond：指向要初始化的条件变量的指针。这是一个由 pthread 库定义的类型，用于表示条件变量。
const pthread_condattr_t *attr：一个指向条件变量属性的指针。这些属性可以用来设置条件变量的特定行为。如果设置为 NULL，则使用默认属性。条件变量属性通常用于控制条件变量的共享性（进程间共享或线程间共享）和是否使用动态分配的条件变量等。

返回值

成功时，pthread_cond_init 返回 0。
失败时，返回一个错误代码，常见的错误代码包括 EINVAL（表示传递了无效的参数，如未对齐的内存地址）、ENOMEM（表示系统内存不足，无法分配所需的资源）等。

静态初始化：全局变量（OS自动回收）

pthread_cond_t cond = PTHREAD_COND_INITIALIZER;

5.3 线程等待：pthread_cond_wait

pthread_cond_wait 是 POSIX 线程（pthread）库中用于线程同步的一个关键函数，它允许线程等待一个条件变量的触发。当线程调用 pthread_cond_wait 时，它会首先解锁（或尝试解锁）与条件变量关联的互斥锁（mutex），然后阻塞当前线程，直到另一个线程通过 pthread_cond_signal 或 pthread_cond_broadcast 唤醒它。唤醒后，线程会重新获取（或尝试获取）之前释放的互斥锁，并继续执行。

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

参数

pthread_cond_t *cond：指向要等待的条件变量的指针。
pthread_mutex_t *mutex：指向已经锁定的互斥锁的指针。这个互斥锁必须与条件变量 cond 关联，并且在调用 pthread_cond_wait 之前，调用线程必须已经持有这个互斥锁。

返回值

成功时，pthread_cond_wait 返回 0。但是，由于这个函数会导致线程阻塞，所以实际上这个返回值并不会立即被调用线程看到。当线程被唤醒并重新获取互斥锁后，它会继续执行后续的代码。
如果调用时条件变量或互斥锁无效，或者发生了其他错误（如线程被取消），pthread_cond_wait 可能会通过返回错误代码来通知调用线程。但在正常情况下，线程是被阻塞的，直到被条件变量唤醒。

5.4 线程唤醒(全部唤醒和唤醒一个)

pthread_cond_signal 是 POSIX 线程（pthread）库中用于线程同步的条件变量函数。它用于唤醒一个等待特定条件变量的线程。如果有多个线程在等待同一个条件变量，那么具体唤醒哪一个线程是由调度策略决定的，通常是不可预测的。

int pthread_cond_signal(pthread_cond_t *cond);

pthread_cond_broadcast 是 POSIX 线程（pthread）库中的一个函数，用于唤醒所有等待特定条件变量的线程。与 pthread_cond_signal 不同，pthread_cond_broadcast 会发送一个广播信号给所有等待该条件变量的线程，而不是只唤醒一个。

int pthread_cond_broadcast(pthread_cond_t *cond);

参数

pthread_cond_t *cond：指向要发送广播信号的条件变量的指针。

返回值

成功时，pthread_cond_broadcast 返回 0。
如果调用时条件变量无效，函数可能会返回错误代码。但在正常情况下，这个函数不会失败，因为它只是发送一个广播信号，而不涉及复杂的资源分配或状态检查。

5.5 销毁条件变量

pthread_cond_destroy 是 POSIX 线程（pthread）库中的一个函数，用于销毁一个条件变量。在条件变量不再需要时调用此函数可以释放与条件变量相关联的资源。

int pthread_cond_destroy(pthread_cond_t *cond);

参数

pthread_cond_t *cond：指向要销毁的条件变量的指针。

返回值

成功时，pthread_cond_destroy 返回 0。
如果调用时条件变量正在被使用（例如，有线程正在等待它），函数将返回错误代码 EBUSY。

5.6 线程同步的示例

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

using namespace std;

pthread_mutex_t gmutex = PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP;
pthread_cond_t gcond = PTHREAD_COND_INITIALIZER;

void *marstCore(void *args)
{
    char *name = static_cast<char *>(args);
    while (1)
    {   
        pthread_cond_signal(&gcond);
        //cout << "I am " << name << endl;
        sleep(1);
    }
    return nullptr;
}

void marststart(vector<pthread_t> *tidnums)
{
    pthread_t tid;
    int n = pthread_create(&tid, nullptr, marstCore, (void *)"pthread-marster");
    if (0 == n)
    {
        cout << "masrst pthread create success" << endl;
    }
    return;
}

void *slaverCore(void *args)
{
    char *name = static_cast<char *>(args);
    while (1)
    {
        //1.加锁
        pthread_mutex_lock(&gmutex);

        //2.条件变量一般都是在加锁和解锁之间运用的
        pthread_cond_wait(&gcond,&gmutex);
        cout << "I am " << name << endl;
        //3.解锁
        pthread_mutex_unlock(&gmutex);
        sleep(1);
    }
    return nullptr;
}

void startslaver(vector<pthread_t> *tidnums, int nums)
{
    for (int i = 0; i < nums; i++)
    {
        char *name = new char[64];
        snprintf(name, 64, "pthread-%d", 1 + i);
        pthread_t tid;
        int n = pthread_create(&tid, nullptr, slaverCore, name);
        if (0 == n)
        {
            cout << name << " create success" << endl;
        }

        tidnums->emplace_back(tid);
    }
    return;
}

void wait(vector<pthread_t> &tidnums, int nums)
{
    for (int i = 0; i < nums; i++)
    {
        pthread_join(tidnums[i], nullptr);
    }
    return;
}

int main()
{
    vector<pthread_t> tidnums;
    marststart(&tidnums);
    startslaver(&tidnums, 5);

    wait(tidnums, 5);
}