• Linux线程同步对象：互斥体、信号量、条件变量、读写锁

    文章目录

    1. Linux互斥体

    Linux互斥体与Windows的临界区对象用法很相似，一般也是通过限制多个线程同时执行某段代码来保护资源的。使用数据结构pthread_mutex_t表示一个互斥体对象（pthread.h中)。

    初始化方式

    1. 使用 PTHREAD_MUTEX_INITIALIZER直接给互斥体变量赋值，示例
    #include 
pthread_mutex_t mtx = PTHREAD_MUTEX_INITIALIZER;

    • 1
    • 2
    1. 若互斥体是动态分配的或者需要设置相关属性，需要使用pthread_mutex_init函数初始化：
    int pthread_mutex_init(pthread_mutex_t* restrict mutex,
					const pthread_mutexattr_t* restrict attr);

    • 1
    • 2

    参数mutex是需要初始化的mutex对象的指针
    attr是需要设置的互斥体属性，通常设置为NULL。
    如果执行成功，返回0；失败返回1个错误吗。

    使用示例：

    #include 

pthread_mutex_t mtx;
pthread_mutex_init(&mtx, NULL);

    • 1
    • 2
    • 3
    • 4

    销毁方式

    当不再需要互斥体对象时，使用pthread_mutex_destroy函数销毁它：

    int pthread_mutex_destroy(pthread_mutex_t* mutex);

    • 1

    执行成功返回0，失败返回错误码。

    注意点：

    1. 无需销毁使用PTHREAD_MUTEX_INITIALIZER初始化的互斥体。
    2. 不要销毁一个已经加锁或正在被条件变量使用的互斥体对象，当互斥体处于已加锁状态或正在和条件变量配合使用时，调用销毁函数会返回 EBUSY错误。

    加解锁操作

    一般使用三个函数:

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

    • 1
    • 2
    • 3

    他们都是执行成功返回0，失败则返回错误码。具体错误码随着互斥体对象属性类型的不同而不同。

    在设置互斥体对象属性时需要创建一个pthread_mutexattr_t类型的对象：

    int pthread_mutexattr_init(pthread_mutexattr_t* attr);
int pthread_mutexattr_destroy(pthread_mutexattr_t* attr);

    • 1
    • 2

    使用pthread_mutexattr_settype/gettype设置或获取想要的属性类型。

    int pthread_mutexattr_settype(pthread_mutexattr_t* attr, int type);
int pthread_mutexattr_gettype(const pthread_mutexattr* restrict attr, 
						int* restrict type); 

    • 1
    • 2
    • 3

    设置属性

    属性类型一般有以下这些值。

    1. PTHREAD_MUTEX_NORMAL（普通锁）

    一个线程如果对一个已经加锁的普通锁再次加锁，那么程序会阻塞在第二次调用pthread_mutex_lock代码处。测试：

    #include 
#include 
#include 
#include 

int main(void)
{
    pthread_mutex_t mymutex;
    pthread_mutexattr_t mutex_attr;
    pthread_mutexattr_init(&mutex_attr);
    pthread_mutexattr_settype(&mutex_attr, PTHREAD_MUTEX_NORMAL);
    pthread_mutex_init(&mymutex, &mutex_attr);

    int ret = pthread_mutex_lock(&mymutex);
    printf("ret = %d\n", ret);

    // 再次加锁
    ret = pthread_mutex_lock(&mymutex);
    printf("ret = %d\n", ret);

    pthread_mutex_destroy(&mymutex);
    pthread_mutexattr_destroy(&mutex_attr);

    return 0;
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25

    运行结果只打印了一行，原因是阻塞在第二个lock处。

    这种情况下，pthread_mutex_trylock()如果拿不到锁，则也不会阻塞，而是会立即返回EBUSY错误码。

    在这里插入图片描述

    在这里插入图片描述

    1. PTHRAED_MUTEX_ERRORCHECK（检错锁）

    如果是这种属性的锁，对于已经加锁的互斥体对象再次加锁，则pthread_mutex_lock会返回EDEADLK。

    示例：

    #include 
#include 
#include 
#include 

int main(void)
{
    pthread_mutex_t mymutex;
    pthread_mutexattr_t mutex_attr;
    pthread_mutexattr_init(&mutex_attr);
    pthread_mutexattr_settype(&mutex_attr, PTHREAD_MUTEX_ERRORCHECK);
    pthread_mutex_init(&mymutex, &mutex_attr);

    int ret = pthread_mutex_lock(&mymutex);
    printf("ret = %d\n", ret);

    // 再次加锁
    ret = pthread_mutex_lock(&mymutex);
    printf("ret = %d\n", ret);
    if (ret == EDEADLK)
    {
        printf("EDEADLK\n");
    }

    pthread_mutex_destroy(&mymutex);
    pthread_mutexattr_destroy(&mutex_attr);

    return 0;
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29

    在这里插入图片描述

    如果是其他线程对这个互斥体再次调用lock，则会阻塞在该函数的调用处。

    1. PTHREAD_MUTEX_RECURSIVE（可重入锁）

    该属性允许同一个线程对其持有的互斥体重复加锁，每成功调用lock一次，该互斥体对象的锁引用技术就会增加1，相反，unlock一次，引用技术减1。引用计数为0时，允许其他线程获取该锁，否则会阻塞在lock调用处。

    2.Linux信号量

    信号量代表一定的资源数量，可以根据当前资源的数量按需唤醒指定数量的资源消费者线程，资源消费者线程一旦获取信号量，就会让资源减少指定的数量，如果减少为0，则消费者线程将全部处于挂起状态；当有新的资源到来时，消费者线程将继续被唤醒。
    另外，信号量有“资源有多份，可以同时被多个线程访问”的意思。

    常用API函数：

    #include 

int sem_init(sem_t* sem, int pshared, unsigned int value);
int sem_destroy(sem_t* sem);
int sem_post(sem_t* sem);
int sem_wait(sem_t* sem);
int sem_trywait(sem_t* sem);
int sem_timedwait(sem_t* sem, const struct timespec* abs_timeout);

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8

    sem_int()第一个参数是信号量对象地址，第二个参数表示该信号量是否可以被共享，取值为0表示只能在同一个进程的多个线程共享，非0可以在多进程间共享，第三个参数表示初始状态资源数量。 函数调用成功返回0，失败返回-1。

    sem_destroy()销毁信号量；
    sem_post()用于将信号量的资源计数递增1，并解锁该信号量对象，这样因使用sem_wait()被阻塞的其他线程会被唤醒。

    如果当前资源计数为0，sem_wait()会阻塞调用线程，直到大于0时被唤醒，唤醒后资源计数-1后返回；

    sem_trywait()是非阻塞版本，如果资源计数为0，则立即返回，不阻塞调用线程，返回值是-1，错误码errno被设置成EAGAIN；

    sem_timedwait()带有等待时间的版本，第二个参数是个结构体：

    struct timespec
{
	time_t tv_sec;	// 秒
	long tv_nsec;	// 纳秒 范围[0~999999999]
};

    • 1
    • 2
    • 3
    • 4
    • 5

    信号量实现生产者消费者

    #include 
#include 
#include 
#include 
#include 
#include 

class Task
{
public:
    Task(int taskID)
    {
        this->taskID = taskID;
    }

    void doTask()
    {
        std::cout << "handle a task, taskID: " << taskID << ", threadID: " << pthread_self() << std::endl;
    }

private:
    int taskID;
};

pthread_mutex_t     mymutex;
std::list<Task*>    tasks;
sem_t               mysemaphore;

void* consumer_thread(void* param)
{
    Task* pTask = NULL;
    while (true)
    {
        if (sem_wait(&mysemaphore) != 0)
        {
            continue;
        }
        if (tasks.empty())
        {
            continue;
        }

        pthread_mutex_lock(&mymutex);
        pTask = tasks.front();
        tasks.pop_front();
        pthread_mutex_unlock(&mymutex);

        pTask->doTask();
        delete pTask;
    }

    return NULL;
}

void* producer_thread(void* param)
{
    int taskID = 0;
    Task* pTask = NULL;

    while (true)
    {
        pTask = new Task(taskID);

        pthread_mutex_lock(&mymutex);
        tasks.push_back(pTask);
        std::cout << "produce a task, taskID: " << taskID << ", threadID: " << pthread_self() << std::endl;

        pthread_mutex_unlock(&mymutex);

        // 释放信号量，通知消费者线程
        sem_post(&mysemaphore);

        taskID++;

        sleep(1);
    }

    return NULL;
}

int main(void)
{
    pthread_mutex_init(&mymutex, NULL);
    // 初始化信号量资源计数为0
    sem_init(&mysemaphore, 0, 0);

    // 创建5个消费者线程
    pthread_t consumerThreadID[5];
    for (int i = 0; i < 5; ++i)
    {
        pthread_create(&consumerThreadID[i], NULL, consumer_thread, NULL);
    }

    // 创建一个生产者线程
    pthread_t producerThreadID;
    pthread_create(&producerThreadID, NULL, producer_thread, NULL);

    pthread_join(producerThreadID, NULL);

    for (int i = 0; i < 5; ++i)
    {
        pthread_join(consumerThreadID[i], NULL);
    }

    sem_destroy(&mysemaphore);
    pthread_mutex_destroy(&mymutex);

    return 0;
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    • 80
    • 81
    • 82
    • 83
    • 84
    • 85
    • 86
    • 87
    • 88
    • 89
    • 90
    • 91
    • 92
    • 93
    • 94
    • 95
    • 96
    • 97
    • 98
    • 99
    • 100
    • 101
    • 102
    • 103
    • 104
    • 105
    • 106
    • 107
    • 108
    • 109

    输出结果：

    在这里插入图片描述

    3.条件变量

    条件变量为什么要与互斥体结合使用

    我们假设条件变量不与互斥体结合，看看效果：

    伪代码：

    // m是互斥体，cv是条件变量
pthread_mutex_lock(&m);
while (condition_is_false)
{
	pthread_mutex_unlock(&m);
	// 解锁之后，等待之前，可能条件已经满足，信号已经发出，但该信号可能被错过
	cond_wait(&cv);
	pthread_mutex_lock(&m);
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9

    假设线程A在执行完第五行代码后CPU时间片被剥夺，此时另一个线程B获得互斥体m,，然后发送条件信号，等线程A重新获得时间片后，由于该信号已经被错过，可能会导致线程A在第七行无限阻塞下去。

    造成这一现象的根源是释放互斥体对象与条件变量等待唤醒不是原子操作。

    条件变量的使用

    初始化和销毁：

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

    • 1
    • 2

    也可以这样初始化：

    pthread_cond_t cond = PTHREAD_COND_INITIALIZER;

    • 1

    等待条件变量被唤醒：

    int pthread_cond_wait(pthread_cond_t* restrict cond, pthread_mutex_t* restrict mutex);
int pthread_cond_timedwait(pthread_cond_t* restrict cond, pthread_mutex_t*
						restrict mutex, const struct timespec* restrict abstime);

    • 1
    • 2
    • 3

    如果条件变量等待的条件没有被满足，那么调用pthread_cond_wait()的线程会一直等待下去。
    timedwait()是非阻塞版本。

    因调用pthread_cond_wait()而等待的线程可以被以下API函数唤醒：

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

    • 1
    • 2

    signal表示一次唤醒一个线程，而且是随机的；
    broadcast唤醒所有调用pthread_cond_wait()等待的线程，相当于广播。。。

    条件变量实现生产者消费者

    将刚才信号量的代码稍微改一改~

    #include 
#include 
#include 
#include 
#include 
#include 

class Task
{
public:
    Task(int taskID)
    {
        this->taskID = taskID;
    }

    void doTask()
    {
        std::cout << "handle a task, taskID: " << taskID << ", threadID: " << pthread_self() << std::endl;
    }

private:
    int taskID;
};

pthread_mutex_t     mymutex;
std::list<Task*>    tasks;
pthread_cond_t      mycv;
//sem_t               mysemaphore;

void* consumer_thread(void* param)
{
    Task* pTask = NULL;
    while (true)
    {
        pthread_mutex_lock(&mymutex);
        while (tasks.empty())
        {
            // 如果获得了互斥锁，但是条件不合适，则pthread_cond_wait会释放锁，不往下执行
            // 发生变化后，如果条件合适，则pthread_cond_wait直接获得锁
            pthread_cond_wait(&mycv, &mymutex);
        }

        pTask = tasks.front();
        tasks.pop_front();

        pthread_mutex_unlock(&mymutex);

        if (pTask == NULL)
            continue;

        pTask->doTask();
        delete pTask;
        pTask = NULL;
    }

    return NULL;
}

void* producer_thread(void* param)
{
    int taskID = 0;
    Task* pTask = NULL;

    while (true)
    {
        pTask = new Task(taskID);

        pthread_mutex_lock(&mymutex);
        tasks.push_back(pTask);
        std::cout << "produce a task, taskID: " << taskID << ", threadID: " << pthread_self() << std::endl;

        pthread_mutex_unlock(&mymutex);

        // 释放信号量，通知消费者线程
        //sem_post(&mysemaphore);
        pthread_cond_signal(&mycv);

        taskID++;

        sleep(1);
    }

    return NULL;
}

int main(void)
{
    pthread_mutex_init(&mymutex, NULL);
    pthread_cond_init(&mycv, NULL);

    // 创建5个消费者线程
    pthread_t consumerThreadID[5];
    for (int i = 0; i < 5; ++i)
    {
        pthread_create(&consumerThreadID[i], NULL, consumer_thread, NULL);
    }

    // 创建一个生产者线程
    pthread_t producerThreadID;
    pthread_create(&producerThreadID, NULL, producer_thread, NULL);

    pthread_join(producerThreadID, NULL);

    for (int i = 0; i < 5; ++i)
    {
        pthread_join(consumerThreadID[i], NULL);
    }

    //sem_destroy(&mysemaphore);
    pthread_cond_destroy(&mycv);
    pthread_mutex_destroy(&mymutex);

    return 0;
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    • 80
    • 81
    • 82
    • 83
    • 84
    • 85
    • 86
    • 87
    • 88
    • 89
    • 90
    • 91
    • 92
    • 93
    • 94
    • 95
    • 96
    • 97
    • 98
    • 99
    • 100
    • 101
    • 102
    • 103
    • 104
    • 105
    • 106
    • 107
    • 108
    • 109
    • 110
    • 111
    • 112
    • 113
    • 114

    在这里插入图片描述

    条件变量最关键的一个地方，就是需要弄清楚pthread_cond_wait()在条件满足于不满足时的两种行为：

    1. pthread_cond_wait()在阻塞时，会释放器绑定的互斥体并阻塞线程。因此在调用该函数前应该对互斥体有个加锁操作。
    2. 收到条件信号时，pthread_cond_wait()会返回并对其绑定的互斥体进行加锁，因此在其下面一定由个对互斥体解锁的操作。

    条件变量的虚假唤醒

    再将互斥体和条件变量配合使用的代码中，有个while()语句，条件变量醒来之后再次判断条件是否满足：

    while (tasks.empty())
{
	pthread_cond_wait(&mycv, &mymutex);
}

    • 1
    • 2
    • 3
    • 4

    为什么不写成下面代码呢？

    if (tasks.empty()) {...}

    • 1

    原因：
    因为某次操作系统唤醒pthread_cond_wait()时tasks.empty()可能仍为true，即操作系统可能在某些情况下唤醒条件变量，也就是说存在没有其他线程向条件变量发送信号，但等待此条件变量的线程有可能醒来的情形。我们将这种行为称为 虚假唤醒（spurious wakeup）。
    因此将条件放在while循环中意味着光唤醒条件变量不行，还必须满足条件，程序才能继续执行正常逻辑。

    为什么会存在虚假唤醒？
    一个原因是pthread_cond_wait()是futex系统调用，属于阻塞型系统调用，当系统调用被信号中断时，会返回-1，并把errno错误码置为EINTR。

    4. 读写锁

    读写锁的应用场景

    在实际应用中，对共享变量的访问大多有个特点：大多数情况下，线程只是读取共享变量的值，只有在极少数情况下才会真正修改共享变量的值。对于这种情况，读请求之间无须同步，他们之间的并发访问是安全的。然而写请求必须锁住读请求和其他写请求。

    读写锁的应用方法

    读写锁在Linux系统中使用pthread_rwlock_t类型表示，初始化和销毁API：

    #include 

int pthread_rwlock_init(pthread_rwlock_t* rwlock, const pthread_rwlockattr* attr);

int pthread_rwlock_destroy(pthread_rwlock_t* rwlock);

    • 1
    • 2
    • 3
    • 4
    • 5

    rwlock参数是需要初始化和销毁的读写锁对象的地址，attr参数用于设置读写锁的属性，一般设置为NULL，表示使用默认属性。若函数调用成功返回0，失败返回非0值，通过errno错误码判断错误原因。

    如果不需要动态创建或者设置非默认属性的读写锁对象，可以使用：

    pthread_rwlock_t myrwlock = PTHREAD_RWLOCK_INITIALIZER;

    • 1
  • 相关阅读:
    在vue项目中下载swiper出现：Do not use ‘new’ for side effects报错问题
    正则表达式
    【网络奇遇记】那年我与计算机网络的初相识 —— 网络的体系结构
    网易严选,使用uni-app实现,包含后台数据文件
    python 基础语法和基本概念
    [C#]vs2022安装后C#创建winform没有.net framework4.8
    Implicit conversion from data type varbinary to datetime2 is not allowed 异常处理
    基于JAVA的图书借阅管理平台【数据库设计、源码、开题报告】
    干货整理！ERP在制造业的发展趋势如何，看这一篇就够了
    DBLink 简介
  • 原文地址：https://blog.csdn.net/m0_56257585/article/details/126152586