TCP并发模型：
1. TCP多线程模型：
缺点：创建线程会带来资源开销，能够实现的并发量比较有限。
2. IO模型：
1. 阻塞IO：
没有数据到来时，可以让任务挂起，节省CPU资源开销，提高系统效率
2. 非阻塞IO：
程序未接收到数据时一直执行，效率很低
举例应用：
write.c
#include "head.h"
int main(void)
{
int fd = 0;
char tmpbuff[1024] = {0};
mkfifo("/tmp/myfifo", 0664);
fd = open("/tmp/myfifo", O_WRONLY);
if(fd == -1)
{
perror("fail to open");
return -1;
}
while(1)
{
memset(tmpbuff, 0, sizeof(tmpbuff));
gets(tmpbuff);
write(fd, tmpbuff, strlen(tmpbuff));
}
close(fd);
return 0;
}后面的举例应用的write.c都为上面所示
read.c
#include "head.h"
int main(void)
{
int fd = 0;
int flags = 0;
ssize_t nsize = 0;
char *pret = NULL;
char tmpbuff[1024] = {0};
mkfifo("/tmp/myfifo", 0664);
fd = open("/tmp/myfifo", O_RDONLY);
if(fd == -1)
{
perror("fail to open");
return -1;
}
flags = fcntl(fd, F_GETFL);
flags |= O_NONBLOCK;
fcntl(fd, F_SETFL, flags);
flags = fcntl(0, F_GETFL);
flags |= O_NONBLOCK;
fcntl(0, F_SETFL, flags);
while(1)
{
memset(tmpbuff, 0, sizeof(tmpbuff));
pret = gets(tmpbuff);
if(pret != NULL)
{
printf("STDIN:%s\n", tmpbuff);
}
memset(tmpbuff, 0, sizeof(tmpbuff));
nsize = read(fd, tmpbuff, sizeof(tmpbuff));
if(nsize > 0)
{
printf("RECV:%s\n", tmpbuff);
}
}
close(fd);
return 0;
}通过fcntl将fd和stdin文件描述符设置为非阻塞IO，所以read.c既可以通过管道接收消息，也可以同终端输入，接收消息，二者不会堵塞。
3. 异步IO：
只能绑定一个文件描述符用来读取数据
举例应用：
read.c
#include "head.h"
int fd = 0;
void handler(int signo)
{
char tmpbuff[1024] = {0};
ssize_t nsize = 0;
memset(tmpbuff, 0, sizeof(tmpbuff));
nsize = read(fd, tmpbuff, sizeof(tmpbuff));
if(nsize > 0)
{
printf("RECV:%s\n", tmpbuff);
}
return;
}
int main(void)
{
int flags = 0;
ssize_t nsize = 0;
char *pret = NULL;
char tmpbuff[1024] = {0};
signal(SIGIO, handler);
mkfifo("/tmp/myfifo", 0664);
fd = open("/tmp/myfifo", O_RDONLY);
if(fd == -1)
{
perror("fail to open");
return -1;
}
flags = fcntl(fd, F_GETFL);
flags |= O_ASYNC;
fcntl(fd, F_SETFL, flags);
fcntl(fd, F_SETOWN, getpid());
while(1)
{
memset(tmpbuff, 0, sizeof(tmpbuff));
gets(tmpbuff);
printf("STDIN:%s\n", tmpbuff);
}
close(fd);
return 0;
}
4. 多路复用：
1. select：
1. select监听的集合中的文件描述符有上限限制
2. select有内核层向用户层数据空间拷贝的过程，占用系统资源开销
原因：在使用select函数时，会涉及到内核层和用户层之间的数据传输。具体来说：select函数在内核层会监视一组文件描述符的状态变化，并在有变化发生时通知用户层。为了是实现这个功能，内核会定期检测文件描述符的状态，并在发生变化时将相关信息传递给用户层。这个过程涉及到内核层和用户层之间的数据传输，即内核层需要将监视的文件描述符的状态信息传递给用户层。这个传输过程会涉及到系统资源的开销，因为需要进行数据拷贝操作，将内核中的数据拷贝到用户空间，这个数据拷贝的过程会消耗一定的系统资源。
3. select必须轮询检测产生事件的文件描述符
4. select只能工作在水平触发模式(低速模式)，无法工作在边沿触发模式(高速模式)
原因：
1. 在水平触发模式下，一旦文件描述符中的数据可读或可写，select就会通知用户程序，并且如果这些描述符在之后的select调用中仍然保持可读或可写状态，select会再次通知程序。换句话说，在水平触发模式下，只要描述符的状态处于可读或可写状态，select就会通知程序。
2. 在边缘触发模式下，只有当描述符的状态发生变化时才会通知程序，这意味着如果描述符中的数据量发生变化或者有新的数据到达，程序才会被通知，而如果描述符的状态保持不变，即使在之后的select调用中它任然处于可读或可写状态，程序也不会被通知。
举例应用：
read.c
#include "head.h"
int main(void)
{
int fd = 0;
int flags = 0;
ssize_t nsize = 0;
char *pret = NULL;
fd_set rdfds;
fd_set tmpfds;
int ret = 0;
char tmpbuff[1024] = {0};
mkfifo("/tmp/myfifo", 0664);
fd = open("/tmp/myfifo", O_RDONLY);
if(fd == -1)
{
perror("fail to open");
return -1;
}
FD_ZERO(&rdfds);
FD_SET(fd, &rdfds);
FD_SET(0, &rdfds);
while(1)
{
tmpfds = rdfds;
ret = select(fd+1, &tmpfds, NULL, NULL, NULL);
if(ret == -1)
{
perror("fail to select");
return -1;
}
if(FD_ISSET(0, &tmpfds))
{
memset(tmpbuff, 0, sizeof(tmpbuff));
gets(tmpbuff);
printf("STDIN:%s\n", tmpbuff);
}
if(FD_ISSET(fd, &tmpfds))
{
memset(tmpbuff, 0, sizeof(tmpbuff));
read(fd, tmpbuff, sizeof(tmpbuff));
printf("FIFO:%s\n", tmpbuff);
}
}
close(fd);
return 0;
}
2. poll：
1. poll有内核层向用户层数据空间拷贝的过程，占用系统资源开销
2. poll必须轮询检测产生事件的文件描述符
3. poll只能工作在水平触发模式(低速模式)，无法工作在边沿触发模式(高速模式)
举例应用：
read.c
#include "head.h"
int main(void)
{
int fd = 0;
int ret = 0;
int flags = 0;
ssize_t nsize = 0;
struct pollfd fds[2];
char tmpbuff[1024] = {0};
mkfifo("/tmp/myfifo", 0664);
fd = open("/tmp/myfifo", O_RDONLY);
if(fd == -1)
{
perror("fail to open");
return -1;
}
fds[0].fd = 0;
fds[0].events = POLLIN;
fds[1].fd = fd;
fds[1].events = POLLIN;
while(1)
{
ret = poll(fds, 2, -1);
if(ret == -1)
{
perror("fail to poll");
return -1;
}
if(fds[0].revents & POLLIN)
{
memset(tmpbuff, 0, sizeof(tmpbuff));
gets(tmpbuff);
printf("STDIN:%s\n", tmpbuff);
}
if(fds[1].revents & POLLIN)
{
memset(tmpbuff, 0, sizeof(tmpbuff));
read(fd, tmpbuff, sizeof(tmpbuff));
printf("FIFO:%s\n", tmpbuff);
}
}
close(fd);
return 0;
}
3. epoll：
1. epoll没有文件描述符上限限制
2. epoll创建内核监听事件表，所以只需要在内核空间完成数据拷贝即可
3. epoll会将产生事件的文件描述符对应的事件直接返回
4. epoll可以工作在水平触发模式(默认：低速模式)，还可以工作在边沿触发模式(高速模式) 在epoll_events的结构体中的events传入EPOLLIN | EPOLLET即可
举例应用：
read.c
#include "head.h"
int main(void)
{
int i = 0;
int fd = 0;
int epfd = 0;
int nready = 0;
ssize_t nsize = 0;
char tmpbuff[1024] = {0};
struct epoll_event env;
struct epoll_event retenv[2];
mkfifo("/tmp/myfifo", 0664);
fd = open("/tmp/myfifo", O_RDONLY);
if(fd == -1)
{
perror("fail to open");
return -1;
}
epfd = epoll_create(2);
if(epfd == -1)
{
perror("fail to epoll_create");
return -1;
}
env.data.fd = 0;
env.events = EPOLLIN;
epoll_ctl(epfd, EPOLL_CTL_ADD, 0, &env);
env.data.fd = fd;
env.events = EPOLLIN;
epoll_ctl(epfd, EPOLL_CTL_ADD, fd, &env);
while(1)
{
nready = epoll_wait(epfd, retenv, 2, -1);
if(nready == -1)
{
perror("fail to epoll_wait");
return -1;
}
for(i = 0; i < nready; i++)
{
if(retenv[i].data.fd == 0)
{
memset(tmpbuff, 0, sizeof(tmpbuff));
gets(tmpbuff);
printf("STDIN:%s\n", tmpbuff);
}
else if(retenv[i].data.fd == fd)
{
memset(tmpbuff, 0, sizeof(tmpbuff));
read(fd, tmpbuff, sizeof(tmpbuff));
printf("FIFO:%s\n", tmpbuff);
}
}
}
close(fd);
return 0;
}
3. 函数接口：
1. select：
int select(int nfds, fd_set *readfds, fd_set *writefds,fd_set *exceptfds, struct timeval *timeout);功能：select监听文件描述符集合中是否有文件描述编程ready状态
参数：
nfds：最大文件描述符的值+1
reafds：读文件描述符集合
writefds：写文件描述符集合
exceptfds：其余文件描述符集合
timeout：等待的时长
NULL：一直等待
返回值：
成功返回文件描述符集合中的文件描述符个数
失败返回-1
void FD_CLR(int fd, fd_set *set);功能：判断文件描述符fd是否仍在集合中
void FD_SET(int fd, fd_set *set);功能：将文件描述符fd加入到集合中
int FD_ISSET(int fd, fd_set *set);功能：判断文件描述符fd是否仍在集合中
void FD_ZERO(fd_set *set);功能：将文件描述符集合清0
2. poll：
int poll(struct pollfd *fds, nfds_t nfds, int timeout);功能：监听文件描述符集合是否有事件发生
参数：
fds：监听文件描述符集合是否有时间发生
nfds：监听文件描述符集合元素个数
timeout：等待时间（-1一直等待）
struct pollfd {
int fd; /* file descriptor */
short events; /* requested events */
short revents; /* returned events */
};fd：监听的文件描述
events：要监听的事件，POLLIN：是否可读，POLLOUT：是否可写
revents：实际产生的事件
3. epoll：
1. epoll_create：
int epoll_create(int size);功能：创建一张内核事件表
参数：
size：事件的个数
返回值：
成功返回文件描述符
失败返回-1
2. epoll_ctl：
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);功能：维护epoll时间表
参数：
epfd：事件表的文件描述符
op：
EPOLL_CTL_ADD 添加事件
EPOLL_CTL_MODE 修改事件
EPOLL_CTL_DEL 删除事件
fd：操作的文件描述符
events：事件对应的事件
typedef union epoll_data {
void *ptr;
in fd;
uint32_t u32;
uint64_t u64;
} epoll_data_t;
struct epoll_event {
uint32_t events; /* Epoll events */
epoll_data_t data; /* User data variable */
};返回值：
成功返回0
失败返回-1
3. epoll_wait：
int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout);功能：监听事件表中的事件
参数：
epfd：文件描述符
events：存放实际产生事件的数组空间首地址
maxevents：最多存放事件的个数
timeout：设定监听的时间（超过该时间则不再监听）
-1：一直监听直到有事件发生
返回值：
成功返回产生事件的文件描述符个数
失败返回-1
如果时间达到仍没有事件发生返回0
4. 练习作业：
1. 编写TCP并发模型之select模型
client.c
#include "head.h"
int CreateTcpClient(char *pip, int port)
{
int ret = 0;
int sockfd = 0;
struct sockaddr_in seraddr;
sockfd = socket(AF_INET, SOCK_STREAM, 0);
if (-1 == sockfd)
{
perror("fail to socket");
return -1;
}
seraddr.sin_family = AF_INET;
seraddr.sin_port = htons(port);
seraddr.sin_addr.s_addr = inet_addr(pip);
ret = connect(sockfd, (struct sockaddr *)&seraddr, sizeof(seraddr));
if (-1 == ret)
{
perror("fail to connect");
return -1;
}
return sockfd;
}
int main(void)
{
int sockfd = 0;
char tmpbuff[4096] = {"hello world"};
int cnt = 0;
ssize_t nsize = 0;
sockfd = CreateTcpClient("192.168.1.125", 50000);
while (1)
{
memset(tmpbuff, 0, sizeof(tmpbuff));
sprintf(tmpbuff, "hello world --- %d", cnt);
cnt++;
nsize = send(sockfd, tmpbuff, strlen(tmpbuff), 0);
if (-1 == nsize)
{
perror("fail to send");
return -1;
}
memset(tmpbuff, 0, sizeof(tmpbuff));
nsize = recv(sockfd, tmpbuff, sizeof(tmpbuff), 0);
if (-1 == nsize)
{
perror("fail to recv");
return -1;
}
printf("RECV:%s\n", tmpbuff);
}
close(sockfd);
return 0;
}后面的client.c都为上面所示
server.c
#include "head.h"
int CreateListenSocket(char *pip, int port)
{
int ret = 0;
int sockfd = 0;
struct sockaddr_in serveraddr;
sockfd = socket(AF_INET, SOCK_STREAM, 0);
if(sockfd == -1)
{
perror("fail to socket");
return -1;
}
serveraddr.sin_family = AF_INET;
serveraddr.sin_port = htons(port);
serveraddr.sin_addr.s_addr = inet_addr(pip);
ret = bind(sockfd, (struct sockaddr *)&serveraddr, sizeof(serveraddr));
if(ret == -1)
{
perror("fail to bind");
return -1;
}
ret =listen(sockfd, 10);
if(ret == -1)
{
perror("fail to listen");
return -1;
}
return sockfd;
}
int HandleTcpClient(int confd)
{
char tmpbuff[4096] = {0};
ssize_t nsize = 0;
memset(tmpbuff, 0, sizeof(tmpbuff));
nsize = recv(confd, tmpbuff, sizeof(tmpbuff), 0);
if(nsize == -1)
{
perror("fail to recv");
return -1;
}
else if(nsize == 0)
{
return 0;
}
sprintf(tmpbuff, "%s ------echo", tmpbuff);
nsize = send(confd, tmpbuff, strlen(tmpbuff), 0);
if(nsize == -1)
{
perror("fail to send");
return -1;
}
return nsize;
}
int main(void)
{
int i = 0;
int ret = 0;
int confd = 0;
int sockfd = 0;
fd_set rdfds;
fd_set tmpfds;
int maxfd = 0;
sockfd = CreateListenSocket("192.168.1.125", 50000);
FD_ZERO(&rdfds);
FD_SET(sockfd, &rdfds);
maxfd = sockfd;
while(1)
{
tmpfds = rdfds;
ret = select(maxfd+1, &tmpfds, NULL, NULL, NULL);
if(ret == -1)
{
perror("fail to select");
return -1;
}
if(FD_ISSET(sockfd, &tmpfds))
{
confd = accept(sockfd, NULL, NULL);
if(confd == -1)
{
perror("fail to accept");
FD_CLR(sockfd, &rdfds);
close(sockfd);
continue;
}
FD_SET(confd, &rdfds);
maxfd = maxfd > confd ? maxfd : confd;
}
for(i = sockfd+1; i <= maxfd; i++)
{
if(FD_ISSET(i, &tmpfds))
{
ret = HandleTcpClient(i);
if(ret == -1)
{
fprintf(stderr, "handle client fialed!\n");
FD_CLR(i, &rdfds);
close(i);
continue;
}
else if(ret == 0)
{
fprintf(stderr, "client disconnected!\n");
FD_CLR(i, &rdfds);
close(i);
continue;
}
}
}
}
close(confd);
close(sockfd);
return 0;
}
2. 编写TCP并发模型之poll模型
server.c
#include "head.h"
int CreateListenSocket(char *pip, int port)
{
int ret = 0;
int sockfd = 0;
struct sockaddr_in serveraddr;
sockfd = socket(AF_INET, SOCK_STREAM, 0);
if(sockfd == -1)
{
perror("fail to socket");
return -1;
}
serveraddr.sin_family = AF_INET;
serveraddr.sin_port = htons(port);
serveraddr.sin_addr.s_addr = inet_addr(pip);
ret = bind(sockfd, (struct sockaddr *)&serveraddr, sizeof(serveraddr));
if(ret == -1)
{
perror("fail to bind");
return -1;
}
ret =listen(sockfd, 10);
if(ret == -1)
{
perror("fail to listen");
return -1;
}
return sockfd;
}
int HandleTcpClient(int confd)
{
char tmpbuff[4096] = {0};
ssize_t nsize = 0;
memset(tmpbuff, 0, sizeof(tmpbuff));
nsize = recv(confd, tmpbuff, sizeof(tmpbuff), 0);
if(nsize == -1)
{
perror("fail to recv");
return -1;
}
else if(nsize == 0)
{
return 0;
}
sprintf(tmpbuff, "%s ------echo", tmpbuff);
nsize = send(confd, tmpbuff, strlen(tmpbuff), 0);
if(nsize == -1)
{
perror("fail to send");
return -1;
}
return nsize;
}
int InitFds(struct pollfd *fds, int maxlen)
{
int i = 0;
for(i = 0; i < maxlen; i++)
{
fds[i].fd = -1;
}
return 0;
}
int AddFds(struct pollfd *fds, int maxlen, int fd, short env)
{
int i = 0;
for(i = 0; i < maxlen; i++)
{
if(fds[i].fd == -1)
{
fds[i].fd = fd;
fds[i].events = env;
break;
}
}
if(i == maxlen)
{
return -1;
}
return 0;
}
int DeleteFds(struct pollfd *fds, int maxlen, int fd)
{
int i = 0;
for(i = 0; i < maxlen; i++)
{
if(fds[i].fd == fd)
{
fds[i].fd = -1;
break;
}
}
return 0;
}
int main(void)
{
int i = 0;
int ret = 0;
int nready = 0;
int confd = 0;
int sockfd = 0;
struct pollfd fds[1024];
sockfd = CreateListenSocket("192.168.1.125", 50000);
InitFds(fds, 1024);
AddFds(fds, 1024, sockfd, POLLIN);
while(1)
{
nready = poll(fds, 1024, -1);
if(nready == -1)
{
perror("fail to poll");
return -1;
}
for(i = 0; i < 1024; i++)
{
if(fds[i] == -1)
{
continue;
}
if(fds[i].revents & POLLIN && fds[i].fd == sockfd)
{
confd = accept(sockfd, NULL, NULL);
if(confd == -1)
{
perror("fail to accept");
close(sockfd);
DeleteFds(fds, 1024, sockfd);
continue;
}
AddFds(fds, 1024, confd, POLLIN);
}
else if(fds[i].revents & POLLIN && fds[i].fd != sockfd)
{
ret = HandleTcpClient(fds[i].fd);
if(ret == -1)
{
fprintf(stderr, "handle tcp client failed!\n");
close(fds[i].fd);
DeleteFds(fds, 1024, fds[i].fd);
continue;
}
else if(ret == 0)
{
fprintf(stderr, "client disconnected!\n");
close(fds[i].fd);
DeleteFds(fds, 1024, fds[i].fd);
continue;
}
}
}
}
close(confd);
close(sockfd);
return 0;
}
3. 编写TCP并发模型之epoll模型
server.c
#include "head.h"
int CreateListenSocket(char *pip, int port)
{
int ret = 0;
int sockfd = 0;
struct sockaddr_in serveraddr;
sockfd = socket(AF_INET, SOCK_STREAM, 0);
if(sockfd == -1)
{
perror("fail to socket");
return -1;
}
serveraddr.sin_family = AF_INET;
serveraddr.sin_port = htons(port);
serveraddr.sin_addr.s_addr = inet_addr(pip);
ret = bind(sockfd, (struct sockaddr *)&serveraddr, sizeof(serveraddr));
if(ret == -1)
{
perror("fail to bind");
return -1;
}
ret =listen(sockfd, 10);
if(ret == -1)
{
perror("fail to listen");
return -1;
}
return sockfd;
}
int HandleTcpClient(int confd)
{
char tmpbuff[4096] = {0};
ssize_t nsize = 0;
memset(tmpbuff, 0, sizeof(tmpbuff));
nsize = recv(confd, tmpbuff, sizeof(tmpbuff), 0);
if(nsize == -1)
{
perror("fail to recv");
return -1;
}
else if(nsize == 0)
{
return 0;
}
sprintf(tmpbuff, "%s ------echo", tmpbuff);
nsize = send(confd, tmpbuff, strlen(tmpbuff), 0);
if(nsize == -1)
{
perror("fail to send");
return -1;
}
return nsize;
}
int AddFd(int epfd, int fd, uint32_t env)
{
int ret = 0;
struct epoll_event tmpenv;
tmpenv.data.fd = fd;
tmpenv.events = env;
ret = epoll_ctl(epfd, EPOLL_CTL_ADD, fd, &tmpenv);
if(ret == -1)
{
perror("fail to epoll_ctl");
return -1;
}
return 0;
}
int DelFd(int epfd, int fd)
{
int ret = 0;
ret = epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL);
if(ret == -1)
{
perror("fail to epoll_ctl");
return -1;
}
return 0;
}
int main(void)
{
int i = 0;
int ret = 0;
int nready = 0;
int confd = 0;
int sockfd = 0;
struct epoll_event retenv[1024];
int epfd = 0;
epfd = epoll_create(1024);
if(epfd == -1)
{
perror("fail to epoll_create");
return -1;
}
sockfd = CreateListenSocket("192.168.1.155", 50000);
AddFd(epfd, sockfd, EPOLLIN);
while(1)
{
nready = epoll_wait(epfd, retenv, 1024, -1);
if(nready == -1)
{
perror("fail to epoll");
return -1;
}
for(i = 0; i < nready; i++)
{
if(retenv[i].data.fd == sockfd)
{
confd = accept(sockfd, NULL, NULL);
if(confd == -1)
{
perror("fail to accept");
DelFd(epfd, sockfd);
close(sockfd);
continue;
}
AddFd(epfd, confd, EPOLLIN);
}
else if(retenv[i].data.fd != sockfd)
{
ret = HandleTcpClient(retenv[i].data.fd);
if(ret == -1)
{
fprintf(stderr, "handle tcp client failed!\n");
DelFd(epfd, retenv[i].data.fd);
close(retenv[i].data.fd);
continue;
}
else if(ret == 0)
{
fprintf(stderr, "client disconnected!\n");
DelFd(epfd, retenv[i].data.fd);
close(retenv[i].data.fd);
continue;
}
}
}
}
close(epfd);
close(sockfd);
return 0;
}
