TinyWebserver的复现与改进(4):主线程的具体实现
GitHub - yzfzzz/MyWebServer: Linux高并发服务器项目,参考了TinyWebServer,将在此基础上进行性能改进与功能增加。为方便读者学习,附带详细注释和博客!
TinyWebserver的复现与改进(1):服务器环境的搭建与测试-CSDN博客
TinyWebserver的复现与改进(2):项目的整体框架-CSDN博客
TinyWebserver的复现与改进(3):线程同步机制类封装及线程池实现-CSDN博客
Reactor模式
今天我们将采用主从Reactor多线程模式,这是是大多数高性能服务器采用的模式
主从Reactor多线程模式要求主线程(I/O处理单元)只需负责:
- 监听文件描述符上是否有事件发生,
- 有的话就立即将该事件通知工作线程(逻辑单元),
- 将 socket 可读可写事件放入请求队列,交给工作线程处理。
除此之外,主线程不做任何其他实质性的工作。读写数据,接受新的连接,以及处理客户请求均在工作线程中完成。
Reactor模式的工作流程
使用同步 I/O(以 epoll_wait 为例)实现的 Reactor 模式的工作流程是:
- 主线程往 epoll 内核事件表中注册 socket 上的读就绪事件。 (即把listenfd、和已连接的客户端socketfd加入到epoll模型中)
- 主线程调用 epoll_wait 等待 socket 上有数据可读。
- 当 socket 上有数据可读时, epoll_wait 通知主线程。主线程则将 socket 可读事件放入请求队列。
- 正在堵塞的某个工作线程被解除堵塞,它从 socket 读取数据,并处理客户请求,然后往 epoll 内核事件表中注册该 socket 上的写就绪事件。
- 当主线程调用 epoll_wait 等待 socket 可写。
- 当 socket 可写时,epoll_wait 通知主线程。主线程将 socket 可写事件放入请求队列。
- 堵塞的某个工作线程被解除堵塞,它往 socket 上写入服务器处理客户请求的结果。
代码实现
main函数流程图如图所示:
初始化
创建线程池
// 定义一个线程池指针
threadpool<http_conn>* pool = NULL;
try {
// 开辟一个线程池
pool = new threadpool<http_conn>;
}catch(...)
{
// 若异常则退出
return 1;
}
使用new创建一个http_conn类型的线程池,返回的指针由pool接收。由于使用了 try……catch(...)
语句,因此如果遇到异常则退出
创建客户端集合
// 开辟一块连续的http_conn数组,保存所有正在连接的客户端信息
http_conn* users = new http_conn[MAX_FD];
创建一个大小为 MAX_FD 的 http_conn 数组,当接收一个新 socket_fd 时,将会在 索引 i = socket的数组处(即 users[socket_fd]),初始化一个 http_conn 对象。这个对象保存着该客户端的所有信息。
创建监听
// 设置监听
int listenfd = socket(AF_INET, SOCK_STREAM, 0);
int ret = 0;
struct sockaddr_in address;
address.sin_addr.s_addr = INADDR_ANY; // 表示本机的所有IP都可以连接客户端
address.sin_family = AF_INET; // 使用ipv4
address.sin_port = htons(port);
// 设置端口复用
int reuse = 1;
setsockopt(listenfd, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse));
// 绑定
ret = bind(listenfd, (struct sockaddr*)&address, sizeof(address));
if(ret == -1)
{
perror("bind");
exit(-1);
}
// 开始监听
ret = listen(listenfd, 5);
if(ret == -1)
{
perror("listen");
exit(-1);
}
-
socket 创建一个套接字,AF_INET表示使用 ipv4,SOCK_STREAM表示使用流式协议,和后面的0搭配表示使用的是TCP协议
-
但此时的socket,没有主机的端口号和ip信息,我们需要使用结构体address存储主机的相关信息:如IP(
address.sin_addr.s_addr = INADDR_ANY
)、ipv4(address.sin_family = AF_INET
)、端口(address.sin_port = htons(port)
),其中htons()
表示 主机字节序 => 网络字节序 ,因为不同的机器在内存中的存放字节的顺序不同,我们需要统一标准 -
假设用户先运行./server.out,关闭后再快速运行./server.out,会报错:该端口被占用。这是因为结束./server,服务端处于FIN_WAIT状态,可以认为TCP通讯的TIME_WAIT时期需要等一小段时间。为了解决这个问题,常用
setsockopt()
表示开启端口复用功能。 -
bind
将fd 和本地的IP和端口进行绑定,然后开始监听
epoll初始化
// 将listend添加到epoll模型中
epoll_event events[MAX_EVENT_NUMBER];
int epollfd = epoll_create(5);
addfd(epollfd, listenfd, false);
http_conn::m_epollfd = epollfd;
1.epoll_event用于检测文件描述符发生了什么事情(如读事件、写事件等)
2.addfd
表示将listenfd加入epollfd中,这是一个自定义函数
epoll轮询
int number = epoll_wait(epollfd, events, MAX_EVENT_NUMBER,-1);
如果没有数据发送给服务器,将一直堵塞再此处
事件处理
有新客户端连接
// 有新的客户端连接
if(sockfd == listenfd)
{
struct sockaddr_in client_address;
socklen_t client_addresslen = sizeof(client_address);
int connfd = accept(listenfd, (struct sockaddr*)&client_address, &client_addresslen);
if(connfd < 0)
{
printf("errno is %d\n", errno);
continue;
}
if(http_conn::m_user_count >= MAX_FD)
{
close(connfd);
continue;
}
users[connfd].init(connfd, client_address);
}
可读
// 有读事件发生(可读)
else if(events[i].events & EPOLLIN)
{
// 有读事件发生
if(users[sockfd].read())
{
// 读的到数据
pool->append(users+sockfd);
}
else
{
// 读不到数据
users[sockfd].close_conn();
}
}
可写
// 有写事件发生(可写)
else if(events[i].events & EPOLLOUT)
{
if(!users[sockfd].write())
{
users[sockfd].close_conn();
}
}
结束
close(epollfd);
close(listenfd);
delete [] users;
delete pool;
return 0;
关闭所有的fd
完整代码
#include <arpa/inet.h>
#include <stdio.h>
#include <unistd.h>
#include <errno.h>
#include <string.h>
#include <fcntl.h>
#include <stdlib.h>
#include <sys/epoll.h>
#include "threadpool.hpp"
#include "locker.h"
#include "http_conn.h"
#include <signal.h>
#include <assert.h>
#define MAX_FD 65536 // 最大的文件描述符
#define MAX_EVENT_NUMBER 10000 // 监听的最大事件数
/*
函数指针的声明: 类型说明符 (*函数名) (参数)
void(handler)(int) 声明了一个名为 handler 的函数指针,它指向一个接受一个 int 参数并返回 void 的函数
*/
void addsig(int sig, void(handler)(int))
{
// sigaction的输入参数
struct sigaction sa;
// 指定sa内存区域的前n个字节都设置为某个特定的值('\0'),用于对新分配的内存进行初始化
memset(&sa, '\0', sizeof(sa));
// 写入函数指针,指向的函数就是信号捕捉到之后的处理函数
sa.sa_handler = handler;
// 设置临时阻塞信号集
sigfillset(&sa.sa_mask);
assert(sigaction(sig, &sa, NULL) != -1);
}
int main(int argc, char* argv[])
{
if(argc <= 1)
{
// 要求输入格式为 ./a.out 10000 其中10000是端口号
printf("usage: %s port_number\n", basename(argv[0]));
return 1;
}
// 端口号 string -> int
int port = atoi(argv[1]);
// 如果向一个没有读端的管道写数据,不用终止进程
addsig(SIGPIPE, SIG_IGN); // SIG_IGN: 忽略信号,这里指的是忽略信号 · SIGPIPE
// 定义一个线程池指针
threadpool<http_conn>* pool = NULL;
try {
// 开辟一个线程池
pool = new threadpool<http_conn>;
}catch(...)
{
// 若异常则退出
return 1;
}
// 开辟一块连续的http_conn数组,保存所有正在连接的客户端信息
http_conn* users = new http_conn[MAX_FD];
// 设置监听
int listenfd = socket(AF_INET, SOCK_STREAM, 0);
int ret = 0;
struct sockaddr_in address;
address.sin_addr.s_addr = INADDR_ANY;
address.sin_family = AF_INET;
address.sin_port = htons(port);
// 设置端口复用
int reuse = 1;
setsockopt(listenfd, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse));
// 绑定
ret = bind(listenfd, (struct sockaddr*)&address, sizeof(address));
if(ret == -1)
{
perror("bind");
exit(-1);
}
// 开始监听
ret = listen(listenfd, 5);
if(ret == -1)
{
perror("listen");
exit(-1);
}
// 将listend添加到epoll模型中
epoll_event events[MAX_EVENT_NUMBER];
int epollfd = epoll_create(5);
addfd(epollfd, listenfd, false);
http_conn::m_epollfd = epollfd;
while(1)
{
// epoll轮询,等待有数据发送
int number = epoll_wait(epollfd, events, MAX_EVENT_NUMBER,-1);
if((number < 0) && (errno != EINTR))
{
printf("epoll failture\n");
break;
}
for(int i = 0; i < number; i++)
{
int sockfd = events[i].data.fd;
// 有新的客户端连接
if(sockfd == listenfd)
{
struct sockaddr_in client_address;
socklen_t client_addresslen = sizeof(client_address);
int connfd = accept(listenfd, (struct sockaddr*)&client_address, &client_addresslen);
if(connfd < 0)
{
printf("errno is %d\n", errno);
continue;
}
if(http_conn::m_user_count >= MAX_FD)
{
close(connfd);
continue;
}
users[connfd].init(connfd, client_address);
}
// 若对方异常端开或错误
else if(events[i].events & (EPOLLRDHUP | EPOLLHUP | EPOLLERR))
{
users[sockfd].close_conn();
}
// 有读事件发生(可读)
else if(events[i].events & EPOLLIN)
{
// 有读事件发生
if(users[sockfd].read())
{
// 读的到数据
pool->append(users+sockfd);
}
else
{
// 读不到数据
users[sockfd].close_conn();
}
}
// 有写事件发生(可写)
else if(events[i].events & EPOLLOUT)
{
if(!users[sockfd].write())
{
users[sockfd].close_conn();
}
}
}
}
close(epollfd);
close(listenfd);
delete [] users;
delete pool;
return 0;
}