一、TCP并发服务器
- 单循环服务器：只能处理一个客户端任务的服务器；
- 并发服务器：可以处理多个客户端任务的服务器（一对多）。
- UDP服务端：本质上直接就具备并发性能。
- TCP服务端：必须建立连接，所以不具备并发性能。
TCP并发服务器的实现方式：
多进程
- 进程资源开销大，在有限平台下，并发量比较小。
多线程
- 资源开销小，在相同平台下，并发量比多进程多。
线程池
- 多线程模型在客户端建立连接时，创建线程和销毁线程有时间消耗。
- 生产者-消费者设计模式
- 主线程：生产任务
- 线程池：执行任务
- 任务队列
IO多路复用
- 在不创建新进程/线程的前提下，使用一个进程同时检测多路IO。
- IO：对文件读写
二、多进程方式构建并发服务器
- socket
- bind
- listen
- connfd = accept
- 创建多个进程pid = fork
- recv
- send
代码示例：
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 |
#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实现
辅助函数（带参宏）：
|
1 2 3 4 5 6 7 8 |
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只能工作在水平触发模式（低速模式），不能工作在边沿触发模式（高速模式）；
函数原型：
|
1 2 3 4 5 6 |
#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)：标准输入，读取键盘输入
写端：
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 |
#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; } |
读端：
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 |
#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并发服务器
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 |
#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
函数原型：
|
1 2 |
#include <sys/epoll.h> int epoll_create(int size); |
功能：
创建一个文件描述符的集合。
参数：
size：允许监测的文件描述符的个数；
返回值：
成功：返回文件描述符的句柄；
失败：返回-1。
4.3.2 epoll_ctl
函数原型：
|
1 2 3 |
#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事件结构体。
|
1 2 3 4 5 6 7 8 9 10 |
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
函数原型：
|
1 2 3 4 |
#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文件：
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 |
#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多路复用：
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 |
#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并发服务器
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 |
#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; } |
总结
以上为个人经验，希望能给大家一个参考，也希望大家多多支持脚本之家。
掌握TCP并发服务器三大实现方案：
多进程、线程池和IO多路复用,深入解析select、poll、epoll的核心差异与实战代码，帮你告别阻塞困境，直接在项目中应用最优架构，显著提升服务器性能。
