基于libuv的TCP设计（一）
本人一直在寻找一个跨平台的网络库，boost与ACE比较庞大，不考虑。对比了libevent,libev,libuv后，最终选择了libuv.可libuv文档少，例子也简单，对于tcp只有个echo-server的例子。网上也找过对其封装的例子，如下
libsourcey库，封装了许多库。对libuv的封装跟其他代码耦合比较紧，难为剥离 http://sourcey.com/libuv-cpp-wrappers/
C++11封装的，可惜VS10未完全支持C++11 https://github.com/larroy/uvpp
C++封装的 https://github.com/keepallsimple/uvpp
本人想实现一个raw tcp server,支持上万链接数的，网上找到的都没合适我的，没办法只能参照各例子自己封装了。
头文件
|
/*************************************** * @file tcpsocket.h * @brief 基于libuv封装的tcp服务器与客户端,使用log4z作日志工具 * @details * @author phata, wqvbjhc@gmail.com * @date 2014-5-13 * @mod 2014-5-13 phata 修正服务器与客户端的错误.现服务器支持多客户端连接 修改客户端测试代码，支持并发多客户端测试 ****************************************/ #ifndef #define #include #include #include #include #define namespace { typedef typedef typedef class class { public: clientdata(int client_handle = (uv_tcp_t*)malloc(sizeof(*client_handle)); client_handle->data = this; readbuffer = uv_buf_init((char*)malloc(BUFFERSIZE), BUFFERSIZE); writebuffer = uv_buf_init((char*)malloc(BUFFERSIZE), BUFFERSIZE); } virtual ~clientdata() { free(readbuffer.base); readbuffer.base = nullptr; readbuffer.len = 0; free(writebuffer.base); writebuffer.base = nullptr; writebuffer.len = 0; free(client_handle); client_handle = nullptr; } int uv_tcp_t* client_handle;//客户端句柄 TCPServer* tcp_server;//服务器句柄(保存是因为某些回调函数需要到) uv_buf_t uv_buf_t uv_write_t server_recvcb }; class { public: TCPServer(uv_loop_t* loop = uv_default_loop()); virtual ~TCPServer(); static public: //基本函数 bool bool void bool bool const return }; virtual virtual virtual protected: int bool //静态回调函数 static static static static static static private: bool bool bool bool bool uv_tcp_t std::map<int,clientdata*> clients_list_;//子客户端链接 uv_mutex_t uv_loop_t *loop_; std::string newconnect bool }; class { //直接调用connect/connect6会进行连接 public: TCPClient(uv_loop_t* loop = uv_default_loop()); virtual ~TCPClient(); static public: //基本函数 virtual virtual virtual virtual void //是否启用Nagle算法 bool bool const return }; protected: //静态回调函数 static static static static static static static bool bool private: enum { CONNECT_TIMEOUT, CONNECT_FINISH, CONNECT_ERROR, CONNECT_DIS, }; uv_tcp_t uv_loop_t *loop_; uv_write_t uv_connect_t uv_thread_t std::string uv_buf_t uv_buf_t uv_mutex_t int client_recvcb void* userdata_;//回调函数的用户数据 std::string int bool }; } |
#endif
// TCPSocket_H
源文件
|
#include #include std::string { std::string err = uv_err_name(retcode); err +=":"; err += uv_strerror(retcode); return } namespace { /*****************************************TCP Server*************************************************************/ TCPServer::TCPServer(uv_loop_t* loop) :newconcb_(nullptr), isinit_(false) { loop_ = loop; } TCPServer::~TCPServer() { close(); LOGI("tcp server exit."); } //初始化与关闭--服务器与客户端一致 bool { if (isinit_) { return } if (!loop_) { errmsg_ = "loop is null on tcp init."; LOGE(errmsg_); return } int if (iret) { errmsg_ = GetUVError(iret); LOGE(errmsg_); return } iret = uv_tcp_init(loop_,&server_); if (iret) { errmsg_ = GetUVError(iret); LOGE(errmsg_); return } isinit_ = true; server_.data = this; //iret = uv_tcp_keepalive(&server_, 1, 60);//调用此函数后后续函数会调用出错 //if (iret) { // errmsg_ = GetUVError(iret); // return false; //} return } void { for (auto auto uv_close((uv_handle_t*)data->client_handle,AfterClientClose); } clients_list_.clear(); LOGI("close server"); if (isinit_) { uv_close((uv_handle_t*) &server_, AfterServerClose); LOGI("close server"); } isinit_ = false; uv_mutex_destroy(&mutex_handle_); } bool { LOGI("server runing."); int if (iret) { errmsg_ = GetUVError(iret); LOGE(errmsg_); return } return } //属性设置--服务器与客户端一致 bool { int if (iret) { errmsg_ = GetUVError(iret); LOGE(errmsg_); return } return } bool { int if (iret) { errmsg_ = GetUVError(iret); LOGE(errmsg_); return } return } //作为server时的函数 bool { struct int if (iret) { errmsg_ = GetUVError(iret); LOGE(errmsg_); return } iret = uv_tcp_bind(&server_, (const if (iret) { errmsg_ = GetUVError(iret); LOGE(errmsg_); return } LOGI("server bind ip="<<ip<<", port="<<port); return } bool { struct int if (iret) { errmsg_ = GetUVError(iret); LOGE(errmsg_); return } iret = uv_tcp_bind(&server_, (const if (iret) { errmsg_ = GetUVError(iret); LOGE(errmsg_); return } LOGI("server bind ip="<<ip<<", port="<<port); return } bool { int if (iret) { errmsg_ = GetUVError(iret); LOGE(errmsg_); return } LOGI("server listen"); return } bool { close(); if (!init()) { return } if (!bind(ip,port)) { return } if (!listen(SOMAXCONN)) { return } if (!run()) { return } LOGI("start listen "<<ip<<": "<<port); return } bool { close(); if (!init()) { return } if (!bind6(ip,port)) { return } if (!listen(SOMAXCONN)) { return } if (!run()) { return } return } //服务器发送函数 int { auto if (itfind == clients_list_.end()) { errmsg_ = "can't find cliendid "; errmsg_ += std::to_string((long LOGE(errmsg_); return -1; } //自己控制data的生命周期直到write结束 if (itfind->second->writebuffer.len < len) { itfind->second->writebuffer.base = (char*)realloc(itfind->second->writebuffer.base,len); itfind->second->writebuffer.len = len; } memcpy(itfind->second->writebuffer.base,data,len); uv_buf_t int if (iret) { errmsg_ = GetUVError(iret); LOGE(errmsg_); return } return } //服务器-新客户端函数 void { if (!server->data) { return; } TCPServer *tcpsock = (TCPServer *)server->data; int clientdata* cdata = new cdata->tcp_server = tcpsock;//保存服务器的信息 int if (iret) { delete tcpsock->errmsg_ = GetUVError(iret); LOGE(tcpsock->errmsg_); return; } iret = uv_accept((uv_stream_t*)&tcpsock->server_, (uv_stream_t*) cdata->client_handle); if ( iret) { tcpsock->errmsg_ = GetUVError(iret); uv_close((uv_handle_t*) cdata->client_handle, NULL); delete LOGE(tcpsock->errmsg_); return; } tcpsock->clients_list_.insert(std::make_pair(clientid,cdata));//加入到链接队列 if (tcpsock->newconcb_) { tcpsock->newconcb_(clientid); } LOGI("new client("<<cdata->client_handle<<") id="<< clientid); iret = uv_read_start((uv_stream_t*)cdata->client_handle, onAllocBuffer, AfterServerRecv);//服务器开始接收客户端的数据 return; } //服务器-接收数据回调函数 void { auto if (itfind != clients_list_.end()) { itfind->second->recvcb_ = cb; } } //服务器-新链接回调函数 void { newconcb_ = cb; } //服务器分析空间函数 void { if (!handle->data) { return; } clientdata *client = (clientdata*)handle->data; *buf = client->readbuffer; } void { if (!handle->data) { return; } clientdata *client = (clientdata*)handle->data;//服务器的recv带的是clientdata if (nread < 0) {/* Error or EOF */ TCPServer *server = (TCPServer *)client->tcp_server; if (nread == UV_EOF) { fprintf(stdout,"客户端(%d)连接断开，关闭此客户端\n",client->client_id); LOGW("客户端("<<client->client_id<<")主动断开"); } else fprintf(stdout,"客户端(%d)异常断开\n",client->client_id); LOGW("客户端("<<client->client_id<<")异常断开"); } else { fprintf(stdout,"%s\n",GetUVError(nread)); LOGW("客户端("<<client->client_id<<")异常断开："<<GetUVError(nread)); } server->DeleteClient(client->client_id);//连接断开，关闭客户端 return; } else } else client->recvcb_(client->client_id,buf->base,nread); } } //服务器与客户端一致 void { if (status < 0) { LOGE("发送数据有误:"<<GetUVError(status)); fprintf(stderr, "Write error %s\n", GetUVError(status)); } } void { //服务器,不需要做什么 } void { clientdata *cdata = (clientdata*)handle->data; LOGI("client "<<cdata->client_id<<" had closed."); delete } int { static return ++s_id; } bool { uv_mutex_lock(&mutex_handle_); auto if (itfind == clients_list_.end()) { errmsg_ = "can't find client "; errmsg_ += std::to_string((long LOGE(errmsg_); uv_mutex_unlock(&mutex_handle_); return } if (uv_is_active((uv_handle_t*)itfind->second->client_handle)) { uv_read_stop((uv_stream_t*)itfind->second->client_handle); } uv_close((uv_handle_t*)itfind->second->client_handle,AfterClientClose); clients_list_.erase(itfind); LOGI("删除客户端"<<clientid); uv_mutex_unlock(&mutex_handle_); return } void { zsummer::log4z::ILog4zManager::GetInstance()->SetLoggerMonthdir(LOG4Z_MAIN_LOGGER_ID, true); zsummer::log4z::ILog4zManager::GetInstance()->SetLoggerDisplay(LOG4Z_MAIN_LOGGER_ID,false); zsummer::log4z::ILog4zManager::GetInstance()->SetLoggerLevel(LOG4Z_MAIN_LOGGER_ID,LOG_LEVEL_DEBUG); zsummer::log4z::ILog4zManager::GetInstance()->SetLoggerLimitSize(LOG4Z_MAIN_LOGGER_ID,100); if (logpath) { zsummer::log4z::ILog4zManager::GetInstance()->SetLoggerPath(LOG4Z_MAIN_LOGGER_ID,logpath); } zsummer::log4z::ILog4zManager::GetInstance()->Start(); } /*****************************************TCP Client*************************************************************/ TCPClient::TCPClient(uv_loop_t* loop) :recvcb_(nullptr),userdata_(nullptr) ,connectstatus_(CONNECT_DIS) , isinit_(false) { readbuffer_ = uv_buf_init((char*) malloc(BUFFERSIZE), BUFFERSIZE); writebuffer_ = uv_buf_init((char*) malloc(BUFFERSIZE), BUFFERSIZE); loop_ = loop; connect_req_.data = this; write_req_.data = this; } TCPClient::~TCPClient() { free(readbuffer_.base); readbuffer_.base = nullptr; readbuffer_.len = 0; free(writebuffer_.base); writebuffer_.base = nullptr; writebuffer_.len = 0; close(); LOGI("客户端("<<this<<")退出"); } //初始化与关闭--服务器与客户端一致 bool { if (isinit_) { return } if (!loop_) { errmsg_ = "loop is null on tcp init."; LOGE(errmsg_); return } int if (iret) { errmsg_ = GetUVError(iret); LOGE(errmsg_); return } iret = uv_mutex_init(&write_mutex_handle_); if (iret) { errmsg_ = GetUVError(iret); LOGE(errmsg_); return } isinit_ = true; fprintf(stdout,"客户端(%p) init type = %d\n",&client_,client_.type); client_.data = this; //iret = uv_tcp_keepalive(&client_, 1, 60);// //if (iret) { // errmsg_ = GetUVError(iret); // return false; //} LOGI("客户端("<<this<<")Init"); return } void { if (!isinit_) { return; } uv_mutex_destroy(&write_mutex_handle_); uv_close((uv_handle_t*) &client_, AfterClose); LOGI("客户端("<<this<<")close"); isinit_ = false; } bool { LOGI("客户端("<<this<<")run"); int if (iret) { errmsg_ = GetUVError(iret); LOGE(errmsg_); return } return } //属性设置--服务器与客户端一致 bool { //http://blog.csdn.net/u011133100/article/details/21485983 int if (iret) { errmsg_ = GetUVError(iret); LOGE(errmsg_); return } return } bool { int if (iret) { errmsg_ = GetUVError(iret); LOGE(errmsg_); return } return } //作为client的connect函数 bool { close(); init(); connectip_ = ip; connectport_ = port; LOGI("客户端("<<this<<")start connect to server("<<ip<<":"<<port<<")"); int if (iret) { errmsg_ = GetUVError(iret); LOGE(errmsg_); return } while ( connectstatus_ == CONNECT_DIS) { #if Sleep(100); #else usleep((100) * 1000) #endif } return } bool { close(); init(); connectip_ = ip; connectport_ = port; LOGI("客户端("<<this<<")start connect to server("<<ip<<":"<<port<<")"); int if (iret) { errmsg_ = GetUVError(iret); LOGE(errmsg_); return } while ( connectstatus_ == CONNECT_DIS) { //fprintf(stdout,"client(%p) wait, connect status %d\n",this,connectstatus_); #if Sleep(100); #else usleep((100) * 1000) #endif } return } void { TCPClient *pclient = (TCPClient*)arg; fprintf(stdout,"client(%p) ConnectThread start\n",pclient); struct int if (iret) { pclient->errmsg_ = GetUVError(iret); LOGE(pclient->errmsg_); return; } iret = uv_tcp_connect(&pclient->connect_req_, &pclient->client_, (const if (iret) { pclient->errmsg_ = GetUVError(iret); LOGE(pclient->errmsg_); return; } fprintf(stdout,"client(%p) ConnectThread end, connect status %d\n",pclient, pclient->connectstatus_); pclient->run(); } void { TCPClient *pclient = (TCPClient*)arg; LOGI("客户端("<<pclient<<")Enter Connect Thread."); fprintf(stdout,"client(%p) ConnectThread start\n",pclient); struct int if (iret) { pclient->errmsg_ = GetUVError(iret); LOGE(pclient->errmsg_); return; } iret = uv_tcp_connect(&pclient->connect_req_, &pclient->client_, (const if (iret) { pclient->errmsg_ = GetUVError(iret); LOGE(pclient->errmsg_); return; } fprintf(stdout,"client(%p) ConnectThread end, connect status %d\n",pclient, pclient->connectstatus_); LOGI("客户端("<<pclient<<")Leave Connect Thread. connect status "<<pclient->connectstatus_); pclient->run(); } void { fprintf(stdout,"start after connect\n"); TCPClient *pclient = (TCPClient*)handle->handle->data; if (status) { pclient->connectstatus_ = CONNECT_ERROR; fprintf(stdout,"connect error:%s\n",GetUVError(status)); return; } int if (iret) { fprintf(stdout,"uv_read_start error:%s\n",GetUVError(iret)); pclient->connectstatus_ = CONNECT_ERROR; } else { pclient->connectstatus_ = CONNECT_FINISH; } LOGI("客户端("<<pclient<<")run"); fprintf(stdout,"end after connect\n"); } //客户端的发送函数 int { //自己控制data的生命周期直到write结束 if (!data || len <= 0) { errmsg_ = "send data is null or len less than zero."; return 0; } uv_mutex_lock(&write_mutex_handle_); if (writebuffer_.len < len) { writebuffer_.base = (char*)realloc(writebuffer_.base,len); writebuffer_.len = len; } memcpy(writebuffer_.base,data,len); uv_buf_t int if (iret) { uv_mutex_unlock(&write_mutex_handle_); errmsg_ = GetUVError(iret); LOGE(errmsg_); return } return } //客户端-接收数据回调函数 void { recvcb_ = cb; userdata_ = userdata; } //客户端分析空间函数 void { if (!handle->data) { return; } TCPClient *client = (TCPClient*)handle->data; *buf = client->readbuffer_; } void { if (!handle->data) { return; } TCPClient *client = (TCPClient*)handle->data;//服务器的recv带的是TCPClient if (nread < 0) { if (nread == UV_EOF) { fprintf(stdout,"服务器(%p)主动断开\n",handle); LOGW("服务器主动断开"); } else fprintf(stdout,"服务器(%p)异常断开\n",handle); LOGW("服务器异常断开"); } else { fprintf(stdout,"服务器(%p)异常断开:%s\n",handle,GetUVError(nread)); LOGW("服务器异常断开"<<GetUVError(nread)); } uv_close((uv_handle_t*)handle, AfterClose); return; } if (nread > 0 && client->recvcb_) { client->recvcb_(buf->base,nread,client->userdata_); } } //服务器与客户端一致 void { TCPClient *client = (TCPClient *)req->handle->data; uv_mutex_unlock(&client->write_mutex_handle_); if (status < 0) { LOGE("发送数据有误:"<<GetUVError(status)); fprintf(stderr, "Write error %s\n", GetUVError(status)); } } //服务器与客户端一致 void { fprintf(stdout,"客户端(%p)已close\n",handle); LOGI("客户端("<<handle<<")已close"); } void { zsummer::log4z::ILog4zManager::GetInstance()->SetLoggerMonthdir(LOG4Z_MAIN_LOGGER_ID, true); zsummer::log4z::ILog4zManager::GetInstance()->SetLoggerDisplay(LOG4Z_MAIN_LOGGER_ID,false); zsummer::log4z::ILog4zManager::GetInstance()->SetLoggerLevel(LOG4Z_MAIN_LOGGER_ID,LOG_LEVEL_DEBUG); zsummer::log4z::ILog4zManager::GetInstance()->SetLoggerLimitSize(LOG4Z_MAIN_LOGGER_ID,100); if (logpath) { zsummer::log4z::ILog4zManager::GetInstance()->SetLoggerPath(LOG4Z_MAIN_LOGGER_ID,logpath); } zsummer::log4z::ILog4zManager::GetInstance()->Start(); } } |
代码已上传到git: https://github.com/wqvbjhc/libuv_tcp
多路，超过uv_write就会assert出错，未找到原因
服务器可以接收几十路连接。万百上千路未测试过，因为没有模拟环境。
最新文章
- 关于Oracle报“ORA-12514 TNS 监听程序当前无法识别连接描述符中请求服务”错误
- 9.13 JS循环
- 解决Odoo日期(时间)无效的问题 [转]
- poj2993 翻转2996
- CodeForces 185A 快速幂
- 使用windows远程桌面连接Windows Azure中的Ubuntu虚拟机
- StopWatch
- leetcode Longest Palindromic Substring python
- poj 1077 Eight(双向bfs)
- js-tab选项卡
- Ionic3学习笔记（四）修改返回按钮文字、颜色
- jBPM学习之部署流程定义
- unit test
- sqlserver 导入数据出现 无法创建 OLE DB 取值函数。请查看列元数据是否有效
- JS 判断滚动底部并加载更多效果。。。。。。。。。
- 算法：快速排序（Quick Sort）
- 23、java 通过System.getProperties()获取系统参数
- 代码处理 iOS 的横竖屏旋转
- ssh自动分发密匙脚本样板
- C#时间的味道——任时光匆匆我只在乎你