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

青旅半醒 2022-07-24 07:29 279阅读 0赞

这种模型应该也可以归类为生产者消费者模型，但是我觉得这种实现方法效率应该没有用队列来实现效率高。后面再用队列来实现一次。  
![Figure1][]  
这个图就是这个程序所要实现的。

先定义两个结构体，`thread_sync_t`是保存着生产者和消费者共享的资源数组,`thread_var`是用来记录这个线程应该处理共享资源中哪一个资源的位置。

struct thread_sync_t{
        string* imagePath;
        bool* collectFinish;
        pthread_mutex_t* collectAndSend;
        pthread_cond_t* collectReady;
    };
    
    struct thread_var{
        int threadIndex;
        thread_sync_t* sync;
    };

一个生产者线程和多个消费者线程，我这里生产者线程是CollectImages，消费者线程是SendImages。  
CollectImages每隔1秒，产生张图片，并发出信号，请求发送进程处理。  
SendImages处理一张图片需要2秒。  
都是些模板代码，没什么好解释的。

对这些线程共享变量的初始化，每个线程都是共享`thread_sync_t`这个结构体的对象。

collectThread = new pthread_t[collectThreadCnt];
        sendThread = new pthread_t[sendThreadCnt];
        threadVar = new thread_var[sendThreadCnt];
        InitThreadVar(threadVar);

void InitThreadVar(thread_var* arg){
        thread_sync_t* psync = new thread_sync_t;
    
        psync->imagePath = new string[sendThreadCnt];
        psync->collectAndSend = new pthread_mutex_t[sendThreadCnt];
        psync->collectReady = new pthread_cond_t[sendThreadCnt];
        psync->collectFinish = new bool[sendThreadCnt];
    
        for(int i = 0; i < sendThreadCnt; ++i){
            pthread_mutex_init(&psync->collectAndSend[i], NULL);
            pthread_cond_init(&psync->collectReady[i], NULL);
            psync->collectFinish[i] = false;
        }
        for(int i = 0; i < sendThreadCnt; ++i){
            arg[i].threadIndex = i;
            arg[i].sync = psync;
        }
    }

全部源代码

#include <iostream>
    #include "util/DBHelper.h"
    #include <pthread.h>
    #include <string>
    #include <cstring>
    #include <ctime>
    #include <sstream>
    #include <unistd.h>
    
    using namespace std;
    
    void* CollectImages(void*);
    void* SendImages(void*);
    void err_exit(int, string);
    
    struct thread_sync_t{
        string* imagePath;
        bool* collectFinish;
        pthread_mutex_t* collectAndSend;
        pthread_cond_t* collectReady;
    };
    
    struct thread_var{
        //int cameraIndex;
        int threadIndex;
        thread_sync_t* sync;
    //    string imagePath;
    //    bool collectFinish;
    //    pthread_mutex_t* collectAndSend;
    //    pthread_cond_t* collectReady;
    };
    
    void InitThreadVar(thread_var*);
    void DestoryThreadVar(thread_var*);
    
    pthread_t* sendThread;
    pthread_t* collectThread;
    thread_var* threadVar;
    int collectThreadCnt = 1;
    int sendThreadCnt = 10;
    
    int main() {
        int err;
    
    
        collectThread = new pthread_t[collectThreadCnt];
        sendThread = new pthread_t[sendThreadCnt];
        threadVar = new thread_var[sendThreadCnt];
        InitThreadVar(threadVar);
    
        for(int i = 0; i < collectThreadCnt; ++i){
            err = pthread_create(&collectThread[i], NULL, CollectImages, (void*)threadVar);
            if(err != 0){
                err_exit(err, "create thread error");
            }
        }
    
    
        for(int i = 0; i < sendThreadCnt; ++i){
            err = pthread_create(&sendThread[i], NULL, SendImages, (void*)(threadVar + i));
            if(err != 0){
                err_exit(err, "create thread error");
            }
        }
    
        while(1){
    
        }
        return 0;
    }
    
    void InitThreadVar(thread_var* arg){
        thread_sync_t* psync = new thread_sync_t;
    
        psync->imagePath = new string[sendThreadCnt];
        psync->collectAndSend = new pthread_mutex_t[sendThreadCnt];
        psync->collectReady = new pthread_cond_t[sendThreadCnt];
        psync->collectFinish = new bool[sendThreadCnt];
    
        for(int i = 0; i < sendThreadCnt; ++i){
            pthread_mutex_init(&psync->collectAndSend[i], NULL);
            pthread_cond_init(&psync->collectReady[i], NULL);
            psync->collectFinish[i] = false;
        }
        for(int i = 0; i < sendThreadCnt; ++i){
            arg[i].threadIndex = i;
            arg[i].sync = psync;
        }
    
    
    }
    
    void DestoryThreadVar(thread_var* arg){
    
    }
    
    void* CollectImages(void* arg){
        bool cond = false;
        while(1){
    
            srand((int)time(0));
    
            for(int threadIndex = 0; threadIndex < sendThreadCnt; ++threadIndex){
                thread_var* threadArg = &((thread_var*) arg)[threadIndex];
    
                //cout << "thread: " << threadIndex << " collect image before lock" <<endl;
                pthread_mutex_lock(&threadArg->sync->collectAndSend[threadIndex]);
                //cout << "thread: " << threadIndex << " collect image lock" <<endl;
                int r = rand() % sendThreadCnt;
                stringstream ss;
                ss << "/home/honkee/Downloads/photo/" << r << ".jpg";
                ss >> threadArg->sync->imagePath[threadIndex];
                threadArg->sync->collectFinish[threadIndex] = true;
                cout << "thread: " << threadIndex << " collect finish: "
                        << threadArg->sync->imagePath[threadIndex] << endl <<flush;
    
                pthread_cond_signal(&threadArg->sync->collectReady[threadIndex]);
                pthread_mutex_unlock(&threadArg->sync->collectAndSend[threadIndex]);
                //cout << "thread: " << threadIndex << " collect image unlock" <<endl;
                sleep(1);
            }
    
    
            if(cond){
                break;
            }
        }
        printf("collect image thread\n");
    
    }
    
    void* SendImages(void* arg){
        thread_var* threadArg = (thread_var*) arg;
        bool cond = false;
        while(1){
            int threadIndex = threadArg->threadIndex;
            //cout << "thread: " << threadIndex << " send image before lock" <<endl;
            pthread_mutex_lock(&threadArg->sync->collectAndSend[threadIndex]);
            //cout << "thread: " << threadIndex << " send image lock" <<endl;
    
            while(threadArg->sync->collectFinish[threadIndex] == false){
                cout << "thread: "<< threadIndex << " waiting..." <<endl <<flush;
    
                pthread_cond_wait(&threadArg->sync->collectReady[threadIndex],
                                    &threadArg->sync->collectAndSend[threadIndex]);
            }
            cout << "thread: " << threadIndex << " send finish: "
                    << threadArg->sync->imagePath[threadIndex] << endl <<flush;
            threadArg->sync->collectFinish[threadIndex] = false;
    
            sleep(2);
            pthread_mutex_unlock(&threadArg->sync->collectAndSend[threadIndex]);
            //cout << "thread: " << threadIndex << " send image unlock" <<endl;
    
            if(cond){
                break;
            }
        }
    
        printf("send image thread\n");
    }
    
    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_comsume1.png