用一个示例讲解我如何一步步实现高并发服务（基于C++）

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

运行环境是centOS 6.5。

刚开始客户端数量少而且访问不频繁&＃xff0c;所以没太关注并发的问题。当时用工具测试大概只能支持的几十 TPS的并发访问&＃xff0c;再加大并发就会有数据串包的情况出现了。下面是我的优化过程。

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

我选择的测试工具是一个tcp客户端工具&＃xff0c;可以快捷的进行多客户端连接的测试。

线程安全的单例模式

版本1.0是通过多线程实现高并发的。我的工程里有两个类是单例模式&＃xff0c;一个参数文件管理类&＃xff0c;一个是日志管理类。一开始我没有考虑线程安全&＃xff0c;于是第一步我先把这两个类改成线程安全测试看看效果。

增加线程安全前的代码片段&＃xff08;只给出参数文件管理类的实现&＃xff09;

//.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; //定义一个静态成员变量&＃xff0c;程序结束时&＃xff0c;系统会自动调用它的析构函数
...

//.cpp
AppCof* AppCof::open_cof(){if(m_pInstance &＃61;&＃61; NULL){m_pInstance &＃61; 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 &＃61; new AppCof();if(m_pInstance !&＃61; NULL){m_pInstance->get_env();m_pInstance->read_cof();}
}AppCof* AppCof::open_cof(){pthread_once(&ponce_, &AppCof::init);return m_pInstance;
}...

这里有两个重点&＃xff0c;一是pthread_once的用法&＃xff0c;还有就是boost::noncopyable。

先说说前者&＃xff0c;

int pthread_once(pthread_once_t *once_control, void (*init_routine) (void))&＃xff1b;


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

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

优化IO读写机制

这部分从select,epoll这些IO处理上下手。我用三个方案分别测试&＃xff0c;效果还是比较明显。

版本1.0是这样的:

void run_srv(const char* i_port){for(;;){client &＃61; 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 &＃61; i_client;i_client &＃61; NULL;TmsProc* tmpc &＃61;new TmsProc(client);tmpc->run();delete tmpc;return ;
}

void TmsProc::run(){Writelog::Trace(9,"业务处理开始");try{for(;;){tm.tv_sec &＃61; 3;tm.tv_usec &＃61; 0;FD_ZERO(&set);FD_SET(p_client->_socket_id,&set);int iret &＃61; select(p_client->_socket_id&＃43;1,&set,NULL,NULL,&tm);if(iret < 0){Writelog::Trace(2,"select出错:%s",strerror(errno));return;}if(iret &＃61;&＃61; 0){Writelog::Trace(2,"select超时");return;}Writelog::Trace(9,"监控到可以进行接收");//收取信息if(read_sock() &＃61;&＃61; false){Writelog::Trace(3,"检测到客户端套接字异常&＃xff0c;准备断开连接");break;}...

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

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

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


void TmsProc::run(){tm.tv_sec &＃61; 60;tm.tv_usec &＃61; 0;try{for(;;){FD_ZERO(&set);FD_SET(p_client->_socket_id,&set);int iret &＃61; select(p_client->_socket_id&＃43;1,&set,NULL,NULL,&tm);if(iret < 0){pLog_tmsProc->Trace(2,"select出错:%s",strerror(errno));return;}else if(iret &＃61;&＃61; 0){pLog_tmsProc->Trace(2,"select超时");return;}if(FD_ISSET(p_client->_socket_id,&set)){pLog_tmsProc->Trace(9,"监控到可以进行接收");if(read_sock() &＃61;&＃61; false){pLog_tmsProc->Trace(3,"检测到客户端套接字异常&＃xff0c;准备断开连接");break;}...


注意到我把超时时间改成了60秒&＃xff0c;

tm.tv_sec &＃61; 60;


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

14:39:40][140579076663072]:准备accept
[14:39:40][140579076663072]:接待并分配文件描述符[44],主服务描述符[3]
[14:39:40][140579076663072]:接到连接请求&＃xff0c;准备启动线程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]:接到连接请求&＃xff0c;准备启动线程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]:接到连接请求&＃xff0c;准备启动线程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]

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

性能大大提高。

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

第二种优化方案思路来源于apache和nginx。

apache和nginx他俩的一个重要区别是前者基于多线程实现高并发&＃xff0c;而后者基于多进程(fork)。

而众所周知&＃xff0c;nginx很多场景的高并发是好于apache的。

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


void run_srv(const char* i_port){for(;;){client &＃61; server.accept_client();pWriteLogInstance->Trace(1,"接到连接请求&＃xff0c;准备启动进程TCP:%p,IP:%s,PORT:%u",client,client->ip().c_str(),client->port());if(client->_socket_id < 0){if(errno &＃61;&＃61; EINTR || errno &＃61;&＃61; ECONNABORTED) continue; else { cout << "accept error" << endl; return; } }fpid &＃61; 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 &＃61;new TmsProc(client);tmpc->run();delete tmpc;if(client !&＃61; NULL){client->close();delete client;client &＃61; NULL;}exit(-6);}usleep(10);

void TmsProc::run(){while(1){//收取信息if(read_sock() &＃61;&＃61; false){pLog_tmsProc->Trace(3,"检测到客户端套接字异常&＃xff0c;准备断开连接");send_info.is_bad_qry &＃61; true;break;}if(stc_tms.un_parse_size &＃61;&＃61; 0){pLog_tmsProc->Trace(3,"没有接受到有效数据&＃xff0c;客户端关闭了");break;}....

测试结果来看效果也是不错的&＃xff0c;而且相比较前一种方案&＃xff0c;没有了超时时间的问题。

第三种方案&＃xff0c;我考虑试试 IO 处理中的王者&＃xff0c;epoll。

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

为了简单起见&＃xff0c;我这里只是用了单线程的epoll&＃xff0c;用循环来轮询客户端的socket id来处理多个客户端连接的情况。单线程的epoll号称也能处理1万以上的并发量&＃xff0c;我要测试下是不是有这边牛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&＃43;1]; int i &＃61; 0, maxi &＃61; 0; int number &＃61; 0;memset(fd,-1, sizeof(fd)); fd[0] &＃61; server._socket_id;pLog_tmsProc->Trace(9,"epoll 开始准备");int epfd &＃61; epoll_create(OPEN_MAX&＃43;1); if( -1 &＃61;&＃61; epfd ){ pLog_tmsProc->Trace(9,"epoll create error");return; } event.data.fd &＃61; server._socket_id; //监听套接字 event.events &＃61; EPOLLIN; // 表示对应的文件描述符可以读//5.事件注册函数&＃xff0c;将监听套接字描述符 sockfd 加入监听事件 int ret &＃61; epoll_ctl(epfd, EPOLL_CTL_ADD, server._socket_id, &event); if(-1 &＃61;&＃61; ret){ pLog_tmsProc->Trace(9,"epoll_ctl error"); return; } pLog_tmsProc->Trace(9,"业务处理开始");while(1){// 监视并等待多个文件&＃xff08;标准输入&＃xff0c;udp套接字&＃xff09;描述符的属性变化&＃xff08;是否可读&＃xff09; // 没有属性变化&＃xff0c;这个函数会阻塞&＃xff0c;直到有变化才往下执行&＃xff0c;这里没有设置超时 pLog_tmsProc->Trace(9,"epoll 开始监听");number &＃61; epoll_wait(epfd, wait_event, OPEN_MAX, -1); for(int i &＃61; 0; i < number; i&＃43;&＃43;){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 &＃61;&＃61; wait_event[i].data.fd ) { while(1){client &＃61; server.accept_client();if(client->_socket_id &＃61;&＃61; -1){if( errno &＃61;&＃61; EAGAIN || errno &＃61;&＃61; EWOULDBLOCK ){break;}else{pLog_tmsProc->Trace(9,"accept error");break;}}Socket::TCP::make_socket_non_blocking(client->_socket_id);event.data.fd &＃61; client->_socket_id; //监听套接字 event.events &＃61; EPOLLIN | EPOLLERR | EPOLLHUP | EPOLLET; // 表示对应的文件描述符可以读 //6.1.3.事件注册函数&＃xff0c;将监听套接字描述符 connfd 加入监听事件 pLog_tmsProc->Trace(9,"为客户端注册epoll监听"); ret &＃61; epoll_ctl(epfd, EPOLL_CTL_ADD, client->_socket_id, &event);if(ret < 0){ pLog_tmsProc->Trace(9,"epoll_ctl error"); } event.data.fd &＃61; client->_socket_id;}}else{//收取信息if(read_sock(wait_event[i].data.fd) &＃61;&＃61; false){pLog_tmsProc->Trace(3,"检测到客户端套接字异常&＃xff0c;准备断开连接");close(wait_event[i].data.fd); }...

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