用一个示例讲解我是如何处理高并发的

去年做了一个远程升级的服务。客户端连接此服务可以下载更新程序。简单点说就是个TCP sever。基于C++。

运行环境是centOS 6.5。

刚开始客户端数量少而且访问不频繁，所以没太关注并发的问题。当时用工具测试大概只能支持的40次/秒的并发访问，而且已经有数据串包的情况出现了。最近有空做了不少的优化并记录了笔记备忘。

下面给出的代码都不是完整的项目源码，我只是截取了关键部分用于说明主题

我选择的测试工具是一个小的tcp客户端工具，可以比价快捷的进行多客户端连接的测试。

线程安全的单例模式

最初的版本是通过多线程实现高并发的。我的工程里有两个类是单例模式，一个参数文件管理类，一个是日志管理类。原来我实现的时候没有考虑线程安全，于是第一步我先把这两个类改成线程安全测试看看效果。

增加线程安全前的代码片段（只给出参数文件管理类的实现）

//.h
class AppCof {
public:
    static AppCof* open_cof();

private:
    AppCof();

    class CGarbo   //它的唯一工作就是在析构函数中删除CSingleton的实例  
    {  
    public:  
        ~CGarbo()  
        {  
            if(AppCof::m_pInstance)  
                delete AppCof::m_pInstance;  
        }  
    };  
    static CGarbo Garbo;  //定义一个静态成员变量，程序结束时，系统会自动调用它的析构函数  
...

//.cpp
AppCof* AppCof::open_cof(){

    if(m_pInstance == NULL){
        m_pInstance = new AppCof();
    }
    return m_pInstance;
}
...

增加线程安全后

//.h
class AppCof:boost::noncopyable
{
public:
    static AppCof* open_cof();
private:
    AppCof();
    static AppCof *m_pInstance;
    static void init();
    static pthread_once_t ponce_;
    ...

void AppCof::init()
{
    m_pInstance = new AppCof();
    if(m_pInstance != NULL)
    {
        m_pInstance->get_env();
        m_pInstance->read_cof();
    }
}

AppCof* AppCof::open_cof(){

    pthread_once(&ponce_, &AppCof::init);
    return m_pInstance;
}

...

这里有两个重点，一是pthread_once的用法，还有就是boost::noncopyable。

先说说前者，

int pthread_once(pthread_once_t *once_control, void (*init_routine) (void))；

本函数使用初值为PTHREAD_ONCE_INIT的once_control变量保证init_routine()函数在本进程执行序列中仅执行一次。在多线程编程环境下，尽管pthread_once()调用会出现在多个线程中，init_routine()函数仅执行一次，究竟在哪个线程中执行是不定的，是由内核调度来决定。

boost::noncopyable这种用法其实从名字可以窥探一二，一个类继承自它就表示该类不能通过赋值，复制等手段创建新的对象了。

优化IO读写机制

这部分从select,epoll这些IO处理上下手。我优化过三个方案分别测试，效果还是比较明显。

最初的代码是这样的:

void run_srv(const char* i_port){
    for(;;){
        client = server.accept_client();
        std::thread t1(base_proc,client,i_port);
        usleep(200);
        t1.detach();
    }

}

void base_proc(Socket::TCP* i_client,const char* i_port){
    Socket::TCP* client = i_client;
    i_client = NULL;
    TmsProc* tmpc =new  TmsProc(client);
    tmpc->run();
    delete tmpc;
    return ;
}

void TmsProc::run(){
    Writelog::Trace(9,"业务处理开始");
    try{

        for(;;){
        tm.tv_sec = 3;
        tm.tv_usec = 0;
        FD_ZERO(&set);
        FD_SET(p_client->_socket_id,&set);

            int iret = select(p_client->_socket_id+1,&set,NULL,NULL,&tm);
            if(iret < 0){
                Writelog::Trace(2,"select出错:%s",strerror(errno));
                return;
            }
            if(iret == 0){
                Writelog::Trace(2,"select超时");
                return;
            }
            Writelog::Trace(9,"监控到可以进行接收");
            //收取信息
            if(read_sock() == false){
                Writelog::Trace(3,"检测到客户端套接字异常，准备断开连接");
                break;
            }
            ...

很简单，主要流程都在run函数里。这个函数可以优化的地方有几处。比如两个if的判断可以改成if elseif的形式。因为两次if虽然是互斥的但是程序都会判断一次，效率非常低。

另外接收数据的条件可以用FD_ISSET判断是否有数据可读，如果有才真正接收，否则不处理。

所以第一种优化方案很快出炉

void TmsProc::run(){

        tm.tv_sec = 60;
        tm.tv_usec = 0;
    try{

        for(;;){


        FD_ZERO(&set);
        FD_SET(p_client->_socket_id,&set);

            int iret = select(p_client->_socket_id+1,&set,NULL,NULL,&tm);
            if(iret < 0){
                pLog_tmsProc->Trace(2,"select出错:%s",strerror(errno));
                return;
            }
            else if(iret == 0){
                pLog_tmsProc->Trace(2,"select超时");
                return;
            }
            if(FD_ISSET(p_client->_socket_id,&set))
            {

            pLog_tmsProc->Trace(9,"监控到可以进行接收");
              if(read_sock() == false){
                      pLog_tmsProc->Trace(3,"检测到客户端套接字异常，准备断开连接");
                      break;
              }
                   ...

注意到我把超时时间改成了60秒，

tm.tv_sec = 60;

这是我在实际测试时发现，当并发量大时，程序在处理数量多的连接时，前面分配成功的线程会超时退出，看下日志就明白了:

14:39:40][140579076663072]:准备accept
[14:39:40][140579076663072]:接待并分配文件描述符[44],主服务描述符[3]
[14:39:40][140579076663072]:接到连接请求，准备启动线程TCP:0x1e89e90,IP:10.0.0.106,PORT:19803
[14:39:40][140579076663072]:启动服务线程于140577928623872
[14:39:40][140579076663072]:等待接收客户端连接
[14:39:40][140579076663072]:准备accept
[14:39:40][140579076663072]:接待并分配文件描述符[46],主服务描述符[3]
[14:39:40][140579076663072]:接到连接请求，准备启动线程TCP:0x1e8a170,IP:10.0.0.106,PORT:19804
[14:39:40][140577928623872]:接到来自TMS端口的请求
[14:39:40][140577928623872]:业务处理开始
[14:39:40][140577918134016]:接到来自TMS端口的请求
[14:39:40][140577918134016]:业务处理开始
[14:39:40][140579076663072]:启动服务线程于140577918134016
[14:39:40][140579076663072]:等待接收客户端连接
[14:39:40][140579076663072]:准备accept
[14:39:40][140579076663072]:接待并分配文件描述符[48],主服务描述符[3]
[14:39:40][140579076663072]:接到连接请求，准备启动线程TCP:0x1e8a450,IP:10.0.0.106,PORT:19805
[14:39:40][140577500821248]:接到来自TMS端口的请求
[14:39:40][140577500821248]:业务处理开始
[14:39:40][140579076663072]:启动服务线程于140577500821248
[14:39:40][140579076663072]:等待接收客户端连接
[14:39:40][140579076663072]:准备accept
[14:39:41][140579076663072]:接待并分配文件描述符[50],主服务描述符[3]

因为工具是模拟多个客户端同时发起请求，于是就有了上面这样的分配线程的过程，会持续的时间比较长(还要写日志)，也就是同时发生的连接数越多，超时时间就要设置越长。超时改成60秒后。经过工具实测，500连接/500毫秒(应该相当于1000次/秒的并发量了吧？)的处理都正常。

性能大大提高。

但是问题还是很明显，就是超时时间。随着连接数的增大，超时也要一直增大才能保证没有线程”掉队”，但是这个时间太大了会影响真正接收数据时的效率。

第二种优化方案思路来源于apache和nginx的性能差异。

我做过java的web应用，对apache和nginx都有过了解。他俩的一个重要区别是前者基于线程并发，而后者基于进程(fork)。众所周知，nginx很多场景的高并发是好于apache的。

所以我的第二种方案，基本思路是为每个连接fork一个单独的进程处理。独立进程有个最大的好处是不需要加锁了(不解释)。修改好的代码片段如下(我已经把所有带锁的地方都去掉了，这个不贴出来了)。

void run_srv(const char* i_port){
    for(;;)
    {
        client = server.accept_client();
        pWriteLogInstance->Trace(1,"接到连接请求，准备启动进程TCP:%p,IP:%s,PORT:%u",client,client->ip().c_str(),client->port());

        if(client->_socket_id < 0)
        {
            if(errno == EINTR || errno == ECONNABORTED)  
                continue;  
            else  
            {  
                cout << "accept error" << endl;  
                return;  
            }  
        }

        fpid = fork();
        if(fpid < 0)
        {
            pWriteLogInstance->Trace(9,"fork error");
        }
        else if(fpid > 0)//father
        {
            pWriteLogInstance->Trace(1,"father process start");
            client->close();
        }
        else //child
        {
            server.close();
            pWriteLogInstance->Trace(1,"child process start");

            TmsProc* tmpc =new  TmsProc(client);

            tmpc->run();

            delete tmpc;

            if(client != NULL)
            {
                client->close();
                delete client;
                client = NULL;
            }

            exit(-6);
        }

        usleep(10);

void TmsProc::run(){
        while(1)
        {

        //收取信息
            if(read_sock() == false){
                pLog_tmsProc->Trace(3,"检测到客户端套接字异常，准备断开连接");
                send_info.is_bad_qry = true;
                break;
            }

            if(stc_tms.un_parse_size == 0)
            {
                pLog_tmsProc->Trace(3,"没有接受到有效数据，客户端关闭了");
                break;
            }
                   ....

测试结果跟我预想的差不多。效果也是不错的。同样是1000次/秒的并发量数据没有出现问题。而且相比较前一种方案，没有了超时时间的困扰。

第三种方案，我考虑试试游戏行业用的比较多的epoll，这个以前从来没玩过，刚好学习下。

epoll是Linux内核为处理大批量文件描述符而作了改进的poll，是Linux下多路复用IO接口select/poll的增强版本，它能显著提高程序在大量并发连接中只有少量活跃的情况下的系统CPU利用率。另一点原因就是获取事件的时候，它无须遍历整个被侦听的描述符集，只要遍历那些被内核IO事件异步唤醒而加入Ready队列的描述符集合就行了。

为了简单起见，我这里只是用了单线程的epoll，用循环来轮询客户端的socket id来处理多个客户端连接的情况。单线程的epoll号称也能处理1万以上的并发量，我要测试下是不是有这边牛X。

epoll方案的代码如下:

void TmsProc::run(){

        struct epoll_event event;   // 告诉内核要监听什么事件    
        struct epoll_event wait_event[OPEN_MAX]; //内核监听完的结果 

        Socket::TCP server;
       pLog_tmsProc->Trace(9,"准备监听端口:%d",this->port);
       server.listen_on_port(this->port,OPEN_MAX);

       Socket::TCP* client;

       //4.epoll相应参数准备  
        int fd[OPEN_MAX+1];  
        int i = 0, maxi = 0;  
        int number = 0;
        memset(fd,-1, sizeof(fd));  
        fd[0] = server._socket_id;


        pLog_tmsProc->Trace(9,"epoll 开始准备");
        int epfd = epoll_create(OPEN_MAX+1); 
        if( -1 == epfd )
        {    
            pLog_tmsProc->Trace(9,"epoll create error");
            return;    
        }    

        event.data.fd = server._socket_id;     //监听套接字    
        event.events = EPOLLIN; // 表示对应的文件描述符可以读

        //5.事件注册函数，将监听套接字描述符 sockfd 加入监听事件    
        int ret = epoll_ctl(epfd, EPOLL_CTL_ADD, server._socket_id, &event);    
        if(-1 == ret){    
            pLog_tmsProc->Trace(9,"epoll_ctl error");   
            return;    
        }   



        pLog_tmsProc->Trace(9,"业务处理开始");

        while(1)
        {
            // 监视并等待多个文件（标准输入，udp套接字）描述符的属性变化（是否可读）    
            // 没有属性变化，这个函数会阻塞，直到有变化才往下执行，这里没有设置超时    

            pLog_tmsProc->Trace(9,"epoll 开始监听");
            number = epoll_wait(epfd, wait_event, OPEN_MAX, -1);   

            for(int i = 0; i < number; i++)
            {
                if( (wait_event[i].events & EPOLLERR) || ( wait_event[i].events & EPOLLHUP ) || !(wait_event[i].events & EPOLLIN) )
                {
                    pLog_tmsProc->Trace(9,"epoll error");
                    close(wait_event[i].data.fd);
                    continue;
                }
                else if(server._socket_id == wait_event[i].data.fd )  
                {   
                    while(1)
                    {
                        client = server.accept_client();
                        if(client->_socket_id == -1)
                         {
                             if( errno == EAGAIN || errno == EWOULDBLOCK )
                             {
                                 break;
                             }
                             else
                             {
                                 pLog_tmsProc->Trace(9,"accept error");
                                 break;
                             }
                         }

                        Socket::TCP::make_socket_non_blocking(client->_socket_id);

                        event.data.fd = client->_socket_id; //监听套接字    
                        event.events = EPOLLIN | EPOLLERR | EPOLLHUP | EPOLLET; // 表示对应的文件描述符可以读  

                        //6.1.3.事件注册函数，将监听套接字描述符 connfd 加入监听事件 
                        pLog_tmsProc->Trace(9,"为客户端注册epoll监听"); 
                        ret = epoll_ctl(epfd, EPOLL_CTL_ADD, client->_socket_id, &event);

                        if(ret < 0){    
                            pLog_tmsProc->Trace(9,"epoll_ctl error");     
                        }   
                        event.data.fd = client->_socket_id;
                    }
                }
                else
                {
                    //收取信息
                   if(read_sock(wait_event[i].data.fd) == false){
                        pLog_tmsProc->Trace(3,"检测到客户端套接字异常，准备断开连接");
                        close(wait_event[i].data.fd); 
                   }
                   ...

只能说epoll确实比较给力，我这只是个单线程的服务，用工具测试上述代码，500个并发也是妥妥的。