C++ 多线程 消费者与生产者

• 伪代码
• • 代码
• 结束

在Java 中我们利用synchronized关键字来实现同步与互斥,在C++ 中我们利用 pthread_mutex_t pthread_cond_t来实现同步与互斥 pthread_mutex_t是锁,pthread_cond_t类似于信号量

SafeQueue.cpp

template
class SafQueue {
private :
    queue queueData;
    pthread_mutex_t *mutex = nullptr;
    pthread_cond_t *cond = nullptr;

public:
    SafeQueue() {
        mutex = new pthread_mutex_t;
        cond = new pthread_cond_t;
        pthread_mutex_init(mutex, 0);
        pthread_cond_init(cond, 0);
    }

    ~SafeQueue() {
        pthread_mutex_destroy(mutex);
        pthread_cond_destroy(cond);
    }

    void add(T data) {
        pthread_mutex_lock(mutex); //加锁
        queueData.push(data);  //放入东西
        pthread_cond_broadcast(cond); //唤醒消费者
        pthread_mutex_unlock(mutex);//解锁

    }

    void get(T &value) {
        pthread_mutex_lock(mutex);//加锁
        while (queueData.empty()) {
            pthread_cond_wait(cond, mutex); //队列中没有东西 阻塞自己
        }
        value = queueData.front(); //取出东西
        queueData.pop();

        pthread_mutex_unlock(mutex);//解锁
    }
};

Main

// 入队函数
void *pushQueue(void *pVoid) {
    jint number = *static_cast(pVoid);
    while (true) {
        srand(time(NULL));
        int value = rand() % 10;
        //LOGI("我是   入  队线程 %d,压入 %d ", number, value);
        safaQueue.add(value);
        sleep(1);

        if (value == 1) {
            LOGI("我是   入  队线程 %d,我已退出", number);
            break;
        }
    }
    return 0;
}

//出队函数
void *popQueue(void *pVoid) {

    jint number = *static_cast(pVoid);
    int value = 0;
    while (true) {
        safaQueue.get(value);
        LOGI("我是 出    队线程 %d,获取到 %d ", number, value);
        if (value == 1) {
            break;
        }
    }
    LOGI("我是 出    队线程 %d,我退出 ", number);

    return 0;
}

int main()
{
int number_1 = 1;

//我们定义两个生产者,三个消费者
    int number_2 = 2;
    int number_3 = 3;
    int number_4 = 4;
    int number_5 = 5;
    pthread_t *pid_1 = new pthread_t();
    pthread_t *pid_2 = new pthread_t();
    pthread_t *pid_3 = new pthread_t();
    pthread_t *pid_4 = new pthread_t();
    pthread_t *pid_5 = new pthread_t();
    pthread_create(pid_1, 0, pushQueue, &number_1);
    pthread_create(pid_2, 0, pushQueue, &number_2);
    pthread_create(pid_3, 0, popQueue, &number_3);
    pthread_create(pid_4, 0, popQueue, &number_4);
    pthread_create(pid_5, 0, popQueue, &number_5);


    pthread_join(*pid_1, 0);
    pthread_join(*pid_2, 0);
    pthread_join(*pid_3, 0);
    pthread_join(*pid_4, 0);
    pthread_join(*pid_5, 0);

    delete pid_1;
    delete pid_2;
    delete pid_3;
    delete pid_4;
    delete pid_5;


return 0;
}