Linux 信号量详解与基于环形队列的生产消费模型实现

RingQueue.hpp

#include "Sem.hpp"
#include <iostream>
#include <vector>

const static int gcap = 5;

template<class T>
class RingQueue {
public:
    RingQueue(int cap = gcap) : _cap(cap), _ring_queue(cap), _space_sem(cap), _data_sem(0), _p_step(0), _c_step(0) {}

    void Enqueue(const T &in) {
        _space_sem.P(); // 进行生产
        _ring_queue[_p_step++] = in; // 维护环形队列
        _p_step %= _cap;
        _data_sem.V();
    }

    void Pop(T *out) {
        _data_sem.P(); // 进行消费
        *out = _ring_queue[_c_step++]; // 维护环形队列
        _c_step %= _cap;
        _space_sem.V();
    }

    ~RingQueue() {}

private:
    std::vector<T> _ring_queue; // 临界资源
    int _cap; // 环形队列容量
    Sem _space_sem; // 空闲空间信号量
    Sem _data_sem; // 数据信号量
    int _p_step; // 生产位置
    int _c_step; // 消费位置
};

main.cc

#include "RingQueue.hpp"
#include <pthread.h>
#include <unistd.h>

void* consumer(void* args) {
    RingQueue<int>* rq = static_cast<RingQueue<int>*>(args);
    while (true) {
        int data = 0;
        rq->Pop(&data);
        std::cout << "消费者消费了一个数据：" << data << std::endl;
        sleep(1);
    }
}

void* producer(void* args) {
    RingQueue<int>* rq = static_cast<RingQueue<int>*>(args);
    int data = 1;
    while (true) {
        rq->Enqueue(data);
        std::cout << "生产者生产了一个数据：" << data << std::endl;
        data++;
    }
}

int main() {
    RingQueue<int>* rq = new RingQueue<int>();
    pthread_t c, p;
    pthread_create(&c, nullptr, consumer, (void*)rq);
    pthread_create(&p, nullptr, producer, (void*)rq);
    pthread_join(c, nullptr);
    pthread_join(p, nullptr);
    return 0;
}

重新理解互斥锁和信号量

1. 为什么上面的单生产单消费没有加锁也能维护好互斥关系？当环形队列为空时，_data_sem 为零，消费者会在 P 操作处阻塞，只有生产者能运行，变相保证了生产者能互斥访问临界资源。
2. 为什么临界区代码中没有判断资源是否就绪就直接访问？因为申请信号量时本质就在对资源是否就绪做判断，将判断从临界区内转移到了入口处。
3. 信号量初值为 1 时即为二元信号量，本质是一把互斥锁。
4. 互斥锁本质就是认为自己只有一份资源的二元信号量。
5. 锁本质就是信号量的一种特殊情况。

多生产多消费模型具体实现

多生产多消费模型中，还需维护好生产者之间和消费者之间的互斥关系。_p_step 和 _c_step 是共享资源，需要保护。此时仅用信号量无法保护好步长，因为只要信号量不为零，同一个 step 位置就可以放进多个生产者或消费者访问，造成数据不一致。

因此需要利用互斥锁。为了兼顾效率，不让生产者和消费者串行访问，需要加两把锁，一把给生产者们使用，一把给消费者们使用。顺序上应先申请信号量再加锁，符合「锁的持有时间压缩到极致」的黄金原则。

Mutex.hpp

#pragma once
#include <pthread.h>

class Mutex {
public:
    Mutex() {
        pthread_mutex_init(&_lock, nullptr);
    }
    void Lock() {
        pthread_mutex_lock(&_lock);
    }
    void Unlock() {
        pthread_mutex_unlock(&_lock);
    }
    pthread_mutex_t* Get() {
        return &_lock;
    }
    ~Mutex() {
        pthread_mutex_destroy(&_lock);
    }
private:
    pthread_mutex_t _lock;
};

class LockGuard {
public:
    LockGuard(Mutex* mutexp) : _mutexp(mutexp) {
        _mutexp->Lock();
    }
    ~LockGuard() {
        _mutexp->Unlock();
    }
private:
    Mutex* _mutexp;
};

RingQueue.hpp

#include "Sem.hpp"
#include "Mutex.hpp"
#include <iostream>
#include <vector>

const static int gcap = 5;

template<class T>
class RingQueue {
public:
    RingQueue(int cap = gcap) : _cap(cap), _ring_queue(cap), _space_sem(cap), _data_sem(0), _p_step(0), _c_step(0) {}

    void Enqueue(const T &in) {
        _space_sem.P();
        {
            LockGuard lockguard(&_p_lock); // 进行生产
            _ring_queue[_p_step++] = in;
            _p_step %= _cap;
        }
        _data_sem.V();
    }

    void Pop(T *out) {
        _data_sem.P();
        {
            LockGuard lockguard(&_c_lock); // 进行消费
            *out = _ring_queue[_c_step++];
            _c_step %= _cap;
        }
        _space_sem.V();
    }

    ~RingQueue() {}

private:
    std::vector<T> _ring_queue;
    int _cap;
    Sem _space_sem;
    Sem _data_sem;
    int _p_step;
    int _c_step;
    Mutex _p_lock;
    Mutex _c_lock;
};

main.cc

#include "RingQueue.hpp"
#include <pthread.h>
#include <unistd.h>

void* consumer(void* args) {
    RingQueue<int>* rq = static_cast<RingQueue<int>*>(args);
    while (true) {
        int data = 0;
        rq->Pop(&data);
        std::cout << "消费者消费了一个数据：" << data << std::endl;
        sleep(1);
    }
}

void* producer(void* args) {
    RingQueue<int>* rq = static_cast<RingQueue<int>*>(args);
    int data = 1;
    while (true) {
        rq->Enqueue(data);
        std::cout << "生产者生产了一个数据：" << data << std::endl;
        data++;
    }
}

int main() {
    RingQueue<int>* rq = new RingQueue<int>();
    pthread_t c[2], p[3];
    pthread_create(p, nullptr, producer, (void*)rq);
    pthread_create(p + 1, nullptr, producer, (void*)rq);
    pthread_create(p + 2, nullptr, producer, (void*)rq);
    pthread_create(c, nullptr, consumer, (void*)rq);
    pthread_create(c + 1, nullptr, consumer, (void*)rq);
    pthread_join(c[0], nullptr);
    pthread_join(c[1], nullptr);
    pthread_join(p[0], nullptr);
    pthread_join(p[1], nullptr);
    pthread_join(p[2], nullptr);
    return 0;
}

四、一句话总结

同步与互斥的关系：互斥是保证临界资源的安全访问，同步是保证线程的执行顺序；信号量可以同时完成同步 + 互斥，互斥锁只能完成互斥。