Linux中epoll+线程池实现高并发

创建一个epoll实例;
while(server running)
{
    epoll等待事件;
    if(新连接到达且是有效连接)
    {
        accept此连接;
        将此连接设置为non-blocking;
　　     为此连接设置event(EPOLLIN | EPOLLET ...);
        将此连接加入epoll监听队列;
        从线程池取一个空闲工作者线程并处理此连接;
    }
    else if(读请求)
    {
        从线程池取一个空闲工作者线程并处理读请求;
    }
    else if(写请求)
    {
        从线程池取一个空闲工作者线程并处理写请求;
    }
    else
        其他事件;     
}

/*************************************************************************
    > File Name: lock.h
    > Author: gushi
    > Mail: 971859774@qq.com 
    > Created Time: 2018年11月22日 星期四 20时06分55秒
 ************************************************************************/

#ifndef LOCK_H
#define LOCK_H
#include <iostream>
#include <pthread.h>
#include <semaphore.h>

using namespace std;

class Sem
{
    private:
        sem_t sem;

    public:
        Sem();
        ~Sem();
        bool wait();
        bool post();
};

Sem::Sem()
{
    if(sem_init(&sem,0,0)!=0)//信号量的初始值和和基于内存的信号量
        cerr<<"sem init error."<<endl;
}

Sem::~Sem()
{
    sem_destroy(&sem);
}

bool Sem::wait()
{
    return sem_wait(&sem)==0?true:false;
}

bool Sem::post()
{
    return sem_post(&sem)==0?true:false;
}

//互斥类
class Mutex
{
    private:
        pthread_mutex_t mutex;

    public:
        Mutex();
        ~Mutex();
        bool mutex_lock();
        bool mutex_unlock();
};

Mutex::Mutex()
{
    if(pthread_mutex_init(&mutex,NULL)!=0)//可用PTHRAD_MUTEX_INITIALIZER宏初始化
        cerr<<"mutex init error"<<endl;
}

Mutex::~Mutex()
{
    pthread_mutex_destroy(&mutex);
}

bool Mutex::mutex_lock()
{
    return pthread_mutex_lock(&mutex)==0?true:false;
}

bool Mutex::mutex_unlock()
{
    return pthread_mutex_unlock(&mutex)==0?true:false;
}

//条件变量的类
class Cond
{
    private:
        pthread_mutex_t mutex;
        pthread_cond_t cond;

    public:
        Cond();
        ~Cond();
        bool wait();
        bool signal();
        bool broadcast();
};

Cond::Cond()
{
    if(pthread_mutex_init(&mutex,NULL)!=0)
    {
        cerr<<"Cond mutex init error"<<endl;
        exit(0);
    }
    if(pthread_cond_init(&cond,NULL)!=0)
    {
        cerr<<"Cond cond init error"<<endl;
        pthread_mutex_destroy(&mutex);
        exit(0);
    }
}

Cond::~Cond()
{
    pthread_mutex_destroy(&mutex);
    pthread_cond_destroy(&cond);
}

bool Cond::wait()
{
    int rs=0;
    pthread_mutex_lock(&mutex);
    rs=pthread_cond_wait(&cond,&mutex);
    pthread_mutex_unlock(&mutex);
    return rs==0?true:false;
}

bool Cond::signal()
{
    return pthread_cond_signal(&cond)==0?true:false;
}

bool Cond::broadcast()
{
    return pthread_cond_broadcast(&cond);
}

#endif

/*************************************************************************
    > File Name: threadpool.h
    > Author: gushi
    > Mail: 971859774@qq.com 
    > Created Time: 2018年11月22日 星期四 20时32分12秒
 ************************************************************************/

#ifndef THREADPOOL_H
#define THREADPOOL_H

#include <queue>
#include <vector>
#include <exception>
#include <errno.h>
#include "lock.h"
#define MAX_THREADS 1024

using namespace std;

template <class T>
class Threadpool
{
    private:
        int idle;//线程池中空闲线程的数量
        int num;//线程池中线程数
        vector<pthread_t> idle_tid;//空闲线程的集合
        vector<pthread_t> busy_tid;//正在执行任务的线程的集合
        queue<T *> task_queue;//任务队列
        Mutex mutex;//互斥锁
        Cond cond;//条件变量锁
        bool is_stop;//是否结束线程

    public:
        static void *worker(void *arg);//线程函数，里面执行run函数
        void run();
        T *get_task();//获取任务函数
        int mv_to_idle(pthread_t tid);//执行任务完成后，放入空闲
        int mv_to_busy(pthread_t tid);//移入到忙碌线程中

    public:
        Threadpool(int n=20);
        ~Threadpool();
        bool append_task(T *task);//添加任务函数
        void start();//开始创建线程池
        void stop();//线程停止函数
};

template <class T>
Threadpool<T>::Threadpool(int n):num(n),idle(n),is_stop(false)
                                       
{
    if(num<=0)
    {
        cerr<<"threadpool can‘t init because num<=0."<<endl;
        exit(1);
    }
}

template <class T>
Threadpool<T>::~Threadpool()
{
    stop();
}

template <class T>
bool Threadpool<T>::append_task(T *task)
{
    mutex.mutex_lock();//临界资源上锁

    bool is_signal=task_queue.empty();
    task_queue.push(task);

    mutex.mutex_unlock();//解锁

    if(is_signal)//signal to null queue
        cond.signal();

    return true;
}

template <class T>
void Threadpool<T>::start()
{
    for(int i=0;i<num;++i)    
    {
        pthread_t tid=0;
        if(pthread_create(&tid,NULL,worker,this)!=0)//this参数的传递对于开启线程运行函数要用到，
        {
            throw exception();
            exit(1);
        }
        idle_tid.push_back(tid);//加入到空闲线程集合
    }
}

template <class T>
void Threadpool<T>::stop()
{
    is_stop=true;
    cond.broadcast();
}

template <class T>
void *Threadpool<T>::worker(void *arg)
{
    Threadpool<T> *thread=(Threadpool<T> *)arg;//thread为一个线程池的指针，指向整个线程池，
    thread->run();//调用线程运行函数，真正执行工作的函数
    return thread;
}

template <class T>
void Threadpool<T>::run()
{
    pthread_t tid=pthread_self();//mutex.mutex_lock();这会造成与append_task函数构成死锁，访问task_queue与idle_tid与busy_tid等临界资源时，可以再各自的函数实现中加锁
    while(1)//if
    {
        T *task=get_task();//函数实现中有互斥锁临界访问
        if(task==NULL)
        {
            cerr<<"task_queue is null.wait()"<<endl;
            cond.wait();
        }
        else
        {
            mv_to_busy(tid);//函数实现中有互斥锁保护访问
            task->doit();//工作函数
            mv_to_idle(tid);
        }
    }
    //mutex.mutex_unlock();
}

template <class T>
T *Threadpool<T>::get_task()
{
    T *task=NULL;
    
    mutex.mutex_lock();
    if(!task_queue.empty())
    {
        task=task_queue.front();
        task_queue.pop();
    }
    mutex.mutex_unlock();

    return task;
}

template <class T>
int Threadpool<T>::mv_to_idle(pthread_t tid)
{
    vector<pthread_t>::iterator busy_iter=busy_tid.begin();
    while(busy_iter!=busy_tid.end())
    {
        if(tid==*busy_iter)
            break;
        ++busy_iter;
    }    
    
    mutex.mutex_lock();
    busy_tid.erase(busy_iter);//此线程空闲，从繁忙任务队列中移除
    idle_tid.push_back(tid);//添加到空闲线程集合中

    //mutex.mutex_lock();
    ++idle;
    mutex.mutex_unlock();
    return 0;
}

template <class T>
int Threadpool<T>::mv_to_busy(pthread_t tid)
{
    vector<pthread_t>::iterator idle_iter=idle_tid.begin();
    while(idle_iter!=idle_tid.end())
    {
        if(tid==*idle_iter)
            break;
        ++idle_iter;
    }
    mutex.mutex_lock();
    idle_tid.erase(idle_iter);
    busy_tid.push_back(tid);

    //mutex.mutex_lock();
    --idle;
    mutex.mutex_unlock();
}

#endif

/*************************************************************************
    > File Name: epollserver.h
    > Author: gushi
    > Mail: 971859774@qq.com 
    > Created Time: 2018年11月23日 星期五 17时07分25秒
 ************************************************************************/

#ifndef EPOLL_SERVER_H
#define EPOLL_SERVER_H

#include <sys/socket.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <sys/epoll.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <string.h>
#include <fcntl.h>
#include <unistd.h>//write头文件
#include <errno.h>
#include "threadpool.h"

#define MAX_EVENT 1024
#define MAX_BUFFER 2048
using namespace std;

class Basetask
{
    public:
        virtual void doit()=0;
};

class Task:public Basetask
{
    private:
        int sockfd;
        char msg[MAX_BUFFER];
    
    public:
        Task(int ,char *);
        void doit();
};

Task::Task(int fd,char *str):sockfd(fd)
{
    memset(msg,0,sizeof(msg));
    strcpy(msg,str);
}

void Task::doit()
{
    cout<<"server reveive message is: "<<msg<<endl;
    write(sockfd,msg,strlen(msg));
    //Threadpool<Task>::mv_to_idle(tid);
}

class Epollserver
{
    private:
        bool is_stop;//是否停止epoll_wait
        int num;//线程数目
        int sockfd;
        int port;
        int epollfd;
        Threadpool<Task> *pool;//线程池的指针
        epoll_event events[MAX_EVENT];
        struct sockaddr_in servaddr;
    
    public:
        Epollserver(int p,int n);
        Epollserver(){}
        ~Epollserver();
        void init();
        void epoll();
        static int setnonblocking(int fd);
        static void addfd(int epollfd,int sockfd,bool onshot);
};
Epollserver::Epollserver(int p,int n):port(p),num(n),is_stop(false),
                        pool(NULL)
{

}

Epollserver::~Epollserver()
{
    delete pool;
}

void Epollserver::init()
{
    bzero(&servaddr,sizeof(servaddr));
    servaddr.sin_family=AF_INET;
    servaddr.sin_addr.s_addr=htonl(INADDR_ANY);
    servaddr.sin_port=htons(port);

    //监听套接字
    sockfd=socket(AF_INET,SOCK_STREAM,0);
    if(sockfd<0)
    {
        cerr<<"Epollserver socket init error"<<endl;
        exit(1);
    }

    int tmp=bind(sockfd,(struct sockaddr *)&servaddr,sizeof(servaddr));
    if(tmp<0)
    {
        cerr<<"Epollserver bind init error"<<endl;
        exit(1);
    }

    tmp=listen(sockfd,10);
    if(tmp<0)
    {
        //cout<<errno<<endl;
        cerr<<"Epollserver listen init error:"<<strerror(errno)<<endl;
        exit(1);
    }

    epollfd=epoll_create(1024);
    if(epollfd<0)
    {
        cerr<<"Epollserver epoll_create init error"<<endl;
        exit(1);
    }

    //创建线程池,num是线程池中线程的个数,调用构造函数
    pool=new Threadpool<Task>(num); 
}

void Epollserver::epoll()
{
    pool->start();//启动线程池

    addfd(epollfd,sockfd,false);
    while(!is_stop)
    {
        int ret=epoll_wait(epollfd,events,MAX_EVENT,-1);
        if(ret<0)
        {
            cerr<<"epoll_wait error"<<endl;
            exit(1);
        }

        for(int i=0;i<ret;++i)
        {
            //int fd=events[i].data.fd;
            if(events[i].data.fd==sockfd)
            {
                struct sockaddr_in cliaddr;
                socklen_t len=sizeof(cliaddr);
                //accpet 返回已连接套接字
                int confd=accept(sockfd,(struct sockaddr *)&cliaddr,&len);
                Epollserver::addfd(epollfd,confd,false);
            }
            else if(events[i].events&EPOLLIN)//有数据可读
            {
                char buffer[MAX_BUFFER];
                int fd=events[i].data.fd;//接受已连接套接字，对客户端进行内容回送
    readagain:    memset(buffer,0,sizeof(buffer));
                ret=read(fd,buffer,MAX_BUFFER-1);
                if(ret==0)//某个fd关闭连接
                {
                    struct epoll_event ev;
                    ev.events=EPOLLIN;
                    ev.data.fd=fd;
                    epoll_ctl(epollfd,EPOLL_CTL_DEL,fd,&ev);
                    shutdown(fd,SHUT_RDWR);
                    cout<<fd<<" exit"<<endl;
                    continue;
                }
                else if(ret<0)//读取失败
                {
                    if(errno==EAGAIN)
                    {
                        cout<<"read error. read again"<<endl;
                        goto readagain;
                        break;
                    }
                }
                else//读取成功
                {
                    Task *task=new Task(fd,buffer);
                    pool->append_task(task);
                }
            }//else if
            else
                cerr<<"something else had happend"<<endl;
        }//for
    }//while
    close(sockfd);
    pool->stop();
}

int Epollserver::setnonblocking(int fd)
{
    int old_opt=fcntl(fd,F_GETFL);
    int new_opt=old_opt|O_NONBLOCK;
    fcntl(fd,F_SETFL,new_opt);
    return old_opt;
}

void Epollserver::addfd(int epollfd,int sockfd,bool oneshot)
{
    epoll_event event;
    event.data.fd=sockfd;
    event.events=EPOLLIN|EPOLLET;
    if(oneshot)
        event.events|=EPOLLONESHOT;
    epoll_ctl(epollfd,EPOLL_CTL_ADD,sockfd,&event);
    Epollserver::setnonblocking(sockfd);
}

#endif

/*************************************************************************
    > File Name: server.cpp
    > Author: gushi
    > Mail: 971859774@qq.com 
    > Created Time: 2018年11月23日 星期五 19时34分29秒
 ************************************************************************/

#include "epollserver.h"
#define INDARRY_PORT 9877

using namespace std;

int main(int argc,char **argv)
{
    Epollserver *epoll=new Epollserver(INDARRY_PORT,20);
    epoll->init();//对初始化服务器(socket，bind，listen,epoll_create...等函数，病完成线程池的初始化)
    epoll->epoll();//开启线程池，完成相应的任务添加之后，自动调用线程池中空闲的函数来完成doit工作
    return 0;
}

/*************************************************************************
    > File Name: client.cpp
    > Author:gushi
    > Mail: 971859774@qq.com 
    > Created Time: 2018年11月24日 星期六 15时36分23秒
 ************************************************************************/

#include <iostream>
#include <sys/socket.h>
#include <sys/types.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <netinet/in.h>
#include <strings.h>
#define SERV_PORT 9877
#define MAXLINE 1204

using namespace std;

void str_cli(FILE *fp,int sockfd)
{
    fd_set set;
    FD_ZERO(&set);

    char buff[1024];
    int stdineof=0,n;
    while(1)
    {
        if(stdineof==0)
            FD_SET(fileno(fp),&set);
        FD_SET(sockfd,&set);

        int maxfd=max(fileno(fp),sockfd)+1;
        
        select(maxfd,&set,NULL,NULL,NULL);

        if(FD_ISSET(sockfd,&set))
        {
            if((n=read(sockfd,buff,MAXLINE))==0)
                if(stdineof==1)
                    return;
                else
                    cerr<<"str_cli: server terinated peraturely"<<endl;

            write(fileno(stdout),buff,n);
        }
        else if(FD_ISSET(fileno(fp),&set))
        {
            if((n=read(fileno(fp),buff,MAXLINE))==0)//客户完成输入
                stdineof=1;
            write(sockfd,buff,n);

            //shutdown(sockfd,SHUT_WR);

            FD_CLR(fileno(fp),&set);
            continue;
        }
        //write(sockfd,buff,n);
    }
    return;
}

int main(int argc,char **argv)
{
    if(argc!=2)
    {
        cerr<<"please input server address."<<endl;
        exit(1);
    }

    int sockfd=socket(AF_INET,SOCK_STREAM,0);

    struct sockaddr_in servaddr;
    bzero(&servaddr,sizeof(servaddr));
    servaddr.sin_family=AF_INET;
    inet_pton(AF_INET,argv[1],&servaddr.sin_addr);
    servaddr.sin_port=htons(SERV_PORT);
    
    connect(sockfd,(struct sockaddr *)&servaddr,sizeof(servaddr));

    str_cli(stdin,sockfd);
    return 0;
}