epoll+線程池實現高併發

版權聲明：本文爲博主原創文章，遵循 CC 4.0 BY-SA 版權協議，轉載請附上原文出處連接和本聲明。
本文連接：https://blog.csdn.net/qq_38506897/article/details/82863066
 

     epoll模型是linux實現高併發的一種方法，基於事件驅動模型，相比於select/poll  模型具備更高的效率，本人對epoll模型作了一個簡易的封裝，更多的功能還在完善中，在這裏僅作學習參考用。

     在epoll編程中，有三個很是重要的函數：
      1. int epoll_create(int size)     :建立epoll 句柄， 入參是表示監聽的數目是多大。
      2. int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event)   ： 事件註冊函數， @param1：epoll句柄，epoli_create()的返回值，  @param2： 表示註冊的行爲， 有ADD事件 、MOD事件、DEL事件， @param3： 註冊的fd，在網絡編程中，通常爲sockfd，@param4：表示事件類型，
      3. int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout)       等待事件的產生，相似於select() 調用。

 

       在線程池上，以前的博文寫過一篇關於線程池的實現，拿來即用便可。造輪子的過程雖然很枯燥也很痛苦，但當你作一個項目用到你造的輪子時仍是頗有成就感的（菜鳥的成就感~）。

    封裝的比較簡單，後續會完善。

#ifndef _CEVENT_H_
#define _CEVENT_H_
 
 
#include <sys/socket.h>
#include <sys/types.h>
#include <stdio.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <errno.h>
#include <string.h>
#include <stdlib.h>
#include <fcntl.h>
#include <sys/wait.h>
#include <sys/epoll.h>
 
#define MAX_SIZE 1024
 
enum EventType
{
    EIN = EPOLLIN,          // 讀事件
    EOUT = EPOLLOUT,      // 寫事件
    ECLOSE = EPOLLRDHUP,  // 對端關閉鏈接或者寫半部
    EPRI = EPOLLPRI,      // 緊急數據到達
    EERR = EPOLLERR,      // 錯誤事件
    EET = EPOLLET,           // 邊緣觸發
    EDEFULT = EIN | ECLOSE | EERR | EET
};
 
 
class CEvent
{
public:
    CEvent();
    ~CEvent();
    int Register_event(int fd,     EventType type = EDEFULT);
    int unRegister_event(int fd);
    void* EventHandle(void* arg);
    void SetNoblocking(int v_sockfd);
    
private:
    int epfd;
    bool is_running;
    pthread_t  m_tid;
    struct events[EPOLL_SIZE];
    CThreadPoolProxy *pool;
};
 
CEvent::CEvent()
{
    
    epfd = epoll_create(MAX_SIZE);
    if(epfd == -1)
    {
        printf("epoll_create failed.");
        return -1;
    }
    pthread_t tid = 0;
    pthread_create(&tid, NULL, EventHandle, (void*)this == 0);
    m_tid = tid;
    //線程池初始化
    pool = CThreadPoolProxy::instance();
    
}
 
 
CEvent::~CEvent()
{
    if(pthread_cancel(m_tid) == 0)
    {
        pthread_join(m_tid, (void **)NULL);
    }
}
 
 
 
void CEvent::SetNoblocking(int v_sockfd)
{
    int opts = fcntl(v_sockfd,F_GETFL);
    if(opts < 0)
    {
        printf("fcntl(sockfd, F_GETFL) failed.");
        opts = opts|O_NONBLOCK;
    }
    fcntl(v_sockfd, F_SETFL, opts);
    
}
 
 
 
 
 
 
int CEvent::Register_event(int fd, EventType type = EDEFULT)
{
    SetNoblocking(fd);
    struct epoll_event ev;
    ev.data.fd = fd
    ev.events = type;
    if(epoll_ctl(epfd, EPOLL_CTL_ADD, fd, &ev) == -1)
    {
        printf("epoll_ctl: EPOLL_CTL_ADD failed, fd[%d].",&fd);
        return -1;
    }
    return 0;
}
 
 
int CEvent::unRegister_event(int fd)
{
    if(epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL) == -1)
    {
        printf("epoll_ctl: EPOLL_CTL_DEL failed, fd[%d].",&fd);
        return -1;
    }
    return 0;
}
 
 
void* CEvent::EventHandle(void* arg)
{
    
    CEvent &event = *(CEvent*)arg;
    while(is_running)
    {
        int ret = epoll_wait(event.epfd, event.events, MAX_SIZE, -1);
        if(ret < 0)
        {
            printf("epoll_wait failed, epfd[%d]",&event.epfd);
        }
        for(int i=0; i<ret; i++)
        {
            int connfd = event.events[i].data.fd;
            
            if(event.events[i].events & EPOLLIN)
            {
                CTask* ta=new CMyTask;       //  具體的方法本身實現。
                ta->SetConnFd(connfd);
                pool->AddTask(ta);
                
            }
        }
    }
    
    
    
}
 
#endif
線程池代碼：

#ifndef __THREAD_H  
#define __THREAD_H  
 
/*********************
** Filename: Thread.h
** Dsricbe: 線程池頭文件
** Date: 2018.7.18
** @author: Mr.xl
***/
 
 
#include <deque>
#include <string>  
#include <pthread.h>  
  
using namespace std;  
  
/** 
 * 執行任務的類，設置任務數據並執行 
 */  
class CTask {
protected:
    string m_strTaskName;   //任務的名稱
    int connfd;    //接收的地址
 
public:
    CTask() = default;
    CTask(string &taskName): m_strTaskName(taskName), connfd(NULL) {}
    virtual int Run() = 0;
    void SetConnFd(int data);   //設置接收的套接字鏈接號。
    int GetConnFd();
    virtual ~CTask() {}
    
};
 
 
 
 
/** 
 * 線程池類的實現 
 */  
class CThreadPool  
{  
private:  
    static  deque<CTask*> m_deqTaskList;     /** 任務隊列 */  
    static  bool shutdown;                    /** 線程退出標誌 */           
    int     m_iThreadNum;                     /** 線程池中啓動的線程數 */  
    pthread_t   *pthread_id;  
      
    static pthread_mutex_t m_pthreadMutex;    /** 線程同步鎖 */  
    static pthread_cond_t m_pthreadCond;      /** 線程同步的條件變量 */  
  
protected:  
    static void* ThreadFunc(void * threadData); /** 新線程的線程回調函數 */  
    static int MoveToIdle(pthread_t tid);       /** 線程執行結束後，把本身放入到空閒線程中 */  
    static int MoveToBusy(pthread_t tid);       /** 移入到忙碌線程中去 */  
      
    int Create();          /** 建立線程池中的線程 */  
  
public:  
    CThreadPool(int threadNum = 10);  
    ~CThreadPool();
    int AddTask(CTask *task);      /** 把任務添加到任務隊列中 */  
    int StopAll();                 /** 使線程池中的線程退出 */  
    int getTaskSize();             /** 獲取當前任務隊列中的任務數 */  
};  
// 代理類,只暴露給別人用的
class CThreadPoolProxy: public CThreadPool
{
public:
    static CThreadPool* instance()
    {
        if(NULL == m_pInstance)
        {
            m_pInstance = new CThreadPoolProxy;
            return m_pInstance;
        }
        return m_pInstance;
    }
    int AddTask(CTask *task)
    {
        return m_pthreadpool->AddTask(task);
    }
    
private:
    CThreadPoolProxy()
    {
        m_pthreadpool = new CThreadPool(5)
    }
    ~CThreadPoolProxy()
    {
        delete m_pthreadpool;
    }
    
private:
    static CThreadPoolProxy* m_pInstance;
    CThreadPool* m_pthreadpool;
};
 
 
#endif  
/******************
** Fliename: Thread.cpp
** Dscribe: 線程池實現文件
** Date: 2018.7.18
** @author: Mr.xl
***/
 
 
#include "Thread.h"  
#include <iostream>  
#include <stdio.h> 
#include <stdlib.h>
#include <deque>
 
 
 
void CTask::SetConnFd(int data)  
{  
    connfd = data;  
}  
 
int CTask::GetConnFd()
{
    return connfd;
}
/**
* 初始化數據
*/
deque<CTask*> CThreadPool::m_deqTaskList;         //任務列表  
bool CThreadPool::shutdown = false;  
      
pthread_mutex_t CThreadPool::m_pthreadMutex = PTHREAD_MUTEX_INITIALIZER;   
pthread_cond_t CThreadPool::m_pthreadCond = PTHREAD_COND_INITIALIZER;  
  
/** 
 * 線程池管理類構造函數 
 */  
CThreadPool::CThreadPool(int threadNum)  
{  
    this->m_iThreadNum = threadNum;  
    cout << "I will create " << threadNum << " threads" << endl;  
    Create();       //*建立對象時便建立線程。
}
 
CThreadPool::~CThreadPool()
{
    pthread_mutex_destroy(&m_pthreadMutex);
    pthread_cond_destroy(&m_pthreadCond);
    
}
 
/** 
 * 線程回調函數 
 */  
void* CThreadPool::ThreadFunc(void* threadData)  
{  
    pthread_t tid = pthread_self();  
    while (1)  
    {  
 
        //* 線程開啓時先上鎖 */
        pthread_mutex_lock(&m_pthreadMutex);  
        while (m_deqTaskList.size() == 0 && !shutdown)  
        {  
            //* 沒有任務時，線程等待狀態（條件變量）*/
            pthread_cond_wait(&m_pthreadCond, &m_pthreadMutex);  
        }  
          
        if (shutdown)  
        {  
            pthread_mutex_unlock(&m_pthreadMutex);  
            printf("thread %lu will exit\n", pthread_self());  
            pthread_exit(NULL);   
        }  
          
        printf("tid %lu run\n", tid);  
            
        /** 
        * 取任務隊列並處理之 
        */ 
 
        //deque<CTask*>::iterator iter = m_deqTaskList.front();
        CTask* task = m_deqTaskList.front();  
        m_deqTaskList.pop_front();
 
        //* 取完任務後釋放鎖*/
        pthread_mutex_unlock(&m_pthreadMutex);  
          
        task->Run(); /** 執行任務 */  
         
    }  
    return (void*)0;  
}  
 
 
 
 
 
/** 
 * 往任務隊列裏邊添加任務併發出線程同步信號 
 */  
int CThreadPool::AddTask(CTask *task)  
{  
    pthread_mutex_lock(&m_pthreadMutex);  
    this->m_deqTaskList.push_back(task);  
    pthread_mutex_unlock(&m_pthreadMutex); 
 
    // * 添加任務 條件變量發信號，非阻塞  */
    pthread_cond_signal(&m_pthreadCond);       
    return 0;  
}  
  
/** 
 * 建立線程 
 */  
int CThreadPool::Create()  
{  
    pthread_id = (pthread_t*)malloc(sizeof(pthread_t) * m_iThreadNum);  
    for(int i = 0; i < m_iThreadNum; i++)  
    {  
        pthread_create(&pthread_id[i], NULL, ThreadFunc, NULL);  
    }  
    return 0;  
}  
 
 
 
 
 
/** 
 * 中止全部線程 
 */  
int CThreadPool::StopAll()  
{  
    /** 避免重複調用 */  
    if (shutdown)  
    {  
        return -1;    
    }  
    printf("Now I will end all threads!!\n");  
    /** 喚醒全部等待線程，線程池要銷燬了 */  
    shutdown = true;  
    pthread_cond_broadcast(&m_pthreadCond);  
      
    /** 阻塞等待線程退出，不然就成殭屍了 */  
    for (int i = 0; i < m_iThreadNum; i++)  
    {  
        pthread_join(pthread_id[i], NULL);    
    }  
      
    free(pthread_id);  
    pthread_id = NULL;  
      
    /** 銷燬條件變量和互斥體 */  
    pthread_mutex_destroy(&m_pthreadMutex);  
    pthread_cond_destroy(&m_pthreadCond);  
      
    return 0;  
}  
 
/** 
 * 獲取當前隊列中任務數 
 */  
int CThreadPool::getTaskSize()  
{  
    return m_deqTaskList.size();      
}  

————————————————
版權聲明：本文爲CSDN博主「可樂小浣熊」的原創文章，遵循 CC 4.0 BY-SA 版權協議，轉載請附上原文出處連接及本聲明。
原文連接：https://blog.csdn.net/qq_38506897/article/details/82863066