线程池模型案列

电玩女神 2023-07-05 10:26 7阅读 0赞

线程是一种池化技术，也是线程的一种使用模式

### 模型 ###

创建一大堆线程和一个安全队列 , 循环从队列中获取任务对象，获取到任务对象后，执行任务对象中的任务接口.

线程池线程的数量取决于可用的并发处理器、处理器内核、内存、网络 sockets 等的数量

由于线程池当中的线程都是采用同一个入口函数，所以入口函数当中的功能一致，怎么处理队列当中不同的任务需求？

可以将处理函数设置为函数指针，将处理任务的函数指针和数据一起作为入口参数抛进线程池

### 线程池的应用场景 ###

如果大量的客户请求涌入服务器，这时候服务器不事先创建线程池，而是接收一个请求，创建一个线程，接收一个请求，创建一个线程，这样对操作系统调度线程带来的开销是非常大的，处理请求的效率特别低，而且如果无限制的开辟新的线程，可能会导致操作系统资源耗尽，造成系统崩溃

如果有对性能要求苛刻的应用，比如要求服务器迅速响应客户请求，也可以采用线程池的技术

所以采用线程池的技术，在程序初始化的时候根据系统资源创建一大堆线程，这样就可以节省当有请求来的时候创建线程消耗的时间和销毁线程的时间

参考代码

#include <stdio.h>
    #include <iostream>
    #include <pthread.h>
    #include <unistd.h>
    #include <queue>
    
    typedef void (*Hander_t)(int);
    #define PTHREADCOUNT 4
    
    // 定义一个任务类
    class ThreadTask{ 
    public:
        ThreadTask(){ 
            data_ = -1;
            hander_ = NULL;
        }
        ThreadTask(int data,Hander_t hander){ 
            data_ = data;
            hander_ = hander;
        }
        // 任务执行函数
        void Run(){ 
            hander_(data_);
        }
    private:
        int data_;
        Hander_t hander_;
    };
    
    class PthreadPool{ 
    public:
        PthreadPool(){ 
            PthreadCapacity_ = PTHREADCOUNT; 
            PthreadCurNum_ = PTHREADCOUNT; 
            pthread_mutex_init(&mutex_,NULL);
            pthread_cond_init(&cond_,NULL);
            IsEsc = false;
            // 创建线程函数
            ThreadCreate();
        }
        ~PthreadPool(){ 
            pthread_mutex_destroy(&mutex_);
            pthread_cond_destroy(&cond_);
        }
        // 线程的清理
        void ClearPool(){ 
            PoolClear();
        }
    private:
        // 创建线程
        void ThreadCreate(){ 
            pthread_t pid[PTHREADCOUNT];
            for(int i =0;i < PTHREADCOUNT;++i){ 
                pthread_create(&pid[i],NULL,StartCon,(void*)this);
            }
        }
        // 线程入口函数
        static void* StartCon(void* arg){ 
           PthreadPool* tp = (PthreadPool*)arg; 
           while(1){ 
               pthread_mutex_lock(&tp->mutex_);
               while(tp->Que_.empty()){ 
                   if(tp->IsEsc){ 
                       tp->ThreadQuit();
                   }
                    pthread_cond_wait(&tp->cond_,&tp->mutex_);
               }
               pthread_mutex_unlock(&tp->mutex_);
               ThreadTask* tt;
               tp->Pop(&tt);
               pthread_mutex_unlock(&tp->mutex_);
               tt->Run();
               delete tt;
           }
        }
    public:
        // 队列插入的数据
        void Push(ThreadTask* Tt) { 
            pthread_mutex_lock(&mutex_);
            if(IsEsc){ 
                pthread_mutex_unlock(&mutex_);
                return;
            }
            Que_.push(Tt);
            pthread_mutex_unlock(&mutex_);
            pthread_cond_signal(&cond_);
        }
        // 队列取数据
        void Pop(ThreadTask** Tt){ 
            *Tt = Que_.front();
            Que_.pop();
        }
    
        // 清理线程池
        // 唤醒等待队列中的线程
        void PoolClear(){ 
            pthread_mutex_lock(&mutex_);
            IsEsc = true;
            pthread_mutex_unlock(&mutex_);
            if(PthreadCurNum_ > 0){ 
                pthread_cond_broadcast(&cond_);
            }
        }
    private:
        // 线程的退出
        void ThreadQuit(){ 
            PthreadCurNum_ --;
            pthread_mutex_unlock(&mutex_);
            pthread_exit(NULL);
        }
    private:
        // 线程池初始化的线程数量
        size_t PthreadCapacity_;
        size_t PthreadCurNum_;
        // 线程池队列
        std::queue<ThreadTask*>Que_;
        // 同步与互斥
        pthread_mutex_t mutex_;
        pthread_cond_t cond_;
        // 退出标志
        bool IsEsc;
    };
    
    void Print(int data){ 
        printf("i am Hander func ,i print [%d]\n",data);
    }
    
    int main(){ 
        PthreadPool* tp = new PthreadPool();
        for(int i =0; i < 10; ++i){ 
            ThreadTask* tt = new ThreadTask(i,Print);
            tp->Push(tt);
        }
        sleep(4);
        // 清理线程池
        tp->ClearPool();
        delete tp;
        return 0;
    }

程序当中的线程是为了处理业务而生的，当我们不断的进行请求的时候，这种测试系统性能的方法叫做压力测试

### 压力测试 ###

构造请求发送给服务器后台

服务器后台可分为两个模块

*  接收模块
 *  业务处理模块
 *  处理完成后的发送模块

接收模块将客户请求放到任务队列中，然后业务模块从任务队列中取出任务对象，并且需要分清楚具体的业务处理逻辑是属于哪个子模块，比如说子模块1，子模块2…，每一个子模块也有一个队列，用于缓冲的作用，每个子模块中的线程进行处理任务，将任务处理完成后发送给客户.