多线程-生产者消费者模式（2）

淡淡的烟草味﹌ 2022-07-24 07:29 205阅读 0赞

现在用队列来实现一次这个生产者和消费者模型，程序的主要思想如下图。  
![Figure1][]

一个全局的队列，一个锁，一个信号量，CollectImages和SendImages线程访问这个队列的时候都加锁。

全部源码

//
    // Created by honkee on 5/17/16.
    //
    
    #include <iostream>
    #include <cstdio>
    #include <cstdlib>
    #include <pthread.h>
    #include <string>
    #include <cstring>
    #include <ctime>
    #include <sstream>
    #include <unistd.h>
    #include <queue>
    
    using namespace std;
    
    void* CollectImages(void*);
    void* SendImages(void*);
    void err_exit(int, string);
    
    pthread_mutex_t queueMutex = PTHREAD_MUTEX_INITIALIZER;
    pthread_cond_t queueCond = PTHREAD_COND_INITIALIZER;
    bool flag = false;
    //struct thread_var {
        
    //    string imagePath
    //};
    pthread_t* sendThread;
    pthread_t* collectThread;
    int collectThreadCnt = 1;
    int sendThreadCnt = 4;
    queue<string> q;
    
    int main(){
        int err;
    
        collectThread = new pthread_t[collectThreadCnt];
        sendThread = new pthread_t[sendThreadCnt];
    
        for(int i = 0; i < collectThreadCnt; ++i){
            err = pthread_create(&collectThread[i], NULL, CollectImages, NULL);
            if(err != 0){
                err_exit(err, "create thread error");
            }
        }
    
        for(int i = 0; i < sendThreadCnt; ++i){
            err = pthread_create(&sendThread[i], NULL, SendImages, NULL);
            if(err != 0){
                err_exit(err, "create thread error");
            }
        }
    
        while(1){
    
        }
        return 0;
    }
    
    void* CollectImages(void* arg){
        while (1){
            srand((int)time(0));
            bool cond = false;
    
            for(int threadIndex = 0; threadIndex < sendThreadCnt; ++threadIndex){
    
    
                //cout << "thread: " << threadIndex << " collect image before lock" <<endl;
                pthread_mutex_lock(&queueMutex);
                //cout << "thread: " << threadIndex << " collect image lock" <<endl;
                int r = rand() % sendThreadCnt;
                stringstream ss;
                string str;
                ss << "/home/honkee/Downloads/photo/" << r << ".jpg";
                ss >> str;
                q.push(str);
                cout << "thread: " << threadIndex << " queue length: " << q.size() << " collect finish: "
                    << str << endl <<flush;
                flag = true;
                pthread_cond_signal(&queueCond);
                //pthread_cond_boardcast(&queueCond)
                // it will wake up other thread then print "thread:  pid  waiting..."
    
                pthread_mutex_unlock(&queueMutex);
                //cout << "thread: " << threadIndex << " collect image unlock" <<endl;
                sleep(3);
            }
    
            if(cond){
                break;
            }
        }
    }
    void* SendImages(void* arg){
        pthread_t pid = pthread_self();
        while (1) {
            bool cond = false;
    
            pthread_mutex_lock(&queueMutex);
            while(flag == false){
                cout << "thread: "<< pid << " waiting..."<< endl;
                pthread_cond_wait(&queueCond, &queueMutex);
            }
            flag = false;
            string str = q.front();
            q.pop();
    
            cout << "thread: " << pid << " queue size: " << q.size() << " send finish: "
                << str << endl <<flush;
    
    
            pthread_mutex_unlock(&queueMutex);
            sleep(5);
    
            if(cond){
                break;
            }
        }
    }
    
    void err_exit(int err, string str){
        printf("%d: %s\n", err, str.c_str());
        exit(err);
    }

[Figure1]: http://7xqdjj.com1.z0.glb.clouddn.com/product_comsume2.png