linux TCP并发服务器使用及说明_Linux

一、tcp并发服务器

单循环服务器：只能处理一个客户端任务的服务器；
并发服务器：可以处理多个客户端任务的服务器（一对多）。
udp服务端：本质上直接就具备并发性能。
tcp服务端：必须建立连接，所以不具备并发性能。

tcp并发服务器的实现方式：

多进程

进程资源开销大，在有限平台下，并发量比较小。

多线程

资源开销小，在相同平台下，并发量比多进程多。

线程池

多线程模型在客户端建立连接时，创建线程和销毁线程有时间消耗。
生产者-消费者设计模式
主线程：生产任务
线程池：执行任务
任务队列

io多路复用

在不创建新进程/线程的前提下，使用一个进程同时检测多路io。
io：对文件读写

二、多进程方式构建并发服务器

socket
bind
listen
connfd = accept
创建多个进程pid = fork
recv
send

代码示例：

#include "head.h"

#define ser_port 50000
#define ser_ip   "192.168.0.152"
#define max_cli_cnt 100

int init_tcp_ser()
{
	int sockfd = socket(af_inet, sock_stream, 0);
	if (sockfd < 0)
	{
		perror("socket error");
		return -1;
	}

	struct sockaddr_in seraddr;
	seraddr.sin_family = af_inet;
	seraddr.sin_port = htons(ser_port);
	seraddr.sin_addr.s_addr = inet_addr(ser_ip);

	int ret = bind(sockfd, (struct sockaddr *)&seraddr, sizeof(seraddr));
	if (ret < 0)
	{
		perror("bind error");
		return -1;
	}

	ret = listen(sockfd, max_cli_cnt);
	if (ret < 0)
	{
		perror("listen error");
		return -1;
	}

	return sockfd;
}

int main(void)
{
	int sockfd = init_tcp_ser();
	if (sockfd < 0)
	{
		return -1;
	}
	struct sockaddr_in cliaddr;
	socklen_t clilen = sizeof(cliaddr);
	
	while (1)
	{
		int connfd = accept(sockfd, (struct sockaddr *)&cliaddr, &clilen);
		if (connfd < 0)
		{
			perror("accept error");
			return -1;
		}
		printf("[%s : %d] get online\n", inet_ntoa(cliaddr.sin_addr), ntohs(cliaddr.sin_port));
		pid_t pid = fork();
		if (0 == pid)
		{
			//connfd-->cli
			char buff[1024] = {0};
			while (1)
			{
				memset(buff, 0, sizeof(buff));
				ssize_t size = recv(connfd, buff, sizeof(buff), 0);
				if (size < 0)
				{
					perror("recv error");
					close(connfd);
					break;
				}
				else if (0 == size)
				{
					close(connfd);
					break;
				}
				printf("[%s : %d] %s\n", inet_ntoa(cliaddr.sin_addr), ntohs(cliaddr.sin_port), buff);
				strcat(buff, "---->ok");
				size = send(connfd, buff, strlen(buff), 0);
				if (size < 0)
				{
					perror("send error");
					close(connfd);
					break;
				}
			}
	}
	}

	return 0;
}
```### 多线程方式构建并发服务器
1. socket
2. bind
3. listen
4. connfd = accept 
5. 循环建立多个线程pthread_create
6. recv
7. send
代码示例：

```c
#include "head.h"

#define ser_port 50000
#define ser_ip   "192.168.0.152"
#define max_cli_cnt 100

int init_tcp_ser()
{
	int sockfd = socket(af_inet, sock_stream, 0);
	if (sockfd < 0)
	{
		perror("socket error");
		return -1;
	}

	struct sockaddr_in seraddr;
	seraddr.sin_family = af_inet;
	seraddr.sin_port = htons(ser_port);
	seraddr.sin_addr.s_addr = inet_addr(ser_ip);

	int ret = bind(sockfd, (struct sockaddr *)&seraddr, sizeof(seraddr));
	if (ret < 0)
	{
		perror("bind error");
		return -1;
	}

	ret = listen(sockfd, max_cli_cnt);
	if (ret < 0)
	{
		perror("listen error");
		return -1;
	}

	return sockfd;
}


void *do_comm(void *arg)
{
	int connfd = *(int *)arg;
	free(arg);
	//connfd-->cli
	char buff[1024] = {0};
	while (1)
	{
		memset(buff, 0, sizeof(buff));
		ssize_t size = recv(connfd, buff, sizeof(buff), 0);
		if (size < 0)
		{
			perror("recv error");
			close(connfd);
			break;
		}
		else if (0 == size)
		{
			close(connfd);
			break;
		}
		printf("%s\n", buff);
		strcat(buff, "---->ok");
		size = send(connfd, buff, strlen(buff), 0);
		if (size < 0)
		{
			perror("send error");
			close(connfd);
			break;
		}
	}

	return null;
}

int main(void)
{
	pthread_t tid;
	int sockfd = init_tcp_ser();
	if (sockfd < 0)
	{
		return -1;
	}
	int connfd = 0;
	struct sockaddr_in cliaddr;
	socklen_t clilen = sizeof(cliaddr);
	
	while (1)
	{
		connfd = accept(sockfd, (struct sockaddr *)&cliaddr, &clilen);
		if (connfd < 0)
		{
			perror("accept error");
			return -1;
		}
		printf("[%s : %d] get online\n", inet_ntoa(cliaddr.sin_addr), ntohs(cliaddr.sin_port));
		int *pconnfd = malloc(sizeof(int));
		*pconnfd = connfd;
		pthread_create(&tid, null, do_comm, pconnfd);
		pthread_detach(tid);
	}
	
	close(sockfd);

	return 0;
}

三、线程池方式构建并发服务器

多线程模型在客户端建立连接时，创建线程和销毁线程有时间消耗。

生产者-消费者设计模式

主线程：生产任务
线程池：执行任务

任务队列

四、io多路复用构建并发服务器

在不创建新进程/线程的前提下，使用一个进程同时检测多路io。

io：对文件读写

普通阻塞 io：fgets()、read()这类调用会阻塞线程。

如果同时监听标准输入 fd=0和管道 fifofd两个文件描述符，调用fgets等待键盘输入时，管道来了数据也没法处理；调用read读管道时，键盘输入也处理不了，一个线程只能干等其中一个 io。

io 多路复用（select/poll/epoll）：把多个 fd 交给内核去监控。

线程阻塞在多路复用函数上，而不是阻塞在 read/fgets 上。内核检测到任意一个 fd 就绪（有数据），就返回通知用户态，这时程序再去调用read/fgets读取对应 fd 的数据。

1. 用户把stdin_fileno(0)、fifofd两个 fd 交给内核的 io 多路复用模块

2. 进程阻塞在 select/poll/epoll，不阻塞在 read/fgets

3. 内核监控这两个 fd，当某个 fd 有数据到达，唤醒用户程序

4. 用户程序再对就绪的 fd 执行read()/fgets()读取数据。

实现方式：

1. select

2. poll

3. epoll

实现流程：

创建文件描述符集合
将关注的文件描述符添加到集合
将文件描述符传递给内核，内核开始检测io事件
内核返回到达事件的结果

4.1 select实现

辅助函数（带参宏）：

void fd_clr(int fd, fd_set *set);
功能：把一个文件描述符从文件中删掉；
int  fd_isset(int fd, fd_set *set);
功能：判断文件描述符在不在集合里；
void fd_set(int fd, fd_set *set);
将文件描述符加入集合里；
void fd_zero(fd_set *set);
功能：集合清零；

4.1.1 select

特点：

使用数组（位图）保存文件描述符集合，最多允许监测1024个文件描述符；
将集合表创建在应用层，存在应用层和内核层数据的反复拷贝，消耗时间；
事件到达时，内核返回整个集合表，应用层需要遍历；
select只能工作在水平触发模式（低速模式），不能工作在边沿触发模式（高速模式）；

函数原型：

#include <sys/time.h>
#include <sys/types.h>
#include <unistd.h>

int select(int nfds, fd_set *readfds, fd_set *writefds,
           fd_set *exceptfds, struct timeval *timeout);

功能：将文件描述符传递给内核开始监测io事件，并返回监测到的结果。

参数：

nfds：监测的文件描述符最大值+1；
readfds：要监测的读事件的文件描述符集合的地址；
writefds：要监测的写事件的文件描述符集合的地址；
exceptfds：要监测的出错事件的文件描述符集合的地址；
timeout：超时时间的地址。（null表示不设置超时时间）

返回值：

成功：实际到达的io事件的个数；
失败：-1；
返回0：设置了超时事件，超时时间到而没有事件到达。

4.1.2 select基础使用代码示例：

功能说明

写端：创建命名管道myfifo，以只写方式打开，每隔 1 秒往管道写入hello world。
读端：使用select同时监听两个文件描述符：

fifofd：命名管道，接收写端发来的数据

stdin_fileno(0)：标准输入，读取键盘输入

写端：

#include "head.h"

int main(void)
{
	mkfifo("myfifo", 0664);
	if(errno != eexist){
    perror("mkfifo");
    return -1;
	}
	int fifofd = open("myfifo", o_wronly);
	if (fifofd < 0)
	{
		perror("open fifo error");
		return -1;
	}
	
	while (1)
	{
		write(fifofd, "hello world", 11);
		sleep(1);
	}

	close(fifofd);

	return 0;
}

读端：

#include "head.h"

int main(void)
{
	char buff[1024] = {0};
	
	mkfifo("myfifo", 0664);
	int fifofd = open("myfifo", o_rdonly);
	if (fifofd < 0)
	{
		perror("open fifo error");
		return -1;
	}
	
	fd_set tmpfds;
	fd_set rdfds;
	fd_zero(&rdfds);	

	fd_set(fifofd, &rdfds);
	int maxfd = fifofd;
	fd_set(stdin_fileno, &rdfds);
	maxfd = maxfd > stdin_fileno ? maxfd : stdin_fileno;

	while (1)
	{
		tmpfds = rdfds;
		int cnt = select(maxfd+1, &tmpfds, null, null, null);
		if (cnt < 0)
		{
			perror("select eror");
			return -1;
		}
		
		if (fd_isset(fifofd, &tmpfds))
		{
			memset(buff, 0, sizeof(buff));
			//read(fifofd, buff, sizeof(buff));
			ssize_t n = read(fifofd, buff, sizeof(buff)-1);
			if(n < 0){
			    perror("read fifo error");
			    fd_clr(fifofd, &rdfds);
			    close(fifofd);
			}else if(n == 0){
			    fd_clr(fifofd, &rdfds);
			    close(fifofd);
			}
			printf("fifo : %s\n", buff);
		}
		if (fd_isset(stdin_fileno, &tmpfds))
		{
			fgets(buff, sizeof(buff), stdin); 
			printf("stdin: %s\n", buff);
		}
	}

	close(fifofd);
	return 0;
}

4.1.3 select实现tcp并发服务器

#include "head.h"

#define ser_port 50000
#define ser_ip   "192.168.0.152"
#define max_cli_cnt 100

int init_tcp_ser()
{
	int sockfd = socket(af_inet, sock_stream, 0);
	if (sockfd < 0)
	{
		perror("socket error");
		return -1;
	}

	struct sockaddr_in seraddr;
	seraddr.sin_family = af_inet;
	seraddr.sin_port = htons(ser_port);
	seraddr.sin_addr.s_addr = inet_addr(ser_ip);

	int ret = bind(sockfd, (struct sockaddr *)&seraddr, sizeof(seraddr));
	if (ret < 0)
	{
		perror("bind error");
		return -1;
	}

	ret = listen(sockfd, max_cli_cnt);
	if (ret < 0)
	{
		perror("listen error");
		return -1;
	}

	return sockfd;
}



int main(void)
{

	int sockfd = init_tcp_ser();
	if (sockfd < 0)
	{
		return -1;
	}
	char buff[1024] = {0};
	struct sockaddr_in cliaddr;
	socklen_t clilen = sizeof(cliaddr);
	int maxfd = 0;
	
	fd_set tmpfds;
	fd_set rdfds;
	fd_zero(&rdfds);
	
	fd_set(sockfd, &rdfds);
	maxfd = sockfd;

	while (1)
	{
		tmpfds = rdfds;
		int cnt = select(maxfd+1, &tmpfds, null, null, null);
		if (cnt < 0)
		{
			perror("select error");
			return -1;
		}
		for (int i = sockfd; i <= maxfd; ++i)
		{
			if (fd_isset(i, &tmpfds))
			{
				if (i == sockfd)
				{
					int connfd = accept(sockfd, (struct sockaddr *)&cliaddr, &clilen);
					if (connfd < 0)
					{
						perror("accept error");
						return -1;
					}
					printf("[%s : %d] get online\n", inet_ntoa(cliaddr.sin_addr), ntohs(cliaddr.sin_port));
					fd_set(connfd, &rdfds);
					maxfd = maxfd > connfd ? maxfd : connfd;
				}
				else
				{
					memset(buff, 0, sizeof(buff));
					ssize_t size = recv(i, buff, sizeof(buff), 0);
					if (size < 0)
					{
						perror("recv error");
						fd_clr(i, &rdfds);
						close(i);
						continue;
					}
					else if (0 == size)
					{
						fd_clr(i, &rdfds);
						close(i);
						continue;
					}
					printf("%s\n", buff);
					strcat(buff, "--->ok!");
					size = send(i, buff, strlen(buff), 0);
					if (size < 0)	
					{
						perror("send error");
						fd_clr(i, &rdfds);
						close(i);
						continue;
					}
				}
			}
		}
	
	}		
	return 0;
}

4.2 poll

特点：

使用链表保存文件描述符集合，理论上允许监测的文件描述符没有上限限制；
创建的文件描述符集合在应用层，需要应用层和内核层的反复数据拷贝；
应用层对于返回的事件结果需要遍历查找；
只能工作在水平触发模式（低速），无法工作在边沿触发模式（高速）。

4.3 epoll实现

特点：

使用树形结构（红黑树）保存文件描述符集合，无文件描述符上限限制，并且查找效率高；
文件描述符集合直接创建在内核层，避免了应用层和内核层的反复数据拷贝；
直接返回到达事件的文件描述符集合；
可以工作在水平触发模式，也可以工作在边沿触发模式。

4.3.1 epoll_create

函数原型：

#include <sys/epoll.h>
int epoll_create(int size);

功能：

创建一个文件描述符的集合。

参数：

size：允许监测的文件描述符的个数；

返回值：

成功：返回文件描述符的句柄；

失败：返回-1。

4.3.2 epoll_ctl

函数原型：

#include <sys/epoll.h>
int  epoll_ctl(int  epfd,  int  op,  int   fd,   struct 
epoll_event *event);

功能：

对文件描述符集合中的文件描述符做操作。

参数：

epfd：epoll建立的文件描述符集合；
op：操作方法：

1. epoll_ctl_add（添加）

2. epoll_ctl_mod（修改）

3. epoll_ctl_del（删除）

fd：要操作的文件描述符；
event：epoll事件结构体。

struct epoll_event {
    uint32_t     events;      /* epoll events */
    epoll_data_t data;        /* user data variable */
};
typedef union epoll_data {
     void        *ptr;
     int          fd;//要监测的文件描述符
     uint32_t     u32;
     uint64_t     u64;
} epoll_data_t;

events：事件类型：

1. epollin：写事件；

2. epollout：读事件。

返回值：

成功：返回文件描述符的句柄；
失败：返回-1。

4.3.3 epoll_wait

函数原型：

#include <sys/epoll.h>

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

功能：

通知内核开始监测事件。

参数：

epfd：监测的文件描述符的集合；
events：保存epoll返回的到达事件的集合；
maxevents：最多监测的事件个数；
timeout：设置超时时间；（-1：不设置超时，阻塞等待）

返回值：

成功：返回实际到达事件的个数；
失败：返回-1。
返回0：超时时间到达，返回的事件数得0.

4.3.4 基础使用代码示例

功能说明

写端：创建命名管道myfifo，以只写方式打开，每隔 1 秒往管道写入hello world。

读端：使用select同时监听两个文件描述符：

fifofd：命名管道，接收写端发来的数据
stdin_fileno(0)：标准输入，读取键盘输入

建立fifo文件：

#include "head.h"

int main(void)
{
	mkfifo("myfifo", 0664);

	int fifofd = open("myfifo", o_wronly);
	if (fifofd < 0)
	{
		perror("open fifo error");
		return -1;
	}
	
	while (1)
	{
		write(fifofd, "hello world", 11);
		sleep(1);
	}

	close(fifofd);

	return 0;
}

使用epoll达到io多路复用：

#include "head.h"

#define max_fd_cnt 2

int add_epoll_fd(int epfds, int fd, uint32_t event)
{
	struct epoll_event ev;
	ev.events = event;
	ev.data.fd = fd;
	int ret = epoll_ctl(epfds, epoll_ctl_add, fd, &ev);
	if (ret < 0)
	{
		perror("epoll_ctl error");
		return -1;
	}
	return 0;
}


int delete_epoll_fd(int epfds, int fd)
{
	int ret = epoll_ctl(epfds, epoll_ctl_del, fd, null);
	if (ret < 0)
	{
		perror("epoll_ctl error");
		return -1;
	}
	return 0;
}

int main(void)
{
	char buff[1024] = {0};
	
	mkfifo("myfifo", 0664);
	int fifofd = open("myfifo", o_rdonly);
	if (fifofd < 0)
	{
		perror("open fifo error");
		return -1;
	}
	
	int epfds = epoll_create(max_fd_cnt);
	if (epfds < 0)
	{
		perror("epoll_create error");
		return -1;
	}
	
	add_epoll_fd(epfds, fifofd, epollin);
	add_epoll_fd(epfds, stdin_fileno, epollin);

	struct epoll_event evs[max_fd_cnt];
	while (1)
	{
		int cnt = epoll_wait(epfds, evs, max_fd_cnt, -1);
		if (cnt < 0)
		{
			perror("epoll_wait error");
			return -1;
		}
		for (int i = 0; i < cnt; ++i)
		{
			if (evs[i].data.fd == fifofd)
			{
				memset(buff, 0, sizeof(buff));
				ssize_t size = read(evs[i].data.fd, buff, sizeof(buff));
				if (size < 0)
				{
					perror("read fifo error");
					delete_epoll_fd(epfds, evs[i].data.fd);
					close(evs[i].data.fd);
				}
				else if (0 == size)
				{
					delete_epoll_fd(epfds, evs[i].data.fd);
					close(evs[i].data.fd);	
				}
				printf("fifo : %s\n", buff);
			}
			else if (evs[i].data.fd == stdin_fileno)
			{
				fgets(buff, sizeof(buff), stdin);
				printf("stdin : %s\n", buff);
			}
		}
	
	}

	close(fifofd);
	return 0;
}

4.3.5 epoll实现tcp并发服务器

#include "head.h"

#define ser_port 50000
#define ser_ip   "192.168.0.152"
#define max_cli_cnt 100

int init_tcp_ser()
{
	int sockfd = socket(af_inet, sock_stream, 0);
	if (sockfd < 0)
	{
		perror("socket error");
		return -1;
	}

	struct sockaddr_in seraddr;
	seraddr.sin_family = af_inet;
	seraddr.sin_port = htons(ser_port);
	seraddr.sin_addr.s_addr = inet_addr(ser_ip);

	int ret = bind(sockfd, (struct sockaddr *)&seraddr, sizeof(seraddr));
	if (ret < 0)
	{
		perror("bind error");
		return -1;
	}

	ret = listen(sockfd, max_cli_cnt);
	if (ret < 0)
	{
		perror("listen error");
		return -1;
	}

	return sockfd;
}


int add_epoll_fd(int epfds, int fd, uint32_t event)
{
	struct epoll_event ev;
	ev.events = event;
	ev.data.fd = fd;
	int ret = epoll_ctl(epfds, epoll_ctl_add, fd, &ev);
	if (ret < 0)
	{
		perror("epoll_ctl add error");
		return -1;
	}

	return 0;
}

int delete_epoll_fd(int epfds, int fd)
{
	int ret = epoll_ctl(epfds, epoll_ctl_del, fd, null);
	if (ret < 0)
	{
		perror("epoll_ctl del error");
		return -1;
	}
	return 0;
}


int main(void)
{

	int sockfd = init_tcp_ser();
	if (sockfd < 0)
	{
		return -1;
	}
	struct sockaddr_in cliaddr;
	socklen_t clilen = sizeof(cliaddr);

	int epfds = epoll_create(max_cli_cnt);
	if (epfds < 0)
	{
		perror("epoll_create error");
		return -1;
	}

	add_epoll_fd(epfds, sockfd, epollin);

	struct epoll_event evs[max_cli_cnt];
	char buff[1024] = {0};
	while (1)
	{
		int cnt = epoll_wait(epfds, evs, max_cli_cnt, -1);
		if (cnt < 0)
		{
			perror("epoll_wait error");
			return -1;
		}
		for (int i = 0; i < cnt; ++i)
		{
			if (evs[i].data.fd == sockfd)
			{
				int connfd = accept(sockfd, (struct sockaddr *)&cliaddr, &clilen);
				if (connfd < 0)
				{
					perror("accept error");
					return -1;
				}
				printf("[%s : %d] get online\n", inet_ntoa(cliaddr.sin_addr), ntohs(cliaddr.sin_port));

				add_epoll_fd(epfds, connfd, epollin);
			}
			else
			{
				memset(buff, 0, sizeof(buff));
				ssize_t size = recv(evs[i].data.fd, buff, sizeof(buff), 0);
				if (size < 0)
				{
					perror("recv error");
					delete_epoll_fd(epfds, evs[i].data.fd);
					close(evs[i].data.fd);
					continue;
				}
				else if (0 == size)
				{
					delete_epoll_fd(epfds, evs[i].data.fd);
					close(evs[i].data.fd);	
					continue;
				}
				printf("%s\n", buff);
				strcat(buff, "--->ok!");
				size = send(evs[i].data.fd, buff, strlen(buff), 0);
				if (size < 0)
				{
					perror("send error");
					delete_epoll_fd(epfds, evs[i].data.fd);
					close(evs[i].data.fd);
					continue;
				}
			}
		}


	}

	return 0;
}