socket、select、poll、epoll实现TCP并发处理

2023-05-16

网络通信

常用网络通信接口大概四种，socket、select、poll、epoll

使用socket实现服务器的并发处理

优点：代码框架简单
缺点：碍于内存的限制，并发量不会大，基本上不会突破10K

void *client_routine(void *arg) { //

	int connfd = *(int *)arg;

	char buff[MAXLNE];

	while (1) {

		int n = recv(connfd, buff, MAXLNE, 0);
        if (n > 0) {
            buff[n] = '\0';
            printf("recv msg from client: %s\n", buff);

	    	send(connfd, buff, n, 0);
        } else if (n == 0) {
            close(connfd);
			break;
        }

	}

	return NULL;
}

int main(int argc, char **argv) 
{
    int listenfd, connfd, n;
    struct sockaddr_in servaddr;
    char buff[MAXLNE];
 
    if ((listenfd = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
        printf("create socket error: %s(errno: %d)\n", strerror(errno), errno);
        return 0;
    }
 
    memset(&servaddr, 0, sizeof(servaddr));
    servaddr.sin_family = AF_INET;
    servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
    servaddr.sin_port = htons(9999);
 
    if (bind(listenfd, (struct sockaddr *)&servaddr, sizeof(servaddr)) == -1) {
        printf("bind socket error: %s(errno: %d)\n", strerror(errno), errno);
        return 0;
    }
 
    if (listen(listenfd, 10) == -1) {
        printf("listen socket error: %s(errno: %d)\n", strerror(errno), errno);
        return 0;
    }
    
	while (1) {

		struct sockaddr_in client;
	    socklen_t len = sizeof(client);
	    if ((connfd = accept(listenfd, (struct sockaddr *)&client, &len)) == -1) {
	        printf("accept socket error: %s(errno: %d)\n", strerror(errno), errno);
	        return 0;
	    }

		pthread_t threadid;
		pthread_create(&threadid, NULL, client_routine, (void*)&connfd);

    }
    close(listenfd);
    return 0;
}

使用select实现服务器的并发处理

IO多路复用选择
利用bit位来监听管理所有链接
每一次调用select，都会将待查询的fd copy到协议栈中进行查询，然后再将查询到的信息copy到用户空间进行返回。所以性能较epoll弱。

优点

一个select可以管控指定数量的fd，多做几个select完全可以突破socket方案中难以突破的10K个并发

缺点

难以突破1000K，因为select中会将所有监听的fd拷贝到内存中判单是否存在需要读取的数据，可能其中只有几个fd可读，也就是效率较低，由于拷贝太多无效数据用以判断，所以存在优化空间

函数原型

int select (int maxfd,fd_set *readset,fd_set *writeset,fd_set *exceptset,const struct timeval * timeout);

maxfd：规定最大socketId，一般是 listen 返回ID数据+1
readset：可读集合
writeset：可写集合
exceptset：
timeout：读取等待时间，为0则阻塞
return: 可操作字的数目

需要配合使用的API

FD_ZERO
FD_SET
FD_ISSET
FD_CLR

该方法定义了一个矩阵，用以实现socket id的管理，该矩阵的大小可以修改，位于文件Posix_types.h
在这里插入图片描述

系统对输出空间的规划

在这里插入图片描述
存在一个队列，从前往后寻找可用于开辟文件描述符的位置作为文件id

int main(int argc, char **argv) 
{
    int listenfd, connfd, n;
    struct sockaddr_in servaddr;
    char buff[MAXLNE];
 
    if ((listenfd = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
        printf("create socket error: %s(errno: %d)\n", strerror(errno), errno);
        return 0;
    }
 
    memset(&servaddr, 0, sizeof(servaddr));
    servaddr.sin_family = AF_INET;
    servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
    servaddr.sin_port = htons(9999);
 
    if (bind(listenfd, (struct sockaddr *)&servaddr, sizeof(servaddr)) == -1) {
        printf("bind socket error: %s(errno: %d)\n", strerror(errno), errno);
        return 0;
    }
 
    if (listen(listenfd, 10) == -1) {
        printf("listen socket error: %s(errno: %d)\n", strerror(errno), errno);
        return 0;
    }

	fd_set rfds, rset, wfds, wset;

	FD_ZERO(&rfds);
	FD_SET(listenfd, &rfds);

	FD_ZERO(&wfds);

	int max_fd = listenfd;

	while (1) {

		rset = rfds;
		wset = wfds;

		int nready = select(max_fd+1, &rset, &wset, NULL, NULL);


		if (FD_ISSET(listenfd, &rset)) { //

			struct sockaddr_in client;
		    socklen_t len = sizeof(client);
		    if ((connfd = accept(listenfd, (struct sockaddr *)&client, &len)) == -1) {
		        printf("accept socket error: %s(errno: %d)\n", strerror(errno), errno);
		        return 0;
		    }

			FD_SET(connfd, &rfds);

			if (connfd > max_fd) max_fd = connfd;

			if (--nready == 0) continue;

		}

		int i = 0;
		for (i = listenfd+1;i <= max_fd;i ++) {

			if (FD_ISSET(i, &rset)) { // 

				n = recv(i, buff, MAXLNE, 0);
		        if (n > 0) {
		            buff[n] = '\0';
		            printf("recv msg from client: %s\n", buff);

					FD_SET(i, &wfds);

					//reactor
					//send(i, buff, n, 0);
		        } else if (n == 0) { //

					FD_CLR(i, &rfds);
					//printf("disconnect\n");
		            close(i);
					
		        }
				if (--nready == 0) break;
			} else if (FD_ISSET(i, &wset)) {

				send(i, buff, n, 0);
				FD_SET(i, &rfds);
			
			}

		}
		

	}
 
    close(listenfd);
    return 0;
}

使用poll实现服务器的并发处理

使用poll与使用select没有多少区别，只是poll不再限制最大连接数，因为它采用了链表的方式存储信息
每一次调用poll，都会将待查询的fd copy到协议栈中进行查询，然后再将查询到的信息copy到用户空间进行返回。所以性能较epoll弱。

int main(int argc, char **argv) 
{
    int listenfd, connfd, n;
    struct sockaddr_in servaddr;
    char buff[MAXLNE];
 
    if ((listenfd = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
        printf("create socket error: %s(errno: %d)\n", strerror(errno), errno);
        return 0;
    }
 
    memset(&servaddr, 0, sizeof(servaddr));
    servaddr.sin_family = AF_INET;
    servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
    servaddr.sin_port = htons(9999);
 
    if (bind(listenfd, (struct sockaddr *)&servaddr, sizeof(servaddr)) == -1) {
        printf("bind socket error: %s(errno: %d)\n", strerror(errno), errno);
        return 0;
    }
 
    if (listen(listenfd, 10) == -1) {
        printf("listen socket error: %s(errno: %d)\n", strerror(errno), errno);
        return 0;
    }

	struct pollfd fds[POLL_SIZE] = {0};
	fds[listenfd].fd = listenfd;
	fds[listenfd].events = POLLIN;

	int max_fd = listenfd;
	int i = 0;
	for (i = 1;i < POLL_SIZE;i ++) {
		fds[i].fd = -1;
	}

	while (1) {

		int nready = poll(fds, max_fd+1, -1);

	
		if (fds[listenfd].revents & POLLIN) {

			struct sockaddr_in client;
		    socklen_t len = sizeof(client);
		    if ((connfd = accept(listenfd, (struct sockaddr *)&client, &len)) == -1) {
		        printf("accept socket error: %s(errno: %d)\n", strerror(errno), errno);
		        return 0;
		    }

			printf("accept \n");
			fds[connfd].fd = connfd;
			fds[connfd].events = POLLIN;

			if (connfd > max_fd) max_fd = connfd;

			if (--nready == 0) continue;
		}

		//int i = 0;
		for (i = listenfd+1;i <= max_fd;i ++)  {

			if (fds[i].revents & POLLIN) {
				
				n = recv(i, buff, MAXLNE, 0);
		        if (n > 0) {
		            buff[n] = '\0';
		            printf("recv msg from client: %s\n", buff);
					fds[i].revents = POLLOUT;
					//send(i, buff, n, 0);
		        } else if (n == 0) { //

					fds[i].fd = -1;

		            close(i);
					
		        }
				if (--nready == 0) break;

			} else if (fds[i].revents & POLLOUT)
			{
				send(i, buff, n, 0);
				fds[i].revents = POLLIN;
			}
		}

	}

 
    close(listenfd);
    return 0;
}

使用epoll实现服务器的并发处理

吹个牛批先：在epoll出现之前，linux仅能够用作嵌入式，因为它的并发量不够高，epoll的出现，打破了这一限制，完成了linux从嵌入式往服务器跨越的进步
其原理依然是对socket的fd进行管理，将本业务中的所有socket引出的fd通过epoll_ctl加入一个epoll中进行管理，通过epoll_wait来轮询epoll中的fd是否出现事件，然后将之读出来处理。
值得一提的是，epoll可以并非IO操作，其中epoll_wait监测的接口并非IO操作
epoll与poll不同，它只是将需要关注的fd加入协议栈进行监测，所以性能更优。

int epoll_create(int size);

创建epoll句柄， size不重要，只要大于0就行。留下该入参，为了兼容老版本linux接口

int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event)

epfd：epoll_create返回的epoll句柄
op：操作类型
4. EPOLL_CTL_ADD
5. EPOLL_CTL_MOD
6. EPOLL_CTL_DEL
fd：需要监听的流
event：需要监听的事件

int epoll_wait(int epfd,struct epoll_event * events,int maxevents,int timeout)

epfd：epoll_create返回的epoll句柄
events：单次epoll_wait取到的链接情况
maxevents：单次epoll_wait最多可取的连接数
timeout：超时时间

int main(int argc, char **argv) 
{
    int listenfd, connfd, n;
    struct sockaddr_in servaddr;
    char buff[MAXLNE];
 
    if ((listenfd = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
        printf("create socket error: %s(errno: %d)\n", strerror(errno), errno);
        return 0;
    }
 
    memset(&servaddr, 0, sizeof(servaddr));
    servaddr.sin_family = AF_INET;
    servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
    servaddr.sin_port = htons(9999);
 
    if (bind(listenfd, (struct sockaddr *)&servaddr, sizeof(servaddr)) == -1) {
        printf("bind socket error: %s(errno: %d)\n", strerror(errno), errno);
        return 0;
    }
 
    if (listen(listenfd, 10) == -1) {
        printf("listen socket error: %s(errno: %d)\n", strerror(errno), errno);
        return 0;
    }

	int epfd = epoll_create(1); //int size

	struct epoll_event events[POLL_SIZE] = {0};
	struct epoll_event ev;

	ev.events = EPOLLIN;
	ev.data.fd = listenfd;

	epoll_ctl(epfd, EPOLL_CTL_ADD, listenfd, &ev);

	while (1) {

		int nready = epoll_wait(epfd, events, POLL_SIZE, 5);
		if (nready == -1) {
			continue;
		}

		int i = 0;
		for (i = 0;i < nready;i ++) {

			int clientfd =  events[i].data.fd;
			if (clientfd == listenfd) {

				struct sockaddr_in client;
			    socklen_t len = sizeof(client);
			    if ((connfd = accept(listenfd, (struct sockaddr *)&client, &len)) == -1) {
			        printf("accept socket error: %s(errno: %d)\n", strerror(errno), errno);
			        return 0;
			    }

				printf("accept\n");
				ev.events = EPOLLIN;
				ev.data.fd = connfd;
				epoll_ctl(epfd, EPOLL_CTL_ADD, connfd, &ev);

			} else if (events[i].events & EPOLLIN) {

				printf("recv\n");
				n = recv(clientfd, buff, MAXLNE, 0);
		        if (n > 0) {
		            buff[n] = '\0';
		            printf("recv msg from client: %s\n", buff);

					send(clientfd, buff, n, 0);
		        } else if (n == 0) { //


					ev.events = EPOLLIN;
					ev.data.fd = clientfd;

					epoll_ctl(epfd, EPOLL_CTL_DEL, clientfd, &ev);

		            close(clientfd);
					
		        }

			}

		}

	}
 
    close(listenfd);
    return 0;
}

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)