【Linux系统编程】（四十四）线程同步下篇：条件变量深度解析与 POSIX 信号量实战

// 错误示范：无互斥量的条件等待
pthread_mutex_lock(&mutex);
while (queue.empty()) { // 判断条件：队列为空
    pthread_mutex_unlock(&mutex); // 解锁后等待
    pthread_cond_wait(&cond, NULL); // 假设可以不传互斥量
    pthread_mutex_lock(&mutex); // 被唤醒后重新加锁
}
// 处理队列数据
pthread_mutex_unlock(&mutex);

// 错误：if判断，存在伪唤醒风险
pthread_mutex_lock(&mutex);
if (queue.empty()) { // 单次判断
    pthread_cond_wait(&cond, &mutex);
}
// 处理数据
pthread_mutex_unlock(&mutex);

// 正确：while循环，重新判断条件
pthread_mutex_lock(&mutex);
while (queue.empty()) { // 循环判断
    pthread_cond_wait(&cond, &mutex);
}
// 处理数据
pthread_mutex_unlock(&mutex);

// 等待条件的标准模板
pthread_mutex_lock(&mutex); // 1. 加锁保护共享资源
while (条件为假) { // 2. 循环判断条件，处理伪唤醒
    pthread_cond_wait(&cond, &mutex); // 3. 原子解锁+等待
}
// 4. 条件为真，操作临界资源
// ... 处理共享资源的代码 ...
pthread_mutex_unlock(&mutex); // 5. 解锁

// 发送信号的标准模板
pthread_mutex_lock(&mutex); // 1. 加锁保护共享资源
// 2. 修改共享资源，使等待条件变为真
// ... 修改共享资源的代码 ...
pthread_cond_signal(&cond); // 3. 发送信号，唤醒等待线程
// 也可以用pthread_cond_broadcast(&cond)唤醒所有等待线程
pthread_mutex_unlock(&mutex); // 4. 解锁

// Lock.hpp 互斥量封装
#pragma once
#include <iostream>
#include <pthread.h>

namespace LockModule
{
    // 互斥量基础类
    class Mutex
    {
    public:
        // 禁用拷贝和赋值
        Mutex(const Mutex &) = delete;
        const Mutex &operator=(const Mutex &) = delete;

        // 构造函数：初始化互斥量
        Mutex()
        {
            int ret = pthread_mutex_init(&_mutex, nullptr);
            if (ret != 0)
            {
                std::cerr << "pthread_mutex_init error!" << std::endl;
                exit(EXIT_FAILURE);
            }
        }

        // 加锁
        void Lock()
        {
            int ret = pthread_mutex_lock(&_mutex);
            if (ret != 0)
            {
                std::cerr << "pthread_mutex_lock error!" << std::endl;
                exit(EXIT_FAILURE);
            }
        }

        // 解锁
        void Unlock()
        {
            int ret = pthread_mutex_unlock(&_mutex);
            if (ret != 0)
            {
                std::cerr << "pthread_mutex_unlock error!" << std::endl;
                exit(EXIT_FAILURE);
            }
        }

        // 获取原生互斥量指针，供条件变量使用
        pthread_mutex_t *GetMutexOriginal()
        {
            return &_mutex;
        }

        // 析构函数：销毁互斥量
        ~Mutex()
        {
            pthread_mutex_destroy(&_mutex);
        }

    private:
        pthread_mutex_t _mutex; // 原生POSIX互斥量
    };

    // RAII风格的锁守卫：自动加锁解锁
    class LockGuard
    {
    public:
        // 构造函数：加锁
        explicit LockGuard(Mutex &mutex) : _mutex(mutex)
        {
            _mutex.Lock();
        }

        // 析构函数：解锁
        ~LockGuard()
        {
            _mutex.Unlock();
        }

        // 禁用拷贝和赋值
        LockGuard(const LockGuard &) = delete;
        const LockGuard &operator=(const LockGuard &) = delete;

    private:
        Mutex &_mutex; // 引用互斥量，避免拷贝
    };
}

// Cond.hpp 条件变量封装
#pragma once
#include <iostream>
#include <pthread.h>
#include "Lock.hpp" // 引入封装的互斥量类

namespace CondModule
{
    // 引入互斥量的命名空间
    using namespace LockModule;

    // 条件变量类
    class Cond
    {
    public:
        // 禁用拷贝和赋值
        Cond(const Cond &) = delete;
        const Cond &operator=(const Cond &) = delete;

        // 构造函数：初始化条件变量
        Cond()
        {
            int ret = pthread_cond_init(&_cond, nullptr);
            if (ret != 0)
            {
                std::cerr << "pthread_cond_init error!" << std::endl;
                exit(EXIT_FAILURE);
            }
        }

        // 等待接口：绑定互斥量，原子解锁+等待
        void Wait(Mutex &mutex)
        {
            // 传入原生互斥量指针
            int ret = pthread_cond_wait(&_cond, mutex.GetMutexOriginal());
            if (ret != 0)
            {
                std::cerr << "pthread_cond_wait error!" << std::endl;
                exit(EXIT_FAILURE);
            }
        }

        // 唤醒一个等待线程
        void Notify()
        {
            int ret = pthread_cond_signal(&_cond);
            if (ret != 0)
            {
                std::cerr << "pthread_cond_signal error!" << std::endl;
                exit(EXIT_FAILURE);
            }
        }

        // 唤醒所有等待线程
        void NotifyAll()
        {
            int ret = pthread_cond_broadcast(&_cond);
            if (ret != 0)
            {
                std::cerr << "pthread_cond_broadcast error!" << std::endl;
                exit(EXIT_FAILURE);
            }
        }

        // 析构函数：销毁条件变量
        ~Cond()
        {
            pthread_cond_destroy(&_cond);
        }

    private:
        pthread_cond_t _cond; // 原生POSIX条件变量
    };
}

// cond_test.cpp 封装后的条件变量使用示例
#include <iostream>
#include <unistd.h>
#include <pthread.h>
#include "Lock.hpp"
#include "Cond.hpp"

// 命名空间
using namespace LockModule;
using namespace CondModule;

// 全局的互斥量和条件变量
Mutex g_mutex;
Cond g_cond;

// 线程函数：等待条件变量通知
void *thread_func(void *arg)
{
    std::string thread_name = (char *)arg;
    while (true)
    {
        LockGuard lock(g_mutex); // RAII自动加锁
        // 循环等待条件（这里简化为永久等待，实际开发中替换为业务条件）
        g_cond.Wait(g_mutex);
        // 被唤醒后执行业务逻辑
        std::cout << thread_name << " 被唤醒，执行业务逻辑..." << std::endl;
    }
    return nullptr;
}

int main()
{
    // 创建两个线程
    pthread_t t1, t2;
    pthread_create(&t1, nullptr, thread_func, (void *)"Thread-1");
    pthread_create(&t2, nullptr, thread_func, (void *)"Thread-2");

    // 主线程每隔1秒发送一次通知
    int cnt = 5;
    while (cnt--)
    {
        sleep(1);
        LockGuard lock(g_mutex); // 加锁保护
        std::cout << "主线程发送通知，剩余次数：" << cnt << std::endl;
        g_cond.NotifyAll(); // 唤醒所有等待线程
    }

    // 等待线程退出
    pthread_join(t1, nullptr);
    pthread_join(t2, nullptr);
    return 0;
}

# 编译：链接pthread库
g++ cond_test.cpp -o cond_test -lpthread
# 运行
./cond_test

主线程发送通知，剩余次数：4
Thread-1 被唤醒，执行业务逻辑...
Thread-2 被唤醒，执行业务逻辑...
主线程发送通知，剩余次数：3
Thread-1 被唤醒，执行业务逻辑...
Thread-2 被唤醒，执行业务逻辑...
主线程发送通知，剩余次数：2
Thread-1 被唤醒，执行业务逻辑...
Thread-2 被唤醒，执行业务逻辑...
主线程发送通知，剩余次数：1
Thread-1 被唤醒，执行业务逻辑...
Thread-2 被唤醒，执行业务逻辑...
主线程发送通知，剩余次数：0
Thread-1 被唤醒，执行业务逻辑...
Thread-2 被唤醒，执行业务逻辑...

int sem_init(sem_t *sem, int pshared, unsigned int value);

int sem_destroy(sem_t *sem);

int sem_wait(sem_t *sem);

int sem_post(sem_t *sem);

// Sem.hpp POSIX信号量封装
#pragma once
#include <iostream>
#include <semaphore.h>
#include <cstdlib>

class Sem
{
public:
    // 禁用拷贝和赋值
    Sem(const Sem &) = delete;
    const Sem &operator=(const Sem &) = delete;

    // 构造函数：初始化信号量，指定初始计数器值
    explicit Sem(unsigned int value = 0)
    {
        int ret = sem_init(&_sem, 0, value);
        if (ret != 0)
        {
            std::cerr << "sem_init error!" << std::endl;
            exit(EXIT_FAILURE);
        }
    }

    // P操作：等待，计数器减1
    void P()
    {
        int ret = sem_wait(&_sem);
        if (ret != 0)
        {
            std::cerr << "sem_wait error!" << std::endl;
            exit(EXIT_FAILURE);
        }
    }

    // V操作：发布，计数器加1
    void V()
    {
        int ret = sem_post(&_sem);
        if (ret != 0)
        {
            std::cerr << "sem_post error!" << std::endl;
            exit(EXIT_FAILURE);
        }
    }

    // 析构函数：销毁信号量
    ~Sem()
    {
        sem_destroy(&_sem);
    }

private:
    sem_t _sem; // 原生POSIX信号量
};

// RingQueue.hpp 基于信号量的环形队列
#pragma once
#include <iostream>
#include <vector>
#include <pthread.h>
#include "Sem.hpp"

// 模板类，支持任意类型的任务
template <typename T>
class RingQueue
{
public:
    // 构造函数：初始化环形队列，指定容量
    explicit RingQueue(int cap) : _cap(cap),
                                  _ring_queue(cap),
                                  _room_sem(cap),   // 空间信号量初始值为容量
                                  _data_sem(0),     // 数据信号量初始值为0
                                  _productor_step(0),
                                  _consumer_step(0)
    {
        // 初始化互斥量
        pthread_mutex_init(&_productor_mutex, nullptr);
        pthread_mutex_init(&_consumer_mutex, nullptr);
    }

    // 入队操作：生产者调用
    void Enqueue(const T &in)
    {
        // 1. P操作获取空闲空间，无空间则阻塞
        _room_sem.P();
        // 2. 生产者互斥，保证入队操作原子性
        pthread_mutex_lock(&_productor_mutex);
        // 3. 入队：下标取模实现环形
        _ring_queue[_productor_step++] = in;
        _productor_step %= _cap;
        // 4. 解锁
        pthread_mutex_unlock(&_productor_mutex);
        // 5. V操作发布数据，通知消费者
        _data_sem.V();
    }

    // 出队操作：消费者调用
    void Pop(T *out)
    {
        // 1. P操作获取有效数据，无数据则阻塞
        _data_sem.P();
        // 2. 消费者互斥，保证出队操作原子性
        pthread_mutex_lock(&_consumer_mutex);
        // 3. 出队：下标取模实现环形
        *out = _ring_queue[_consumer_step++];
        _consumer_step %= _cap;
        // 4. 解锁
        pthread_mutex_unlock(&_consumer_mutex);
        // 5. V操作发布空间，通知生产者
        _room_sem.V();
    }

    // 析构函数：销毁互斥量
    ~RingQueue()
    {
        pthread_mutex_destroy(&_productor_mutex);
        pthread_mutex_destroy(&_consumer_mutex);
    }

    // 禁用拷贝和赋值
    RingQueue(const RingQueue &) = delete;
    const RingQueue &operator=(const RingQueue &) = delete;

private:
    std::vector<T> _ring_queue; // 环形队列的底层容器
    int _cap;                   // 环形队列的容量
    Sem _room_sem;              // 空间信号量：生产者关注
    Sem _data_sem;              // 数据信号量：消费者关注
    int _productor_step;        // 生产者入队下标
    int _consumer_step;         // 消费者出队下标
    pthread_mutex_t _productor_mutex; // 生产者互斥锁
    pthread_mutex_t _consumer_mutex;  // 消费者互斥锁
};

// ringqueue_test.cpp 环形队列的生产者消费者测试
#include <iostream>
#include <unistd.h>
#include <pthread.h>
#include <random>
#include "RingQueue.hpp"

// 定义环形队列容量
const int QUEUE_CAP = 5;
// 全局环形队列对象，存储int类型数据
RingQueue<int> g_rq(QUEUE_CAP);

// 生产者线程函数：不断生产随机数入队
void *productor_func(void *arg)
{
    std::string productor_name = (char *)arg;
    // 随机数生成器
    std::random_device rd;
    std::mt19937 gen(rd());
    std::uniform_int_distribution<> dis(1, 100);

    while (true)
    {
        int data = dis(gen);
        g_rq.Enqueue(data);
        std::cout << productor_name << " 生产数据：" << data << std::endl;
        // 随机休眠，模拟生产耗时
        usleep(dis(gen) * 1000);
    }
    return nullptr;
}

// 消费者线程函数：不断出队并打印数据
void *consumer_func(void *arg)
{
    std::string consumer_name = (char *)arg;
    // 随机数生成器
    std::random_device rd;
    std::mt19937 gen(rd());
    std::uniform_int_distribution<> dis(1, 100);

    while (true)
    {
        int data;
        g_rq.Pop(&data);
        std::cout << consumer_name << " 消费数据：" << data << std::endl;
        // 随机休眠，模拟消费耗时
        usleep(dis(gen) * 1000);
    }
    return nullptr;
}

int main()
{
    // 创建2个生产者线程，3个消费者线程
    pthread_t p1, p2, c1, c2, c3;
    pthread_create(&p1, nullptr, productor_func, (void *)"Productor-1");
    pthread_create(&p2, nullptr, productor_func, (void *)"Productor-2");
    pthread_create(&c1, nullptr, consumer_func, (void *)"Consumer-1");
    pthread_create(&c2, nullptr, consumer_func, (void *)"Consumer-2");
    pthread_create(&c3, nullptr, consumer_func, (void *)"Consumer-3");

    // 等待线程退出（实际会一直运行）
    pthread_join(p1, nullptr);
    pthread_join(p2, nullptr);
    pthread_join(c1, nullptr);
    pthread_join(c2, nullptr);
    pthread_join(c3, nullptr);

    return 0;
}

g++ ringqueue_test.cpp -o ringqueue_test -lpthread -std=c++11

./ringqueue_test

Productor-1 生产数据：45
Productor-2 生产数据：78
Consumer-1 消费数据：45
Consumer-2 消费数据：78
Productor-1 生产数据：23
Productor-2 生产数据：91
Consumer-3 消费数据：23
Consumer-1 消费数据：91
Productor-1 生产数据：56
Productor-2 生产数据：88
Consumer-2 消费数据：56
Consumer-3 消费数据：88