支持百万并发和实现百万并发不是一个意思,能支持是实现的前提,本篇带你在配置较差的情况下,实现一个支持百万并发的服务器连接测试。
原理:基于epoll的Reactor模式。绝大多数的开源服务器框架,底层都是使用的epoll,Reactor是一种高效事件处理模式。
实验条件:准备四台虚拟机,一台当作服务器使用,另外三台当作客户端不断请求连接。如果你的实验环境和我一样,本机配置较差,可以只用两台虚拟机,一台做服务器,一台做客户机,客户机只需达到340000的连接,即可表明,三台客户机就可以达到上百万的并发连接。
在开始之前,我们先了解一下socket
socket的英文名是插座的意思,它是客户端和服务器建立通信连接的唯一标识,为何说它唯一,因为它是由五元组(源端口,目的端口,源ip,目的ip,通信协议)构成的,我们要知道,目的端口,源ip和目的ip都是确定的,这里唯一可以改变的就是源端口,我们都知道总的端口号也就是65535个,而且在一个进程中,默认单个进程能使用的端口号大概是28000(可以自己设置,但并没有解决实际问题),所以就目前的情况来看单个端口是不行的。
实验过程中会遇到的问题:
1.一个进程的默认支持打开文件数为1024,有的是65535,但这显然是不够的,可以通过ulimit -a 来看到,我们可以看到表示最大文件描述符打开数的对应条目是 -n ,我们可以通过ulimit -n 数字 来修改最大打开文件描述符数量,其他数值同理。
这里我们先修改成ulimit -n 1024576,足够一百万使用了。
当然这种方式修改后重启后就没有了
可以在 /etc/sysctl.conf 下添加如下对fs.file-max值得修改,再往下一行表示得是tcp协议栈占用内存大小的现在,最后两行表示的是套接字发送和接收缓冲区的大小,可以看出每项都对应3个值,中间表示默认值,左边是最小值,右边是最大值。
2.若只使用一个端口,那么最大连接数目在28000左右,当然我们可以自己设置,但意义不大,没有解决实际问题,这里我们可以使用监听多个端口的方式,当然,在实际开发中,我们不会使用监听多端口的方式,而是使用多进程,这里有朋友可能要问,为什么是多进程而不是多线程,因为多线程的本质是同一个进程,没有扩大可打开的文件描述符数量,而每个进程拥有它独立的文件系统空间,所以我们需要使用多进程。
3. 客户端在连接进行到65535左右的时候还会出现问题,这里是由于,客户端本身对于一个进程的connect连接数量有限制,当到达一定的限制后,就无法发送出connect连接的消息,这里我们需要在 /etc/sysctl.conf 下添加如下内容来接触限制。
到此为止就可以完成测试了,需要注意的一点是做服务器的那台虚拟机,内存必须设置的大一点,因为每一个socket都含有发送和接收缓冲区,它们占用一定的内存空间,上百万的连接,自然要耗费不少空间,如果使用两台虚拟机进行测试,建议服务器端内存设置在8个G。当然,我们可以通过setsockopt这个函数设置socket的接受和发送缓冲区的大小,越小,同一内存大小的机器上就可以接受更多的连接。
单台客户机测试效果:
达到了340000万的连接数,没截图,这里是断开连接时的效果,可以看到fd数量在30多万。
若在自行测试的过程中遇到任何问题, 欢迎评论区留言。
服务器端代码:
有两版,这一版是单线程的,下面会给出一般使用了池的代码提高效率
#include#include #include #include #include #include #include #include #include #define BUFFER_LENGTH 1024 #define LISTEN_PORT 100 struct sockitem { // int sockfd; int (*callback)(int fd, int events, void *arg); char recvbuffer[BUFFER_LENGTH]; // char sendbuffer[BUFFER_LENGTH]; int rlength; int slength; }; // mainloop / eventloop --> epoll --> struct reactor { int epfd; struct epoll_event events[512]; }; struct reactor *eventloop = NULL; int recv_cb(int fd, int events, void *arg); int send_cb(int fd, int events, void *arg) { struct sockitem *si = (struct sockitem*)arg; send(fd, si->sendbuffer, si->slength, 0); // struct epoll_event ev; ev.events = EPOLLIN | EPOLLET; //ev.data.fd = clientfd; si->sockfd = fd; si->callback = recv_cb; ev.data.ptr = si; epoll_ctl(eventloop->epfd, EPOLL_CTL_MOD, fd, &ev); } // ./epoll 8080 int recv_cb(int fd, int events, void *arg) { //int clientfd = events[i].data.fd; struct sockitem *si = (struct sockitem*)arg; struct epoll_event ev; //char buffer[1024] = {0}; int ret = recv(fd, si->recvbuffer, BUFFER_LENGTH, 0); if (ret < 0) { if (errno == EAGAIN || errno == EWOULDBLOCK) { // return -1; } else { } ev.events = EPOLLIN; //ev.data.fd = fd; epoll_ctl(eventloop->epfd, EPOLL_CTL_DEL, fd, &ev); close(fd); free(si); } else if (ret == 0) { // // printf("disconnect %dn", fd); ev.events = EPOLLIN; //ev.data.fd = fd; epoll_ctl(eventloop->epfd, EPOLL_CTL_DEL, fd, &ev); close(fd); free(si); } else { printf("Recv: %s, %d Bytesn", si->recvbuffer, ret); si->rlength = ret; memcpy(si->sendbuffer, si->recvbuffer, si->rlength); si->slength = si->rlength; struct epoll_event ev; ev.events = EPOLLOUT | EPOLLET; //ev.data.fd = clientfd; si->sockfd = fd; si->callback = send_cb; ev.data.ptr = si; epoll_ctl(eventloop->epfd, EPOLL_CTL_MOD, fd, &ev); } } int accept_cb(int fd, int events, void *arg) { struct sockaddr_in client_addr; memset(&client_addr, 0, sizeof(struct sockaddr_in)); socklen_t client_len = sizeof(client_addr); int clientfd = accept(fd, (struct sockaddr*)&client_addr, &client_len); if (clientfd <= 0) return -1; char str[INET_ADDRSTRLEN] = {0}; printf("recv from %s at port %dn", inet_ntop(AF_INET, &client_addr.sin_addr, str, sizeof(str)), ntohs(client_addr.sin_port)); struct epoll_event ev; ev.events = EPOLLIN | EPOLLET; //ev.data.fd = clientfd; struct sockitem *si = (struct sockitem*)malloc(sizeof(struct sockitem)); si->sockfd = clientfd; si->callback = recv_cb; ev.data.ptr = si; epoll_ctl(eventloop->epfd, EPOLL_CTL_ADD, clientfd, &ev); return clientfd; } int init_sock(short port) { int sockfd = socket(AF_INET, SOCK_STREAM, 0); if (sockfd < 0) { return -1; } struct sockaddr_in addr; memset(&addr, 0, sizeof(struct sockaddr_in)); addr.sin_family = AF_INET; addr.sin_port = htons(port); addr.sin_addr.s_addr = inet_addr("192.168.5.128"); //addr.sin_addr.s_addr = INADDR_ANY; if (bind(sockfd, (struct sockaddr*)&addr, sizeof(struct sockaddr_in)) < 0) { return -2; } if (listen(sockfd, 5) < 0) { return -3; } printf("port:%dn", port); return sockfd; } int main(int argc, char *argv[]) { if (argc < 2) { return -1; } eventloop = (struct reactor*)malloc(sizeof(struct reactor)); // epoll opera eventloop->epfd = epoll_create(1); int port = atoi(argv[1]); for(int i = 0; i < LISTEN_PORT; i++) { int sockfd = init_sock(port + i); struct epoll_event ev; ev.events = EPOLLIN; struct sockitem *si = (struct sockitem*)malloc(sizeof(struct sockitem)); si->sockfd = sockfd; si->callback = accept_cb; ev.data.ptr = si; epoll_ctl(eventloop->epfd, EPOLL_CTL_ADD, sockfd, &ev); } while (1) { int nready = epoll_wait(eventloop->epfd, eventloop->events, 512, -1); if (nready < -1) { break; } int i = 0; for (i = 0;i < nready;i ++) { if (eventloop->events[i].events & EPOLLIN) { //printf("sockitemn"); struct sockitem *si = (struct sockitem*)eventloop->events[i].data.ptr; si->callback(si->sockfd, eventloop->events[i].events, si); } if (eventloop->events[i].events & EPOLLOUT) { struct sockitem *si = (struct sockitem*)eventloop->events[i].data.ptr; si->callback(si->sockfd, eventloop->events[i].events, si); } } } }
服务器端代码(使用池):建议使用
#include#include #include #include #include #include #include #include #include #define BUFFER_LENGTH 1024 #define MAX_EPOLL_EVENTS 1024*1024 //connection #define MAX_EPOLL_ITEM 102400 //con #define SERVER_PORT 8888 #define LISTEN_PORT_COUNT 100 typedef int NCALLBACK(int ,int, void*); struct ntyevent { int fd; int events; void *arg; int (*callback)(int fd, int events, void *arg); int status; char buffer[BUFFER_LENGTH]; int length; long last_active; }; struct ntyreactor { int epfd; struct ntyevent *events; // 1024 * 1024 }; int recv_cb(int fd, int events, void *arg); int send_cb(int fd, int events, void *arg); void nty_event_set(struct ntyevent *ev, int fd, NCALLBACK callback, void *arg) { ev->fd = fd; ev->callback = callback; ev->events = 0; ev->arg = arg; ev->last_active = time(NULL); return ; } int nty_event_add(int epfd, int events, struct ntyevent *ev) { struct epoll_event ep_ev = {0, {0}}; ep_ev.data.ptr = ev; ep_ev.events = ev->events = events; int op; if (ev->status == 1) { op = EPOLL_CTL_MOD; } else { op = EPOLL_CTL_ADD; ev->status = 1; } if (epoll_ctl(epfd, op, ev->fd, &ep_ev) < 0) { printf("event add failed [fd=%d], events[%d]n", ev->fd, events); return -1; } return 0; } int nty_event_del(int epfd, struct ntyevent *ev) { struct epoll_event ep_ev = {0, {0}}; if (ev->status != 1) { return -1; } ep_ev.data.ptr = ev; ev->status = 0; epoll_ctl(epfd, EPOLL_CTL_DEL, ev->fd, &ep_ev); return 0; } int recv_cb(int fd, int events, void *arg) { struct ntyreactor *reactor = (struct ntyreactor*)arg; struct ntyevent *ev = reactor->events+fd; int len = recv(fd, ev->buffer, BUFFER_LENGTH, 0); nty_event_del(reactor->epfd, ev); if (len > 0) { ev->length = len; ev->buffer[len] = ' '; printf("C[%d]:%sn", fd, ev->buffer); nty_event_set(ev, fd, send_cb, reactor); nty_event_add(reactor->epfd, EPOLLOUT, ev); } else if (len == 0) { close(ev->fd); printf("[fd=%d] pos[%ld], closedn", fd, ev-reactor->events); } else { close(ev->fd); printf("recv[fd=%d] error[%d]:%sn", fd, errno, strerror(errno)); } return len; } int send_cb(int fd, int events, void *arg) { struct ntyreactor *reactor = (struct ntyreactor*)arg; struct ntyevent *ev = reactor->events+fd; int len = send(fd, ev->buffer, ev->length, 0); if (len > 0) { printf("send[fd=%d], [%d]%sn", fd, len, ev->buffer); nty_event_del(reactor->epfd, ev); nty_event_set(ev, fd, recv_cb, reactor); nty_event_add(reactor->epfd, EPOLLIN, ev); } else { close(ev->fd); nty_event_del(reactor->epfd, ev); printf("send[fd=%d] error %sn", fd, strerror(errno)); } return len; } int accept_cb(int fd, int events, void *arg) { struct ntyreactor *reactor = (struct ntyreactor*)arg; if (reactor == NULL) return -1; struct sockaddr_in client_addr; socklen_t len = sizeof(client_addr); int clientfd; if ((clientfd = accept(fd, (struct sockaddr*)&client_addr, &len)) == -1) { if (errno != EAGAIN && errno != EINTR) { } printf("accept: %sn", strerror(errno)); return -1; } int i = 0; do { #if 0 for (i = 0;i < MAX_EPOLL_EVENTS;i ++) { if (reactor->events[i].status == 0) { break; } } if (i == MAX_EPOLL_EVENTS) { printf("%s: max connect limit[%d]n", __func__, MAX_EPOLL_EVENTS); break; } #endif int flag = 0; if ((flag = fcntl(clientfd, F_SETFL, O_NONBLOCK)) < 0) { printf("%s: fcntl nonblocking failed, %dn", __func__, MAX_EPOLL_EVENTS); break; } nty_event_set(&reactor->events[clientfd], clientfd, recv_cb, reactor); nty_event_add(reactor->epfd, EPOLLIN, &reactor->events[clientfd]); } while (0); printf("new connect [%s:%d][time:%ld], pos[%d]n", inet_ntoa(client_addr.sin_addr), ntohs(client_addr.sin_port), reactor->events[i].last_active, i); return 0; } int init_sock(short port) { int fd = socket(AF_INET, SOCK_STREAM, 0); fcntl(fd, F_SETFL, O_NONBLOCK); struct sockaddr_in server_addr; memset(&server_addr, 0, sizeof(server_addr)); server_addr.sin_family = AF_INET; server_addr.sin_addr.s_addr = htonl(INADDR_ANY); server_addr.sin_port = htons(port); bind(fd, (struct sockaddr*)&server_addr, sizeof(server_addr)); if (listen(fd, 20) < 0) { printf("listen failed : %sn", strerror(errno)); } printf("listen port : %dn", port); return fd; } int ntyreactor_init(struct ntyreactor *reactor) { if (reactor == NULL) return -1; memset(reactor, 0, sizeof(struct ntyreactor)); reactor->epfd = epoll_create(1); if (reactor->epfd <= 0) { printf("create epfd in %s err %sn", __func__, strerror(errno)); return -2; } reactor->events = (struct ntyevent*)malloc((MAX_EPOLL_EVENTS) * sizeof(struct ntyevent)); if (reactor->events == NULL) { printf("create epfd in %s err %sn", __func__, strerror(errno)); close(reactor->epfd); return -3; } } int ntyreactor_destory(struct ntyreactor *reactor) { close(reactor->epfd); free(reactor->events); } int ntyreactor_addlistener(struct ntyreactor *reactor, int sockfd, NCALLBACK *acceptor) { if (reactor == NULL) return -1; if (reactor->events == NULL) return -1; nty_event_set(&reactor->events[sockfd], sockfd, acceptor, reactor); nty_event_add(reactor->epfd, EPOLLIN, &reactor->events[sockfd]); return 0; } int ntyreactor_run(struct ntyreactor *reactor) { if (reactor == NULL) return -1; if (reactor->epfd < 0) return -1; if (reactor->events == NULL) return -1; struct epoll_event events[MAX_EPOLL_ITEM]; int checkpos = 0, i; while (1) { #if 0 long now = time(NULL); for (i = 0;i < 100;i ++, checkpos ++) { if (checkpos == MAX_EPOLL_EVENTS) { checkpos = 0; } if (reactor->events[checkpos].status != 1) { continue; } long duration = now - reactor->events[checkpos].last_active; if (duration >= 60) { close(reactor->events[checkpos].fd); printf("[fd=%d] timeoutn", reactor->events[checkpos].fd); nty_event_del(reactor->epfd, &reactor->events[checkpos]); } } #endif int nready = epoll_wait(reactor->epfd, events, MAX_EPOLL_ITEM, 1000); if (nready < 0) { printf("epoll_wait error, exitn"); continue; } for (i = 0;i < nready;i ++) { struct ntyevent *ev = (struct ntyevent*)events[i].data.ptr; if ((events[i].events & EPOLLIN) && (ev->events & EPOLLIN)) { ev->callback(ev->fd, events[i].events, ev->arg); } if ((events[i].events & EPOLLOUT) && (ev->events & EPOLLOUT)) { ev->callback(ev->fd, events[i].events, ev->arg); } } } } int main(int argc, char *argv[]) { unsigned short port = SERVER_PORT; if (argc == 2) { port = atoi(argv[1]); } struct ntyreactor *reactor = (struct ntyreactor*)malloc(sizeof(struct ntyreactor)); ntyreactor_init(reactor); int listenfd[LISTEN_PORT_COUNT] = {0}; int i = 0; for (i = 0;i < LISTEN_PORT_COUNT;i ++) { listenfd[i] = init_sock(port+i); ntyreactor_addlistener(reactor, listenfd[i], accept_cb); } ntyreactor_run(reactor); ntyreactor_destory(reactor); for (i = 0;i < LISTEN_PORT_COUNT;i ++) { close(listenfd[i]); } return 0; }
客户端代码
#include#include #include #include #include #include #include #include #include #include #include #include #include #define MAX_BUFFER 128 #define MAX_EPOLLSIZE (384*1024) #define MAX_PORT 100 #define TIME_SUB_MS(tv1, tv2) ((tv1.tv_sec - tv2.tv_sec) * 1000 + (tv1.tv_usec - tv2.tv_usec) / 1000) int isContinue = 0; static int ntySetNonblock(int fd) { int flags; flags = fcntl(fd, F_GETFL, 0); if (flags < 0) return flags; flags |= O_NONBLOCK; if (fcntl(fd, F_SETFL, flags) < 0) return -1; return 0; } static int ntySetReUseAddr(int fd) { int reuse = 1; return setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, (char *)&reuse, sizeof(reuse)); } int main(int argc, char **argv) { if (argc <= 2) { printf("Usage: %s ip portn", argv[0]); exit(0); } const char *ip = argv[1]; int port = atoi(argv[2]); int connections = 0; char buffer[128] = {0}; int i = 0, index = 0; struct epoll_event events[MAX_EPOLLSIZE]; int epoll_fd = epoll_create(MAX_EPOLLSIZE); strcpy(buffer, " Data From MulClientn"); struct sockaddr_in addr; memset(&addr, 0, sizeof(struct sockaddr_in)); addr.sin_family = AF_INET; addr.sin_addr.s_addr = inet_addr(ip); struct timeval tv_begin; gettimeofday(&tv_begin, NULL); while (1) { if (++index >= MAX_PORT) index = 0; struct epoll_event ev; int sockfd = 0; if (connections < 340000 && !isContinue) { sockfd = socket(AF_INET, SOCK_STREAM, 0); if (sockfd == -1) { perror("socket"); goto err; } //ntySetReUseAddr(sockfd); addr.sin_port = htons(port+index); if (connect(sockfd, (struct sockaddr*)&addr, sizeof(struct sockaddr_in)) < 0) { perror("connect"); goto err; } ntySetNonblock(sockfd); ntySetReUseAddr(sockfd); sprintf(buffer, "Hello Server: client --> %dn", connections); send(sockfd, buffer, strlen(buffer), 0); ev.data.fd = sockfd; ev.events = EPOLLIN | EPOLLOUT; epoll_ctl(epoll_fd, EPOLL_CTL_ADD, sockfd, &ev); connections ++; } //connections ++; if (connections % 1000 == 999 || connections >= 340000) { struct timeval tv_cur; memcpy(&tv_cur, &tv_begin, sizeof(struct timeval)); gettimeofday(&tv_begin, NULL); int time_used = TIME_SUB_MS(tv_begin, tv_cur); printf("connections: %d, sockfd:%d, time_used:%dn", connections, sockfd, time_used); int nfds = epoll_wait(epoll_fd, events, connections, 100); for (i = 0;i < nfds;i ++) { int clientfd = events[i].data.fd; if (events[i].events & EPOLLOUT) { sprintf(buffer, "data from %dn", clientfd); send(sockfd, buffer, strlen(buffer), 0); } else if (events[i].events & EPOLLIN) { char rBuffer[MAX_BUFFER] = {0}; ssize_t length = recv(sockfd, rBuffer, MAX_BUFFER, 0); if (length > 0) { printf(" RecvBuffer:%sn", rBuffer); if (!strcmp(rBuffer, "quit")) { isContinue = 0; } } else if (length == 0) { printf(" Disconnect clientfd:%dn", clientfd); connections --; close(clientfd); } else { if (errno == EINTR) continue; printf(" Error clientfd:%d, errno:%dn", clientfd, errno); close(clientfd); } } else { printf(" clientfd:%d, errno:%dn", clientfd, errno); close(clientfd); } } } usleep(1 * 1000); } return 0; err: printf("error : %sn", strerror(errno)); return 0; }



