多线程同步之条件变量

编程我只用CPP

2018 年 12 月 28 日

286 次浏览

暂无评论

3626字数

编程语言

　一、条件变量

条件变量是多线程的一种同步方式，它允许多个线程以无竞争的方式等待特定事件发生。无竞争的意思是，当条件满足时，条件满足这个讯号会发送给所有的监听者线程，但多个线程中只有一个能获取到特定事件。条件变量需要配合互斥量一起使用。

相关数据结构和函数：

#include <pthread.h>

// 销毁条件变量
int pthread_cond_destroy(pthread_cond_t *cond);
// 初始化条件变量
int pthread_cond_init(pthread_cond_t *restrict cond,
       const pthread_condattr_t *restrict attr);
// 初始化条件变量
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;

条件变量的数据结构为pthread_cond_t，和其他同步的数据结构一样，也提供了两种方式来初始化。一种是通过赋值的方式，一种是通过函数的方式。在使用pthread_cond_init对初始化条件变量的时候，attr参数一般为NULL。

对条件变量加锁的方式：

#include <pthread.h>

// 等待条件满足
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);

在对条件变量加锁的时候，需要传入一把已经处于加锁状态的互斥量，此时函数会把当前线程放到条件等待的列表上并对互斥量解锁。此时线程进入条件等待状态，当有信号到达时便会触发。

通知条件满足的函数：

#include <pthread.h>
int pthread_cond_signal(pthread_cond_t *cond);
int pthread_cond_broadcast(pthread_cond_t *cond);

pthread_cond_signal函数会唤醒条件等待队列上的至少一个线程，pthread_cond_broadcast会唤醒条件队列上的所有线程。

条件队列一般用于消息队列，用于统一、协调多个生产者和消费者之间的竞争关系。

二、使用示例

以下使用了三个线程作为消费者，分别从全局队列上获取消息消费：

#include <stdio.h>
#include <unistd.h>
#include <pthread.h>
#include "log.h"

#define THREAD_COUNT 3

// 消息结构
struct msg_st {
    unsigned int msg_id;
    struct msg_st *next;
};

static struct msg_st *msg_queue = NULL;
static pthread_cond_t g_cond = PTHREAD_COND_INITIALIZER;
static pthread_mutex_t g_mutex = PTHREAD_MUTEX_INITIALIZER;

// 消费线程函数
void *handle_msg(void *arg) {
    struct msg_st *msg;
    while (1) {
        pthread_mutex_lock(&g_mutex);
        while (msg_queue == NULL) {
            pthread_cond_wait(&g_cond, &g_mutex);
        }

        // 提取消息
        msg = msg_queue;
        msg_queue = msg_queue->next;
        pthread_mutex_unlock(&g_mutex);

        // 退出
        if (msg->msg_id == (unsigned int) -1) {
            info("Thread 0x%x exit!", (unsigned int) pthread_self());
            break;
        }

        info("Thread 0x%x: msg_id = %u", (unsigned int) pthread_self(), msg->msg_id);
    }
    return NULL;
}

// 生产消息函数
void create_msg(struct msg_st *msg) {
    pthread_mutex_lock(&g_mutex);
    msg->next = msg_queue;
    msg_queue = msg;
    pthread_mutex_unlock(&g_mutex);
    pthread_cond_signal(&g_cond);
}

int main() {
    int i;
    struct msg_st msg[10];
    pthread_t pid[THREAD_COUNT];

    debug("start create threads");
    // 创建3个线程作为消费者
    for (i = 0; i < 3; i++) {
        pthread_create(&pid[i], NULL, handle_msg, NULL);
    }

    debug("start create msgs");
    // 生产10个消息
    for (i = 0; i < 10; i++) {
        msg[i].msg_id = i + 1;
        msg[i].next = NULL;
        create_msg(&msg[i]);
    }

    // 休眠1秒，确保所有消息都被消费者消费完成
    sleep(1);

    // 退出所有线程
    debug("start create exit msgs");
    for (i = 0; i < THREAD_COUNT; i++) {
        msg[i].msg_id = (unsigned int) -1;
        msg[i].next = NULL;
        create_msg(&msg[i]);
    }

    // 回收所有线程
    debug("start join threads");
    for (i = 0; i < THREAD_COUNT; i++) {
        pthread_join(pid[i], NULL);
    }

    debug("program end");

    return 0;
}

运行结果，10个消息分别被3个线程打印出来了：