Cleofwine | 开发者

后端技术_实现百万并发

最近更新：2024-09-23 | 字数总计：2k | 阅读估时：9分钟 | 阅读量：次

服务端测评指标
C10X: C10K与C10M

服务端测评指标

并发（能够同时承载的客户端数量）
qps：每秒处理请求/事务数量
最大时延长/响应时间：一般取平均响应时间
新建，建链
吞吐量（并发数、QPS、响应时间）

QPS = 并发数 / 平均响应时间
并发数 = QPS * 平均响应时间

C10X: C10K与C10M

C10K 就是单机同时处理 1 万个请求（并发连接 1 万）的问题
C1000K 也就是单机支持处理 100 万个请求（并发连接 100 万）的问题
C10M 也就是单机支持处理 1000 万个请求（并发连接 1000 万）的问题

EPOLL实现C1000K

##define EPOLL_EVENT_LENGTH 1024 // 由于这种初始分配内存的方式，导致并发仅有千个客户端
int zv_init_reactor(zv_reactor_t *reactor){
	if(!reactor) return -1;

	reactor->blockheader = malloc (sizeof(zv_connblock_t) + EPOLL_EVENT_LENGTH * sizeof(zv_connect_t)); 
	if(reactor->blockheader == NULL) return -1;
	reactor->blockheader->block = (zv_connect_t*)(reactor->blockheader + 1);
	reactor->epfd = epoll_create(1);
}

改进

当fd > 1024时扩展块出来

##include <stdio.h>
##include <errno.h>
##include <unistd.h>
##include <string.h>
##include <stdlib.h>
##include <sys/socket.h>
##include <netinet/in.h>

##include <fcntl.h>
##include <pthread.h>
##include <sys/poll.h>
##include <sys/epoll.h>


##define BUFFER_LENGTH 1024
##define EPOLL_EVENT_LENGTH 1024

typedef int (*ZVCALLBACK)(int fd, int events, void *arg);

typedef struct zv_connect_s{
	int fd;

	ZVCALLBACK cb;
	
	char rbuffer[BUFFER_LENGTH];
	int rc;
	int count;
	char wbuffer[BUFFER_LENGTH];
	int wc;
}zv_connect_t;

typedef struct zv_connblock_s {
	zv_connect_t * block;
	struct zv_connblock_s *next;
}zv_connblock_t;

typedef struct zv_reactor_s{
	int epfd;
	int blkcnt;
	zv_connblock_t *blockheader; 
}zv_reactor_t;



int zv_init_reactor(zv_reactor_t *reactor){
	if(!reactor) return -1;

##if 0
	reactor -> blockheader = malloc(sizeof(zv_connblock_t));
	if(reactor->blockheader == NULL) return -1;

	reactor -> blockheader -> block = calloc(1024, sizeof(zv_connect_t));
	if(reactor->blockheader->block == NULL) return -1;
##else
	reactor->blockheader = malloc (sizeof(zv_connblock_t) + EPOLL_EVENT_LENGTH * sizeof(zv_connect_t)); 
	if(reactor->blockheader == NULL) return -1;
	reactor->blockheader->block = (zv_connect_t*)(reactor->blockheader + 1);
	reactor->blkcnt = 1; // 第一块
	reactor->blockheader->next = NULL;
##endif

	reactor->epfd = epoll_create(1);

}

void zv_dest_reactor(zv_reactor_t *reactor){
	if(!reactor) return;

	if(!reactor->blockheader) free(reactor->blockheader);

	close(reactor->epfd);
}

// 扩展块，尾插法
int zv_connect_block(zv_reactor_t *reactor){
	if(!reactor) return -1;
	
	zv_connblock_t *blk = reactor->blockheader;
	while(blk->next != NULL) blk = blk->next;

	zv_connblock_t *connblock = malloc (sizeof(zv_connblock_t) + EPOLL_EVENT_LENGTH * sizeof(zv_connect_t)); 
	if(connblock == NULL) return -1;
	connblock->block = (zv_connect_t*)(reactor->blockheader + 1);
	connblock->next = NULL;

	blk->next = connblock;

	reactor->blkcnt++; 

	return 0;
}

// 根据fd找到对应的块及块中的位置，返回具体的connect
// 若fd超出目前的上限，则扩展块
zv_connect_t* zv_connect_idx(zv_reactor_t *reactor, int fd){
	if(!reactor) return NULL;

	int blkidx = fd / EPOLL_EVENT_LENGTH;

	while(blkidx >= reactor->blkcnt){ // 需要进行块扩展
		int ret = zv_connect_block(reactor);
		if(ret != 0)
			return NULL;
	}

	int i = 0;
	zv_connblock_t* blk = reactor->blockheader;
	while(i++ < blkidx){
		blk = blk->next;
	}
	return blk->block +(fd % EPOLL_EVENT_LENGTH);
}



int init_server(short port)
{

	int sockfd = socket(AF_INET, SOCK_STREAM, 0);

	struct sockaddr_in servaddr;
	memset(&servaddr, 0, sizeof(struct sockaddr_in));
	servaddr.sin_family = AF_INET;
	servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
	servaddr.sin_port = htons(port);

	if(-1 == bind(sockfd, (struct sockaddr*)&servaddr, sizeof(struct sockaddr))){
		printf("bind failed: %s", strerror(errno));
		return -1;
	}
	// 开始监听
	listen(sockfd, -1);

	return sockfd;
}

int recv_cb(int fd, int events, void *arg);


int send_cb(int fd, int events, void *arg){
	printf("send_cb \n ");
	zv_reactor_t * reactor = (zv_reactor_t *)arg;
	
	if(!reactor || !reactor->blockheader) return -1;

	zv_connect_t *conn = zv_connect_idx(reactor, fd);
	send(fd, conn->wbuffer, conn->wc, 0);
	conn->cb = recv_cb;
	struct epoll_event ev;
	ev.events = EPOLLIN;
	ev.data.fd = fd;
	epoll_ctl(reactor->epfd, EPOLL_CTL_MOD, fd, &ev);
}


int recv_cb(int fd, int events, void *arg){
	printf("recv_cb \n ");
	zv_reactor_t * reactor = (zv_reactor_t *)arg;
	
	if(!reactor || !reactor->blockheader) return -1;

	zv_connect_t* conn =  zv_connect_idx(reactor, fd);
	int ret = recv(fd, conn->rbuffer+conn->rc, conn->count, 0);
	if(ret < 0){
	
	}else if(ret == 0){
		conn->fd = -1;
		conn->rc = 0;
		conn->wc = 0;

		epoll_ctl(reactor->epfd, EPOLL_CTL_DEL, fd, NULL);
		close(fd);
	}
	conn->rc += ret;
	// printf("rbuffer: %s, %d\n", conn->rbuffer, conn->rc);
	// conn->count = *(short*)conn->rbuffer;
	conn->count = 20;


	// ECHO 模式

	memcpy(conn->wbuffer, conn->rbuffer, conn->rc);
	conn->wc = conn->rc;


	conn->cb = send_cb;
	struct epoll_event ev;
	ev.events = EPOLLOUT;
	ev.data.fd = fd;
	epoll_ctl(reactor->epfd, EPOLL_CTL_MOD, fd, &ev);
}



int accept_cb(int fd, int events, void *arg)
{

	printf("accept_cb \n ");

	struct sockaddr_in clientaddr;
	socklen_t len = sizeof(clientaddr);
	int clientfd = accept(fd, (struct sockaddr*)&clientaddr, &len);

	if(clientfd < 0)
	{
		printf("accept err ,%d \n",errno);
		return -1;
	}

	zv_reactor_t * reactor = (zv_reactor_t *)arg;
	
	if(!reactor || !reactor->blockheader) return -1;

	zv_connect_t* conn =  zv_connect_idx(reactor, clientfd);
	
	conn->fd = clientfd;
	conn->cb = recv_cb;
	conn->count = 2;

	printf("client fd : %d", clientfd);
	struct epoll_event ev;
	ev.events = EPOLLIN;
	ev.data.fd = clientfd;
	epoll_ctl(reactor->epfd, EPOLL_CTL_ADD, clientfd, &ev);
}


int set_listener(zv_reactor_t *reactor, int fd, ZVCALLBACK cb){
	if(!reactor || !reactor->blockheader) return -1;
	reactor->blockheader->block[fd].fd = fd;
	reactor->blockheader->block[fd].cb = cb;

	struct epoll_event ev;
	ev.events = EPOLLIN; // 默认是水平触发LT
	ev.data.fd = fd;

	epoll_ctl(reactor->epfd, EPOLL_CTL_ADD, fd, &ev);
}

int main(int argc, char* argv[])
{
	if(argc < 2) return -1;

	int port = atoi(argv[1]);
	int sockfd = init_server(port);
	
	zv_reactor_t reactor;
	zv_init_reactor(&reactor);


	set_listener(&reactor,sockfd, accept_cb);

	struct epoll_event events[EPOLL_EVENT_LENGTH] = {0};
	while(1){
		int nready = epoll_wait(reactor.epfd, events, EPOLL_EVENT_LENGTH, -1); // -1等待, 0不等, 1一定时间等待
		if(nready < 0) continue;

		int i = 0;

		for(i = 0;i < nready;i++){
			int connfd = events[i].data.fd;
			zv_connect_t *conn = zv_connect_idx(&reactor, connfd);

			if(events[i].events & EPOLLIN){
				conn->cb(connfd, events[i].events, &reactor);
			}
			if(events[i].events & EPOLLOUT){
				conn->cb(connfd, events[i].events, &reactor);
			}
		}
	}

}

结论

epoll本身就支持百万 io管理，一百万才是正常的
reactor 单线程就可以达到一百万
并发情况（回答）：1.测试结果；2.线上实际情况
C1000K 的解决方法总结，本质上还是构建在 epoll 的非阻塞 I/O 模型上。只不过，除了 I/O 模型之外，还需要从应用程序到 Linux 内核、再到 CPU、内存和网络等各个层次的深度优化，特别是需要借助硬件，来卸载那些原来通过软件处理的大量功能。

C10M问题

在 C1000K 问题中，各种软件、硬件的优化很可能都已经做到头了。特别是当升级完硬件（比如足够多的内存、带宽足够大的网卡、更多的网络功能卸载等）后，你可能会发现，无论你怎么优化应用程序和内核中的各种网络参数，想实现 1000 万请求的并发，都是极其困难的。
究其根本，还是 Linux 内核协议栈做了太多太繁重的工作。从网卡中断带来的硬中断处理程序开始，到软中断中的各层网络协议处理，最后再到应用程序，这个路径实在是太长了，就会导致网络包的处理优化，到了一定程度后，就无法更进一步了
要解决这个问题，最重要就是跳过内核协议栈的冗长路径，把网络包直接送到要处理的应用程序那里去。这里有两种常见的机制，DPDK 和 XDP:

DPDK:是用户态网络的标准。它跳过内核协议栈，直接由用户态进程通过轮询的方式，来处理网络接收。
XDP:则是 Linux 内核提供的一种高性能网络数据路径。它允许网络包，在进入内核协议栈之前，就进行处理，也可以带来更高的性能。XDP 底层跟我们之前用到的 bcc-tools 一样，都是基于 Linux 内核的 eBPF 机制实现的

服务端并发瓶颈及总结

C10K 问题的根源

一方面在于系统有限的资源；
另一方面，也是更重要的因素，是同步阻塞的 I/O 模型以及轮询的套接字接口，限制了网络事件的处理效率。

C10K解决：

Linux 2.6 中引入的 epoll ，完美解决了 C10K 的问题，现在的高性能网络方案都基于 epoll

从 C10K 到 C100K ，可能只需要增加系统的物理资源就可以满足；但从 C100K 到 C1000K ，就不仅仅是增加物理资源就能解决的问题了。这时，就需要多方面的优化工作了，从硬件的中断处理和网络功能卸载、到网络协议栈的文件描述符数量、连接状态跟踪、缓存队列等内核的优化，再到应用程序的工作模型优化，都是考虑的重点。
要实现 C10M ，就不只是增加物理资源，或者优化内核和应用程序可以解决的问题了。

这时候，就需要用 XDP 的方式，在内核协议栈之前处理网络包；
或者用 DPDK 直接跳过网络协议栈，在用户空间通过轮询的方式直接处理网络包

2023-11-29 该篇文章被 Cleofwine 归为分类: 服务端