Linux高并發服務器開發（九）Tcp狀態轉移和IO多路復用

文章目錄

0 包裹函數
1 多進程服務器
- 流程
- 代碼
2 多線程服務器
3 TCP狀態轉移
- 半關閉
- 心跳包
4 端口復用
5 IO多路復用技術
- 高并發服務器
6 select
- 代碼
- 總結
7 POLL
- API
- 代碼
- poll相對select的優缺點
8 epoll（重點）
- API
- 監聽管道
- 代碼
- EPOLL 高并發服務器
9 Epoll的兩種工作方式
- 邊緣觸發代碼

0 包裹函數

用于創建socket，綁定端口ip和監聽時，添加了錯誤時報錯的包裹函數

warp.h

#ifndef __WRAP_H_
#define __WRAP_H_
#include <stdlib.h>
#include <stdio.h>
#include <unistd.h>
#include <errno.h>
#include <string.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <strings.h>void perr_exit(const char *s);
int Accept(int fd, struct sockaddr *sa, socklen_t *salenptr);
int Bind(int fd, const struct sockaddr *sa, socklen_t salen);
int Connect(int fd, const struct sockaddr *sa, socklen_t salen);
int Listen(int fd, int backlog);
int Socket(int family, int type, int protocol);
ssize_t Read(int fd, void *ptr, size_t nbytes);
ssize_t Write(int fd, const void *ptr, size_t nbytes);
int Close(int fd);
ssize_t Readn(int fd, void *vptr, size_t n);
ssize_t Writen(int fd, const void *vptr, size_t n);
ssize_t my_read(int fd, char *ptr);
ssize_t Readline(int fd, void *vptr, size_t maxlen);
int tcp4bind(short port,const char *IP);
#endif

wrap.c

#include <stdlib.h>
#include <stdio.h>
#include <unistd.h>
#include <errno.h>
#include <string.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <strings.h>void perr_exit(const char *s)
{perror(s);exit(-1);
}int Accept(int fd, struct sockaddr *sa, socklen_t *salenptr)
{int n;again:if ((n = accept(fd, sa, salenptr)) < 0) {if ((errno == ECONNABORTED) || (errno == EINTR))goto again;elseperr_exit("accept error");}return n;
}int Bind(int fd, const struct sockaddr *sa, socklen_t salen)
{int n;if ((n = bind(fd, sa, salen)) < 0)perr_exit("bind error");return n;
}int Connect(int fd, const struct sockaddr *sa, socklen_t salen)
{int n;if ((n = connect(fd, sa, salen)) < 0)perr_exit("connect error");return n;
}int Listen(int fd, int backlog)
{int n;if ((n = listen(fd, backlog)) < 0)perr_exit("listen error");return n;
}int Socket(int family, int type, int protocol)
{int n;if ((n = socket(family, type, protocol)) < 0)perr_exit("socket error");return n;
}ssize_t Read(int fd, void *ptr, size_t nbytes)
{ssize_t n;again:if ( (n = read(fd, ptr, nbytes)) == -1) {if (errno == EINTR)goto again;elsereturn -1;}return n;
}ssize_t Write(int fd, const void *ptr, size_t nbytes)
{ssize_t n;again:if ( (n = write(fd, ptr, nbytes)) == -1) {if (errno == EINTR)goto again;elsereturn -1;}return n;
}int Close(int fd)
{int n;if ((n = close(fd)) == -1)perr_exit("close error");return n;
}/*參三: 應該讀取的字節數*/
ssize_t Readn(int fd, void *vptr, size_t n)
{size_t  nleft;              //usigned int 剩余未讀取的字節數ssize_t nread;              //int 實際讀到的字節數char   *ptr;ptr = vptr;nleft = n;while (nleft > 0) {if ((nread = read(fd, ptr, nleft)) < 0) {if (errno == EINTR)nread = 0;elsereturn -1;} else if (nread == 0)break;nleft -= nread;ptr += nread;}return n - nleft;
}ssize_t Writen(int fd, const void *vptr, size_t n)
{size_t nleft;ssize_t nwritten;const char *ptr;ptr = vptr;nleft = n;while (nleft > 0) {if ( (nwritten = write(fd, ptr, nleft)) <= 0) {if (nwritten < 0 && errno == EINTR)nwritten = 0;elsereturn -1;}nleft -= nwritten;ptr += nwritten;}return n;
}static ssize_t my_read(int fd, char *ptr)
{static int read_cnt;static char *read_ptr;static char read_buf[100];if (read_cnt <= 0) {
again:if ( (read_cnt = read(fd, read_buf, sizeof(read_buf))) < 0) {if (errno == EINTR)goto again;return -1;} else if (read_cnt == 0)return 0;read_ptr = read_buf;}read_cnt--;*ptr = *read_ptr++;return 1;
}ssize_t Readline(int fd, void *vptr, size_t maxlen)
{ssize_t n, rc;char    c, *ptr;ptr = vptr;for (n = 1; n < maxlen; n++) {if ( (rc = my_read(fd, &c)) == 1) {*ptr++ = c;if (c  == '\n')break;} else if (rc == 0) {*ptr = 0;return n - 1;} elsereturn -1;}*ptr  = 0;return n;
}int tcp4bind(short port,const char *IP)
{struct sockaddr_in serv_addr;int lfd = Socket(AF_INET,SOCK_STREAM,0);bzero(&serv_addr,sizeof(serv_addr));if(IP == NULL){//如果這樣使用 0.0.0.0,任意ip將可以連接serv_addr.sin_addr.s_addr = INADDR_ANY;}else{if(inet_pton(AF_INET,IP,&serv_addr.sin_addr.s_addr) <= 0){perror(IP);//轉換失敗exit(1);}}serv_addr.sin_family = AF_INET;serv_addr.sin_port   = htons(port);Bind(lfd,(struct sockaddr *)&serv_addr,sizeof(serv_addr));return lfd;
}

1 多進程服務器

流程

因為read 和 write都是阻塞的，所以如果多個客戶端進行連接時，會阻塞。
在這里插入圖片描述
理念就是連接給一個專用的進程，然后這個進程來分配其他進程進行讀寫的操作

流程
創建套接字
綁定
監聽

while(1)
{
提取連接
fork創建子進程
子進程中關閉lfd 服務客戶端（連接）
父進程關閉 cfd（讀寫），回收子進程資源

}

代碼

#include<sys/socket.h>
#include<stdio.h>
#include<arpa/inet.h>
#include <unistd.h>
#include "wrap.h"
#include<stdlib.h>
#include <signal.h>
#include<sys/wait.h>
#include<sys/types.h>void *free_process(int signum)
{pid_t pid;while(1){pid = waitpid(-1,NULL,WNOHANG);if(pid <= 0)    // 沒有要回收的子進程{break;}else{printf("child pid =%d\n",pid);}}}int main()
{// 阻塞信號集，在子進程創建之前// 在創建子進程之后添加信號sigset_t set;sigemptyset(&set);sigaddset(&set, SIGCHLD);sigprocmask(SIG_BLOCK, &set, NULL);// 創建套接字,綁定 鏈接socketint lfd = tcp4bind(8000,NULL);// 監聽Listen(lfd, 128);// 提取// 回射struct sockaddr_in cliaddr;socklen_t len = sizeof(cliaddr);while(1){// 讀取socketint cfd = Accept(lfd, (struct sockaddr*)&cliaddr,&len);char ip[16] = "";printf("new client ip= %s port = %d\n",inet_ntop(AF_INET,&cliaddr.sin_addr.s_addr,ip,16),ntohs(cliaddr.sin_port));// fork 創建子進程pid_t pid;pid = fork();if(pid<0){perror("");exit(0);}else if(pid == 0) // 子進程{// 關閉lfdclose(lfd);while(1){char buf[1024];int n = read(cfd,buf,sizeof(buf));if(n < 0){perror("");close(cfd);exit(0);}else if(n == 0) // 對方關閉{printf("client close\n");close(cfd);exit(0);}else{printf("%s", buf);write(cfd,buf,n);// exit(0);}}}else{close(cfd); // 回收// 注冊信號回調struct sigaction act;act.sa_flags = 0;act.sa_handler = free_process;sigemptyset(&act.sa_mask);sigaction(SIGCHLD,&act,NULL);sigprocmask(SIG_UNBLOCK, &set, NULL);}}// 回收return 0;
}

2 多線程服務器

#include<sys/socket.h>
#include<stdio.h>
#include<arpa/inet.h>
#include <unistd.h>
#include "wrap.h"
#include<stdlib.h>
#include <signal.h>
#include<sys/wait.h>
#include<sys/types.h>
#include<pthread.h>typedef struct c_info
{int cfd;struct sockaddr_in cliaddr;}CINFO;int main(int argc, char *argv[])
{if(argc < 2){printf("argc < 2???\n ./a.out 8000\n");}// 初始化線程屬性pthread_attr_t attr;pthread_attr_init(&attr);pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);short port = atoi(argv[1]);int lfd = tcp4bind(port, NULL);Listen(lfd, 128);struct sockaddr_in cliaddr;socklen_t len = sizeof(cliaddr);CINFO *info;while(1){int cfd = Accept(lfd,(struct sockaddr*)&cliaddr,&len);char ip[16] = "";printf("new client ip = %s port = %d\n", inet_ntop(AF_INET,&cliaddr.sin_addr.s_addr,ip,16),ntohs(cliaddr.sin_port));pthread_t pthid;info = malloc(sizeof(CINFO));info->cfd = cfd;info->cliaddr = cliaddr;pthread_create(&pthid,NULL,client_fun,&info);}return 0;
}void * client_fun(void *arg)
{CINFO *info = (CINFO *)arg;char ip[16] = "";printf("new client ip = %s port = %d\n", inet_ntop(AF_INET,&(info->cliaddr.sin_addr.s_addr),ip,16),ntohs(info->cliaddr.sin_port));while(1){char buf[1024] = "";int count = 0;count = read(info->cfd, buf, sizeof(buf));if(count < 0){perror("");break;}else if(count == 0){printf("client close");break;}else{   printf("%s", buf);write(info->cfd, buf, count);}}close(info->cfd);free(info);}

3 TCP狀態轉移

在這里插入圖片描述
TIME_WAIT -> CLOSE 2MML

半關閉

處于FIN_WAIT2時，處于半關閉狀態，此時只能讀數據不能收數據

手動半關閉
在這里插入圖片描述

心跳包

每隔一段時間服務器向客戶端發送一個包，客戶端需要在一定時間內返回一個規定好的包，用于測試連接是否還存在，如果對方沒有回復，則斷開連接

在這里插入圖片描述

4 端口復用

端口重新啟用
使用 setsockopt設置端口重新使用
放在綁定之前
在這里插入圖片描述

5 IO多路復用技術

高并發服務器

1.阻塞等待
一個進程服務一個客戶端
消耗資源

2.非阻塞忙輪詢
重復查看進程是否有需求，是否有新連接

3.多路io
通過監聽多個文件描述符，監聽文件描述符是否還在讀寫
內核有三種方式
select：windows使用 select select跨平臺
poll：少用
epoll： linux下使用

內核監聽多個文件描述符的屬性（讀寫緩沖區）變化，如果某個文件描述符的讀緩沖區變化了，這個時候就是可以讀了，將這個事件告知應用層

在這里插入圖片描述

6 select

使用select監聽文件描述符
在這里插入圖片描述
注意：變化的文件描述符，會存放在監聽的集合中，未變化的文件描述符會被刪除

代碼

#include<sys/socket.h>
#include<stdio.h>
#include<arpa/inet.h>
#include <unistd.h>
#include<sys/time.h>
#include<sys/types.h>
#include<sys/select.h>
#include "wrap.h"#define PORT 8888
int main(int argc, char *argv[])
{// 創建套接字，綁定int lfd = tcp4bind(PORT,NULL);// 監聽Listen(lfd, 128);int maxfd = lfd;    // 最大的文件描述符fd_set oldset,rset;// 清空集合FD_ZERO(&oldset);FD_ZERO(&rset);// 將lfd加入到oldset集合中FD_SET(lfd, &oldset);// whilewhile(1){rset = oldset;  // 將oldset賦值給需要監聽的集合rsetint n = select(maxfd + 1,&rset,NULL,NULL,NULL);if(n<0){perror("");break;}else if(n==0){continue;}else    // n>0  監聽到了文件描述符{// lfd變化, 則進行提取if(FD_ISSET(lfd,&rset)){struct sockaddr_in cliaddr;socklen_t len = sizeof(cliaddr);char ip[16] = "";// 提取新的連接int cfd = Accept(lfd, (struct sockaddr*)&cliaddr, & len);printf("new client ip = %s, port = %d\n",inet_ntop(AF_INET, &cliaddr.sin_addr.s_addr,ip,16),ntohs(cliaddr.sin_port));// 將cfd添加到oldset集合中，下次進行監聽FD_SET(cfd,&oldset);// 更新maxfdif(cfd > maxfd)maxfd = cfd;// 如果只有lfd變化，continueif(--n == 0)continue;}// cfd  遍歷cfd，看lfd之后的文件描述符是否在rset，如果在則cfd變化for(int i = lfd+1;i<=maxfd;i++){// 如果i文件描述符在rset集合中if(FD_ISSET(i,&rset)){char buf[1024] = "";int ret = Read(i,buf,sizeof(buf));if(ret < 0) //出錯，將cfd關閉，從oldset刪除cfd{perror("");close(i);FD_CLR(i,&oldset);}else if(ret == 0){printf("client close\n");close(i);FD_CLR(i,&oldset);}else{printf("%s\n", buf);Write(i,buf,ret);}}}}}// select監聽return 0;
}

總結

優缺點
優點：跨平臺
缺點：文件描述符1024的限制由于FD_SETSIZE的限制
只是返回變化的文件描述符的個數，具體哪個變化需要遍歷
每次都需要將需要監聽的文件描述集合由應用層拷貝到內核

7 POLL

在這里插入圖片描述

API

在這里插入圖片描述

代碼

//IO多路復用技術poll函數的使用 
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/types.h>
#include <errno.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <poll.h>
#include "wrap.h"int main()
{int i;int n;int lfd;int cfd;int ret;int nready;int maxfd;char buf[1024];socklen_t len;int sockfd;fd_set tmpfds, rdfds;struct sockaddr_in svraddr, cliaddr;//創建socketlfd = Socket(AF_INET, SOCK_STREAM, 0);//允許端口復用int opt = 1;setsockopt(lfd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(int));//綁定bindsvraddr.sin_family = AF_INET;svraddr.sin_addr.s_addr = htonl(INADDR_ANY);svraddr.sin_port = htons(8888);ret = Bind(lfd, (struct sockaddr *)&svraddr, sizeof(struct sockaddr_in));//監聽listenret = Listen(lfd, 128);struct pollfd client[1024];for(i=0; i<1024; i++){client[i].fd = -1;}		//將監聽文件描述符委托給內核監控----監控讀事件client[0].fd = lfd;client[0].events = POLLIN;maxfd = 0; //maxfd表示內核監控的范圍while(1){nready = poll(client, maxfd+1, -1);if(nready<0){perror("poll error");exit(1);}//有客戶端連接請求if(client[0].fd==lfd && (client[0].revents & POLLIN)){cfd = Accept(lfd, NULL, NULL);//尋找client數組中的可用位置for(i=1; i<1024; i++){if(client[i].fd==-1){client[i].fd = cfd;client[i].events = POLLIN;break;}}//若沒有可用位置, 則關閉連接if(i==1024){Close(cfd);continue;}if(maxfd<i){maxfd = i;}if(--nready==0){continue;}}//下面是有數據到來的情況for(i=1; i<=maxfd; i++){//若fd為-1, 表示連接已經關閉或者沒有連接if(client[i].fd==-1)	{continue;}sockfd = client[i].fd;memset(buf, 0x00, sizeof(buf));n = Read(sockfd, buf, sizeof(buf));if(n<=0){printf("read error or client closed,n==[%d]\n", n);Close(sockfd);client[i].fd = -1; //fd為-1,表示不再讓內核監控}else{printf("read over,n==[%d],buf==[%s]\n", n, buf);write(sockfd, buf, n);}if(--nready==0){break;}}}Close(lfd);return 0;
}

poll相對select的優缺點

優點：相對于select沒有最大1024文件描述符限制
請求和返回是分離

缺點：
每次都需要將需要監聽的文件描述符從應用層拷貝到內核
每次都需要將數組中的元素遍歷一遍才知道哪個變化
大量并發、少量活躍率低

8 epoll（重點）

1.創建紅黑樹
2.將監聽的文件描述符上樹
3.監聽

特點：
沒有文件描述符1024的限制
以后每次監聽都不需要在此將需要監聽的文件描述符拷貝在內核
返回的是已經變化的文件描述符，不需要遍歷樹

工作原理：
在這里插入圖片描述

在這里插入圖片描述

API

1.創建紅黑樹
在這里插入圖片描述

2.上樹下樹修改節點

在這里插入圖片描述

3. 監聽

在這里插入圖片描述

監聽管道

在這里插入圖片描述

代碼

#include<sys/socket.h>
#include<stdio.h>
#include<arpa/inet.h>
#include <unistd.h>
#include<sys/time.h>
#include<sys/types.h>
#include<sys/select.h>
#include "wrap.h"
#include<sys/epoll.h>int main()
{int fd[2];pipe(fd);pid_t pid;pid = fork();if(pid < 0)perror("");else if(pid == 0){close(fd[0]);char buf[5];char ch = 'a';while(1){sleep(3);memset(buf,ch,sizeof(buf));write(fd[1],buf,5);}}else{close(fd[1]);// 創建樹int epfd = epoll_create(1);struct epoll_event ev, evs[1];ev.data.fd = fd[0];ev.events = EPOLLIN;epoll_ctl(epfd,EPOLL_CTL_ADD,fd[0],&ev);while(1){int n = epoll_wait(epfd, &evs[1],-1,-1);if(n == 1){   char buf[128] = "";int ret = read(fd[0],buf,sizeof(buf));if(ret <= 0){close(fd[0]);epoll_ctl(epfd,EPOLL_CTL_DEL,fd[0],&ev);break;}else{printf("%s\n",buf);}}}}return 0;}

EPOLL 高并發服務器

#include<sys/socket.h>
#include<stdio.h>
#include<arpa/inet.h>
#include <unistd.h>
#include<sys/time.h>
#include<sys/types.h>
#include<sys/select.h>
#include "wrap.h"
#include<sys/epoll.h>#define PORT 8000
int main()
{// 創建套接字int lfd = tcp4bind(PORT,NULL);// 監聽Listen(lfd,128);// 創建樹int epfd = epoll_create(1);// 將lfd上樹struct epoll_event ev,evs[1024];ev.data.fd = lfd;ev.events = EPOLLIN;epoll_ctl(epfd,EPOLL_CTL_ADD,lfd,&ev);// while循環監聽while(1){int nready = epoll_wait(epfd,evs,-1,-1);if (nready < 0){perror("");break;}else if(nready == 0){continue;}else   // nread > 0 文件描述符有變化{for(int i =0;i<nready;i++){// 判斷lfd變換，并且是讀事件變換if(evs->data.fd == lfd && evs[i].events & EPOLLIN){struct sockaddr_in cliaddr;char ip[16] = "";socklen_t len = sizeof(cliaddr);int cfd = Accept(lfd,(struct sockaddr *)&cliaddr, &len);printf("new client ip = %s port =%d\n",inet_ntop(AF_INET,&cliaddr.sin_addr.s_addr,ip,16),ntohs(cliaddr.sin_port));// 將cfd上樹ev.data.fd = cfd;ev.events = EPOLLIN;epoll_ctl(epfd,EPOLL_CTL_ADD,cfd,&ev);}else if(evs[i].events & EPOLLIN) // cfd 變換,而且是讀事件變換{char buf[1024] = "";int n = read(evs[i].data.fd,buf,sizeof(buf));if(n < 0)   // 出錯{perror("");epoll_ctl(epfd,EPOLL_CTL_DEL,evs[i].data.fd,&evs[i]);}else if(n == 0) // 客戶端關閉，下樹{printf("client close]\n");close(evs[i].data.fd); // 關閉cfdepoll_ctl(epfd,EPOLL_CTL_DEL,evs[i].data.fd,&evs[i]); }else    // 服務端進行處理{printf("%s\n",buf);write(evs[i].data.fd,buf,n);}}}}}return 0;
}

9 Epoll的兩種工作方式

在這里插入圖片描述

監聽讀緩存區的變化
水平觸發
只要緩存區有數據，就會觸發epoll_wait
邊緣觸發
數據來一次，epoll_wait只觸發一次

2.監聽寫緩存區的變化
水平觸發:只要可以寫，就會觸發
邊沿觸發：數據從有到無就會觸發

邊緣觸發
在這里插入圖片描述

觸發一次的時候只讀4位，但發送了10位，所以雖然只讀一次，但是讀不完

設置為一次讀完
設置cfd為非阻塞

在這里插入圖片描述

因為設置水平觸發，只要緩存區有數據epoll_wait就會被觸發，epoll_wait 是一個系統調用，盡量少調用邊緣觸發，邊緣觸發數據來一次只觸發一次，這個時候要求一次性將數據讀完，所以while循環讀，堵到最后read默認帶阻塞，不能讓read阻塞，因為不能再去監聽，設置cfd為非阻塞，read堵到最后一次返回值為-1，判斷errno的值為eagain，則代表數據讀干凈、

工作中邊緣觸發 + 非阻塞 = 高速模式

邊緣觸發代碼

#include<sys/socket.h>
#include<stdio.h>
#include<arpa/inet.h>
#include <unistd.h>
#include<sys/time.h>
#include<sys/types.h>
#include<sys/select.h>
#include "wrap.h"
#include<sys/epoll.h>
#include <fcntl.h>#define PORT 8000
int main()
{// 創建套接字int lfd = tcp4bind(PORT,NULL);// 監聽Listen(lfd,128);// 創建樹int epfd = epoll_create(1);// 將lfd上樹struct epoll_event ev,evs[1024];ev.data.fd = lfd;ev.events = EPOLLIN;epoll_ctl(epfd,EPOLL_CTL_ADD,lfd,&ev);// while循環監聽while(1){int nready = epoll_wait(epfd,evs,-1,-1);printf("epoll wait");if (nready < 0){perror("");break;}else if(nready == 0){continue;}else   // nread > 0 文件描述符有變化{for(int i =0;i<nready;i++){// 判斷lfd變換，并且是讀事件變換if(evs->data.fd == lfd && evs[i].events & EPOLLIN){struct sockaddr_in cliaddr;char ip[16] = "";socklen_t len = sizeof(cliaddr);int cfd = Accept(lfd,(struct sockaddr *)&cliaddr, &len);printf("new client ip = %s port =%d\n",inet_ntop(AF_INET,&cliaddr.sin_addr.s_addr,ip,16),ntohs(cliaddr.sin_port));// 獲得cfd的標志位int flag = fcntl(cfd, F_GETFL); // 設置為非阻塞flag |= O_NONBLOCK;fcntl(cfd,F_SETFL,flag);// 將cfd上樹ev.data.fd = cfd;ev.events = EPOLLIN | EPOLLET;epoll_ctl(epfd,EPOLL_CTL_ADD,cfd,&ev);}else if(evs[i].events & EPOLLIN) // cfd 變換,而且是讀事件變換{while(1){char buf[4] = "";// 如果讀一個緩沖區，緩沖區域沒有數據，如果是阻塞，就阻塞等待// 是非阻塞，返回值等于 -1，并且會將errorno值設置為EAGAIN   int n = read(evs[i].data.fd,buf,sizeof(buf));if(n < 0)   // 出錯{   // 如果緩沖區讀干凈了，這個時候應該跳出while循環，繼續監聽if(errno == EAGAIN){break;}// 普通錯誤perror("");close(evs[i].data.fd); // 關閉cfdepoll_ctl(epfd,EPOLL_CTL_DEL,evs[i].data.fd,&evs[i]);break;}else if(n == 0) // 客戶端關閉，下樹{printf("client close]\n");close(evs[i].data.fd); // 關閉cfdepoll_ctl(epfd,EPOLL_CTL_DEL,evs[i].data.fd,&evs[i]); break;}else    // 服務端進行處理{// printf("%s\n",buf);write(STDOUT_FILENO,buf,4);write(evs[i].data.fd,buf,n);}}}}}}return 0;
}