简单实现Tcp服务器的百万并发
准备工作
1.准备好4个虚拟机
一个 4G内存，2核cpu 作为Server端
三个 2G内存，1核cpu 作为Client端
目标
利用客户端不断地向服务端发起请求，把并发量跑到100万
问题的发现和解决方案
刚开始我们尝试只使用服务端的一个端口去试验
我们分别跑起来四台虚拟机里的代码
发现其中一个客户端只连接了1024左右的连接量
并且报出提示
1. Connection refused
这时我们排查到时进程文件限制最多打来文件数量是1024，因为每建立一次连接，客户端都需要去开一个文件描述符，因此我们要把这个限制调大
如果调不上去或者想永久生效的话，改下配置文件
或者也可能是系统文件总限制数量
接着往下跑又出现问题
显示情况是地址被用完了，已经没有可以分配的地址了
这个现象的原因分析：
对于客户端而言：
建立连接的五元组 socketfd ------> (远程ip，远程端口，本机ip, 本机端口，TCP)（对于客户端而言）
每个连接都有这样一个唯一的五元组，一一对应，现在对于远程（服务器）ip和端口都定了，本机ip也是定的，所以变的只能是客户端的端口，而客户端的端口很少只有65536个所以，是很有限的
解决办法是
考虑把服务端地址加大100则可加大连接数
继续跑又出现：的问题
connect:Connection timed out
error :Connection timed out
排查到时连接跟踪系统（conntrack）最大条目数的参数，全称是 net.nf_conntrack_max
通过系统文件修改即可
最后我们继续考虑到协议栈的限制：
1. net.ipv4.tcp_mem
作用：控制 TCP 协议栈的内存使用上限，由三个数值组成，分别对应 “低阈值、压力阈值、高阈值”（单位：页，1 页通常为 4KB）。
252144：低阈值，当内存使用低于此值时，TCP 不会主动回收内存；
524288：压力阈值，当内存使用超过此值时，TCP 开始主动回收内存；
786432：高阈值，当内存使用超过此值时，TCP 会拒绝新的连接请求，避免内存耗尽。
2. net.ipv4.tcp_wmem
作用：控制 TCP 发送窗口（write buffer）的内存大小范围（单位：字节）。
1024：最小发送窗口大小；
1024：默认发送窗口大小；
2048：最大发送窗口大小。
该参数影响 TCP 发送数据的速率，增大数值可提升高带宽场景下的发送效率。
3. net.ipv4.tcp_rmem
作用：控制 TCP 接收窗口（read buffer）的内存大小范围（单位：字节），结构与 tcp_wmem 一致。
1024：最小接收窗口大小；
1024：默认接收窗口大小；
2048：最大接收窗口大小。
该参数影响 TCP 接收数据的速率，增大数值可提升高并发或大流量场景下的接收效率。
在系统文件中修改即可
sudo vim /etc/sysctl.conf
改完别忘了生效一下
sudo sysctl -p
此时排除各种问题，达到了百万并发！！！
此时我们直接结束程序几十万的连接突然断开，会发现cpu被顶的很高的现象
当大量连接（如你截图中的340000+连接）集中断开时，客户端和服务端需要执行一系列资源回收操作：
关闭套接字（close）、释放文件描述符；
清理内存中的连接元数据（如缓冲区、状态标记）；
处理 TCP 的TIME_WAIT/CLOSE_WAIT状态回收。
这些操作在短时间内并发执行，会导致 CPU 集中处理 “资源销毁” 任务，从而出现 CPU 占用率飙升的情况。
服务端的代码：
#include
#include
#include
#include
#include
#include
#include
#include
#include
#define BUFFER_LENGTH 1024
#define EPOLL_SIZE 1024
#define MAX_PORT 100
void *client_routine(void *arg) {
int clientfd = *(int *)arg;
while (1) {
char buffer[BUFFER_LENGTH] = {0};
int len = recv(clientfd, buffer, BUFFER_LENGTH, 0);
if (len < 0) {
close(clientfd);
break;
} else if (len == 0) { // disconnect
close(clientfd);
break;
} else {
printf("Recv: %s, %d byte(s)
", buffer, len);
}
}
}
int islistenfd(int fd, int *fds) {
int i = 0;
for (i = 0;i < MAX_PORT;i ++) {
if (fd == *(fds+i)) return fd;
}
return 0;
}
// ./tcp_server
int main(int argc, char *argv[]) {
if (argc < 2) {
printf("Param Error
");
return -1;
}
int port = atoi(argv[1]); // start
int sockfds[MAX_PORT] = {0}; // listen fd
int epfd = epoll_create(1);
int i = 0;
for (i = 0;i < MAX_PORT;i ++) {
int sockfd = socket(AF_INET, SOCK_STREAM, 0);
struct sockaddr_in addr;
memset(&addr, 0, sizeof(struct sockaddr_in));
addr.sin_family = AF_INET;
addr.sin_port = htons(port+i); // 8888 8889 8890 8891 .... 8987
addr.sin_addr.s_addr = INADDR_ANY;
if (bind(sockfd, (struct sockaddr*)&addr, sizeof(struct sockaddr_in)) < 0) {
perror("bind");
return 2;
}
if (listen(sockfd, 5) < 0) {
perror("listen");
return 3;
}
printf("tcp server listen on port : %d
", port + i);
struct epoll_event ev;
ev.events = EPOLLIN;
ev.data.fd = sockfd;
epoll_ctl(epfd, EPOLL_CTL_ADD, sockfd, &ev);
sockfds[i] = sockfd;
}
//
#if 0
while (1) {
struct sockaddr_in client_addr;
memset(&client_addr, 0, sizeof(struct sockaddr_in));
socklen_t client_len = sizeof(client_addr);
int clientfd = accept(sockfd, (struct sockaddr*)&client_addr, &client_len);
pthread_t thread_id;
pthread_create(&thread_id, NULL, client_routine, &clientfd);
}
#else
struct epoll_event events[EPOLL_SIZE] = {0};
while (1) {
int nready = epoll_wait(epfd, events, EPOLL_SIZE, 5); // -1, 0, 5
if (nready == -1) continue;
int i = 0;
for (i = 0;i < nready;i ++) {
int sockfd = islistenfd(events[i].data.fd, sockfds);
if (sockfd) { // listen 2
struct sockaddr_in client_addr;
memset(&client_addr, 0, sizeof(struct sockaddr_in));
socklen_t client_len = sizeof(client_addr);
int clientfd = accept(sockfd, (struct sockaddr*)&client_addr, &client_len);
fcntl(clientfd, F_SETFL, O_NONBLOCK);
int reuse = 1;
/**
SO_REUSEADDR 更常用于监听套接字（sockfd）（服务器用于接受连接的套接字），
避免服务器重启时因旧连接的 TIME_WAIT 状态导致绑定失败。
*/
setsockopt(clientfd, SOL_SOCKET, SO_REUSEADDR, (char *)&reuse, sizeof(reuse));
struct epoll_event ev;
ev.events = EPOLLIN | EPOLLET;
ev.data.fd = clientfd;
epoll_ctl(epfd, EPOLL_CTL_ADD, clientfd, &ev);
} else {
int clientfd = events[i].data.fd;
char buffer[BUFFER_LENGTH] = {0};
int len = recv(clientfd, buffer, BUFFER_LENGTH, 0);
if (len < 0) {
close(clientfd);
struct epoll_event ev;
ev.events = EPOLLIN;
ev.data.fd = clientfd;
epoll_ctl(epfd, EPOLL_CTL_DEL, clientfd, &ev);
} else if (len == 0) { // disconnect
close(clientfd);
struct epoll_event ev;
ev.events = EPOLLIN;
ev.data.fd = clientfd;
epoll_ctl(epfd, EPOLL_CTL_DEL, clientfd, &ev);
} else {
printf("Recv: %s, %d byte(s), clientfd: %d
", buffer, len, clientfd);
}
}
}
}
#endif
return 0;
}
客户端代码
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#define MAX_BUFFER 128
#define MAX_EPOLLSIZE (384*1024)
#define MAX_PORT 99
#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 port
", 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 MulClient
");
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 --> %d
", 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:%d
", 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 %d
", 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:%s
", rBuffer);
if (!strcmp(rBuffer, "quit")) {
isContinue = 0;
}
} else if (length == 0) {
printf(" Disconnect clientfd:%d
", clientfd);
connections --;
close(clientfd);
} else {
if (errno == EINTR) continue;
printf(" Error clientfd:%d, errno:%d
", clientfd, errno);
close(clientfd);
}
} else {
printf(" clientfd:%d, errno:%d
", clientfd, errno);
close(clientfd);
}
}
}
usleep(1 * 1000);
}
return 0;
err:
printf("error : %s
", strerror(errno));
return 0;
}
引用自0voice
