server参考是别人的代码
#include#include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #define MAXBUF 1024#define MAXEPOLLSIZE 10000/* setnonblocking - 设置句柄为非堵塞方式 */int setnonblocking(int sockfd){ if (fcntl(sockfd, F_SETFL, fcntl(sockfd, F_GETFD, 0)|O_NONBLOCK) == -1) { return -1; } return 0;}/* handle_message - 处理每一个 socket 上的消息收发 */int handle_message(int new_fd){ char buf[MAXBUF + 1]; int len; /* 開始处理每一个新连接上的数据收发 */ bzero(buf, MAXBUF + 1); /* 接收client的消息 */ len = recv(new_fd, buf, MAXBUF, 0); if (len > 0) { printf("%d接收消息成功:'%s'。共%d个字节的数据\n", new_fd, buf, len); } else { if (len < 0) printf ("消息接收失败！错误代码是%d。错误信息是'%s'\n", errno, strerror(errno)); close(new_fd); return -1; } /* 处理每一个新连接上的数据收发结束 */ return len;}int main(){ int listener, new_fd, kdpfd, nfds, n, ret, curfds; socklen_t len; struct sockaddr_in my_addr, their_addr; unsigned int myport, lisnum; struct epoll_event ev; struct epoll_event events[MAXEPOLLSIZE]; struct rlimit rt; myport = 6000; lisnum = 2; /* 设置每一个进程同意打开的最大文件数 */ rt.rlim_max = rt.rlim_cur = MAXEPOLLSIZE; if (setrlimit(RLIMIT_NOFILE, &rt) == -1) { perror("setrlimit"); exit(1); } else { printf("设置系统资源參数成功！
\n"); } /* 开启 socket 监听 */ if ((listener = socket(PF_INET, SOCK_STREAM, 0)) == -1) { perror("socket"); exit(1); } else { printf("socket 创建成功！\n"); } setnonblocking(listener); bzero(&my_addr, sizeof(my_addr)); my_addr.sin_family = PF_INET; my_addr.sin_port = htons(myport); my_addr.sin_addr.s_addr = INADDR_ANY; if (bind(listener, (struct sockaddr *) &my_addr, sizeof(struct sockaddr)) == -1) { perror("bind"); exit(1); } else { printf("IP 地址和端口绑定成功\n"); } if (listen(listener, lisnum) == -1) { perror("listen"); exit(1); } else { printf("开启服务成功！
\n"); } /* 创建 epoll 句柄，把监听 socket 增加到 epoll 集合里 */ kdpfd = epoll_create(MAXEPOLLSIZE); len = sizeof(struct sockaddr_in); ev.events = EPOLLIN | EPOLLET; ev.data.fd = listener; if (epoll_ctl(kdpfd, EPOLL_CTL_ADD, listener, &ev) < 0) { fprintf(stderr, "epoll set insertion error: fd=%d\n", listener); return -1; } else { printf("监听 socket 增加 epoll 成功！\n"); } curfds = 1; while (1) { /* 等待有事件发生 */ nfds = epoll_wait(kdpfd, events, curfds, -1); if (nfds == -1) { perror("epoll_wait"); break; } /* 处理全部事件 */ for (n = 0; n < nfds; ++n) { if (events[n].data.fd == listener) { new_fd = accept(listener, (struct sockaddr *) &their_addr,&len); if (new_fd < 0) { perror("accept"); continue; } else { printf("有连接来自于： %s:%d， 分配的 socket 为:%d\n", inet_ntoa(their_addr.sin_addr), ntohs(their_addr.sin_port), new_fd); } setnonblocking(new_fd); ev.events = EPOLLIN | EPOLLET; ev.data.fd = new_fd; if (epoll_ctl(kdpfd, EPOLL_CTL_ADD, new_fd, &ev) < 0) { fprintf(stderr, "把 socket '%d' 增加 epoll 失败。%s\n", new_fd, strerror(errno)); return -1; } curfds++; } else { ret = handle_message(events[n].data.fd); if (ret < 1 && errno != 11) { epoll_ctl(kdpfd, EPOLL_CTL_DEL, events[n].data.fd,&ev); curfds--; } } } } close(listener); return 0; }
client代码
// echo_client.cpp// g++ -o echo_client -O3 echo_client.cpp -lboost_system -lboost_thread#includenamespace asio = boost::asio;using asio::ip::tcp;#include #include #include #include #include #include FILE* g_hLog = 0;class session : public boost::enable_shared_from_this {public: session(asio::io_service& io) : socket_(io) { } tcp::socket& socket() { return socket_; } void start(int nindex) { m_nIndex = nindex; snprintf(output_buffer_, sizeof(output_buffer_), "this is %d client.", m_nIndex); //fprintf(g_hLog, "%s\n", output_buffer_); asio::async_write(socket_, asio::buffer(output_buffer_, 12), boost::bind(&session::handle_write, shared_from_this(), _1, _2)); } void handle_write(const boost::system::error_code& ec, std::size_t bytes_transfered) { fprintf(g_hLog, "%d write %s\n", m_nIndex, ec.message().data()); if(!ec) { //asio::async_read(socket_, asio::buffer(input_buffer_, 12), boost::bind(&session::handle_read, shared_from_this(), _1, _2)); } else { std::cerr << "write error:" << ec.message() << std:: endl; } } void handle_read(const boost::system::error_code& ec, std::size_t bytes_transfered) { if(ec) { std::cerr << "read error:" << ec.message() << std::endl; } }private: tcp::socket socket_; int m_nIndex; char output_buffer_[50]; char input_buffer_[50];};void handle_connect(boost::shared_ptr session_ptr, int nindex, const boost::system::error_code& ec){ if(ec) { fprintf(g_hLog, "%d connect error: %s\n", nindex, ec.message().data()); std::cerr << "connect error:" << ec.message() << std::endl; } else { session_ptr->start(nindex); }}int main(/*int argc, char* argv[]*/){ g_hLog = fopen("log.txt", "wr"); if(!g_hLog) return 0; struct timeval time1, time2; gettimeofday(&time1, 0); asio::io_service io; tcp::resolver resolver(io); tcp::resolver::iterator endpoint = resolver.resolve(tcp::resolver::query("localhost", "6000")); boost::shared_ptr session_ptr; for(int i = 0; i < 10000; i++) { session_ptr.reset(new session(io)); asio::async_connect(session_ptr->socket(), endpoint, boost::bind(handle_connect, session_ptr, i, _1)); } io.run(); gettimeofday(&time2, 0); std::cout << "has time:" << double((time2.tv_sec-time1.tv_sec)+(time2.tv_usec-time1.tv_usec)/1000000.0) << std::endl; fprintf(g_hLog, "has time: %f s", double((time2.tv_sec-time1.tv_sec)+(time2.tv_usec-time1.tv_usec)/1000000.0)); fclose(g_hLog);}
Linux默认最大的文件句柄数是1024，须要改动此值才干正常測试。打开终端运行例如以下命令
echo ulimit -n 65535 >> /etc/profile注销账户并又一次登录能够。
版权声明：本文博主原创文章。博客，未经同意不得转载。