c++ 线程安全队列

示例

#define LOGE(...) __android_log_print(ANDROID_LOG_ERROR,"FFMPEG",__VA_ARGS__)
SafeQueue<char *> safeQueue;
void * pushData(void *){
    int i=0;
    while(1){
         i++;
        char *d= 0;//另一种方法 char *d= static_cast<char *>(malloc(4 * sizeof(char *))); sprintf(d,"第%d个",i);
         if(asprintf(&d,"第%d个",i)<0){
             return 0;//内存分配失败
         }

        LOGE("存放%s",d);
         safeQueue.push(d);
         usleep(2000*1000);
         free(d);// malloc 动态内存分配 或者是 asprintf传了指针的地址，内部动态分配了内存。 需要手动释放内存。
     } 
}
void * popData(void *){
    char * dd=0;
    while (1){
        usleep(1000*1000);
        safeQueue.pop(dd);
        if(dd!=0){
            LOGE("获取%s",dd);
        } 
    } 
}
void main(){
    pthread_t p1,p2;
    pthread_create(&p1,0,pushData,0);
    pthread_create(&p2,0,popData,0);
}

SafeQueue队列

/**
 * Created by maple on 2019-10-25 16:25
 * E-Mail Address：740917401@qq.com
 */
#ifndef DNRECORDER_SAFE_QUEUE_H
#define DNRECORDER_SAFE_QUEUE_H

#include <queue>
#include <pthread.h>


//todo 宏 开关 是否使用c++11
//#define C11
#ifdef C11
#include <thread>
#endif


using namespace std;

template<typename T>
class SafeQueue {
public:
    SafeQueue() {
#ifdef C11

#else
        pthread_mutex_init(&mutex, NULL);
        pthread_cond_init(&cond, NULL);
#endif

    }

    ~SafeQueue() {
#ifdef C11
#else
        pthread_cond_destroy(&cond);
        pthread_mutex_destroy(&mutex);
#endif

    }

    void push(T new_value) {
#ifdef C11
        //锁 和智能指针shared_ptr原理类似，自动释放
        lock_guard<mutex> lk(mt);
        if (work) {
            q.push(new_value);
            cv.notify_one();
        }
#else
        pthread_mutex_lock(&mutex);
        if (work) {
            q.push(new_value);
            pthread_cond_signal(&cond);//通知写入完成
        }
        pthread_mutex_unlock(&mutex);
#endif

    }


    int pop(T &value) {
        int ret = 0;
#ifdef C11
        //占用空间相对lock_guard 更大一点且相对更慢一点，但是配合条件必须使用它，更灵活
        unique_lock<mutex> lk(mt);
        //第二个参数 lambda表达式：false则不阻塞 往下走
        cv.wait(lk,[this]{return !work || !q.empty();});
        if (!q.empty()) {
            value = q.front();
            q.pop();
            ret = 1;
        }
#else
        pthread_mutex_lock(&mutex);
        //在多核处理器下 由于竞争可能虚假唤醒 包括jdk也说明了 所以用while循环来等待 而不是if
        while (work && q.empty()) {
            pthread_cond_wait(&cond, &mutex);//等待写入
        }
        if (!q.empty()) {
            value = q.front();//把顶部的数据拿出来一个
            q.pop();//把顶部的数据出队一个
            ret = 1;
        }
        pthread_mutex_unlock(&mutex);
#endif
        return ret;
    }

    void setWork(int work) {
#ifdef C11
        lock_guard<mutex> lk(mt);
        this->work = work;
#else
        pthread_mutex_lock(&mutex);
        this->work = work;
        pthread_cond_signal(&cond);
        pthread_mutex_unlock(&mutex);
#endif

    }

    int empty() {
        return q.empty();
    }

    int size() {
        return q.size();
    }

    void clear() {
#ifdef C11
        lock_guard<mutex> lk(mt);
        int size = q.size();
        for (int i = 0; i < size; ++i) {
            T value = q.front();
            releaseHandle(value);
            q.pop();
        }
#else
        pthread_mutex_lock(&mutex);
        int size = q.size();
        for (int i = 0; i < size; ++i) {
            T value = q.front();
            q.pop();
        }
        pthread_mutex_unlock(&mutex);
#endif

    }


private:

#ifdef C11
    mutex mt;
    condition_variable cv;
#else
    pthread_cond_t cond;
    pthread_mutex_t mutex;
#endif

    queue<T> q;
    //是否工作的标记 1 ：工作 0：不接受数据 不工作
    int work;
};


#endif //DNRECORDER_SAFE_QUEUE_H