【项目篇】高并发服务器 - 从 Buffer 到 TcpServer 构建高并发服务器引擎

文章目录

• • Buffer模块：缓冲区模块
  • Socket模块：套接字模块
  • Channel模块：事件管理模块
  • Poller模块：描述符IO事件监控模块
  • EventLoop模块：事件监控以及处理的模块
  • TimerWheel定时器模块
  • Connection模块:对通信连接进行整体管理的模块
  • Acceptor模块：对监听套接字进行管理
  • LoopThread模块
  • LoopThreadPoll模块
  • TcpServer模块

Buffer模块：缓冲区模块

功能：存入数据，提取数据 实现思想： （1）实现缓冲得有一块内存空间，采用vector来实现。vector底层使用的就是一块线性的内存空间。 （2）要素： ① 默认的空间大小 ②当前的读取数据位置 ③当前的写入数据位置

（3）操作： ①写入数据 当前写入位置指向哪里，就从哪里开始写入。 如果当前写入时后面的空间不够了，有两种情况：

1.考虑整体缓冲区空闲位置够不够，读位置将会向后移动，前面的位置有空闲空间。 足够：将数据移动到其实位置即可 不够：扩容，从当前写位置开始扩容足够大小 数据一旦写成功，当前的写位置就要向后偏移

②读取数据 当前的读取位置指向哪里，就从哪里读起。前提是有数据可读 可读数据大小：当前写入位置减去当前读取位置

实现一个缓冲区类该如何设计：

class Buffer
{
private：
vector<char> _buffer;
//位置是一个相对偏移量，而不是绝对地址
uint64It _read_idx; //相对读偏移量
uint64It _write_idx; //相对写偏移量
public：
//1.获取当前写位置地址
//2.确保可写位置足够（足够移动，不够扩容）
//3.获取前沿空闲空间大小
//4.获取后沿空闲空间大小（可读位置以前的空间）
//5.将写位置向后移动指定长度
//6.获取当前读位置的地址
//7.获取可读数据大小
//8.将读位置向后移动指定长度
//9.清理功能
};

Buffer模块的代码设计：

#include <iostream>
#include <cstring>
#include <vector>
#include <cassert>

#define BUFFER_DAFAULT_SIZE 1024
class Buffer
{
private:
std::vector<char> _buffer; //使用vector进行内存空间管理
uint64_t _read_idx; //读偏移
uint64_t _write_idx; //写偏移
public:
Buffer():_read_idx(0),_write_idx(0),_buffer(BUFFER_DAFAULT_SIZE)
{}
char *Begin(){return &*_buffer.begin();} //获取起始元素的迭代器再对迭代器进行解引用，再取地址（&）就是第0个元素的空间地址

//1.获取当前写入位置的地址,//buffer空间起始地址 + 写偏移量
char *WritePosition(){return Begin() + _write_idx;}
//2.获取当前读取数据位置的地址
char *ReadPosition(){return Begin() + _read_idx;}
//3.获取缓冲区末尾空闲空间大小—写偏移之后的空闲空间 总体空间大小 – 写偏移
uint64_t TailIdleSize(){return _buffer.size() – _write_idx;}
//4.获取缓冲区起始空闲空间大小—读偏移之前的空闲空间大小
uint64_t HeadIdleSize(){return _read_idx;}
//5.获取可读数据大小
uint64_t ReadAbleSize(){return _write_idx – _read_idx;}
//6.将读偏移向后移动
void MoveReadOffset(uint64_t len)
{
//向后移动大小必须小于可读数据大小
assert(len <= ReadAbleSize());
_read_idx += len;}
//7.将写偏移向后移动
void MoveWriteOffset(uint64_t len)
{
//向后移动大小必须小于当前后边空间的大小
assert(len <= TailIdleSize());
_write_idx += len;
}
//8.确保可写位置足够（足够移动，不够扩容）
void EnsureWriteSpace(uint64_t len)
{
//如果末尾空闲空间大小足够，直接返回
if(len <= TailIdleSize())
{
return;
}
//如果末尾空间不够，判断加上起始位置空闲空间大小看够不够,够了将数据移动到起始位置
if(len <= TailIdleSize() + HeadIdleSize())
{
//将数据移动到起始位置
uint64_t rsz = ReadAbleSize();//将当前可读数据大小先保存起来
std::copy(ReadPosition(),ReadPosition() + rsz,Begin());//把可读数据拷贝到起始位置
_read_idx = 0; //将读位置置0
_write_idx = rsz; //将写位置置为可读数据大小，因为当前的可读数据大小就是当前的写偏移
}
else
{
//总体空间不够，则需要扩容
//不移动数据，直接给写偏移之后扩容足够空间
_buffer.resize(_write_idx + len);
}
}
//9.写数据
void Write(const void* data,uint64_t len)
{
//1.保证有足够空间 2.拷贝数据进去
const char *d = (const char*)data;
EnsureWriteSpace(len);
std::copy(d,d+len,WritePosition());
}
void WriteAndPush(const void *data,uint64_t len)
{
Write(data,len);
MoveWriteOffset(len);
}
//写入string类型的数据
void WriteString(const std::string &data)
{
return Write(data.c_str(),data.size());
}
void WriteStringAndPush(const std::string &data)
{
Write(data.c_str(),data.size());
MoveWriteOffset(data.size());
}
//写入Buffer类型数据
void WriteBuffer(Buffer &data)
{
return Write(data.ReadPosition(),data.ReadAbleSize());
}
void WriteBufferAndPush(Buffer &data)
{
WriteBuffer(data);
MoveWriteOffset(data.ReadAbleSize());
}
//10.读数据
void Read(void* buf,uint64_t len)
{
//获取数据大小必须小于可读数据大小
assert(len <= ReadAbleSize());
std::copy(ReadPosition(),ReadPosition() + len,(char*)buf);
}
//将数据读取并弹出
void ReadAndPop(void *buf,uint64_t len)
{
Read(buf,len);
MoveReadOffset(len);
}
//以string的方式读取数据
std::string ReadAsString(uint64_t len)
{
assert(len <= ReadAbleSize());
std::string str;
str.resize(len);
Read(&str[0],len);
return str;
}
//以string的方式读取数据并弹出
std::string ReadAsStringAndPop(uint64_t len)
{
std::string str = ReadAsString(len);
MoveReadOffset(len);
return str;
}
//查找换行符
char *FindCRLF()
{
void *res = memchr(ReadPosition(),'\\n',ReadAbleSize());
return (char*)res;
}
//通常获取一行数据，这种情况针对的是asscii码值
std::string GetLine()
{
char *pos = FindCRLF();
if(pos == nullptr)
{
return "";
}
//加1是为了把换行字符也提取出来
return ReadAsString(pos – ReadPosition() + 1);
}
std::string GetLineAndPop()
{
std::string str = GetLine();
MoveReadOffset(str.size());
return str;
}
//11.清空缓冲区
void Clean()
{
//只需要将偏移量归0即可
_read_idx = 0;
_write_idx = 0;
}
};

日志打印宏的编写

#define INF 0
#define DBG 1
#define ERR 2
#define LOG_LEVEL INF
#define LOG(level,format,...)do{\\
if(level < LOG_LEVEL) break;\\
time_t t = time(nullptr);\\
struct tm *ltm = localtime(&t);\\
char tmp[32] = {0};\\
strftime(tmp,31,"%H:%M:%S",ltm);\\
fprintf(stdout,"[%s %s:%d]" format "\\n",tmp,__FILE__,__LINE__,##__VA_ARGS__);\\
}while(0)
#define INF_LOG(format,...) LOG(INF,format,##__VA_ARGS__)
#define DBG_LOG(format,...) LOG(DBG,format,##__VA_ARGS__)
#define ERR_LOG(format,...) LOG(ERR,format,##__VA_ARGS__)

上面的代码做了三层封装：

①定义日志等级

#define INF 0
#define DBG 1
#define ERR 2
#define LOG_LEVEL INF

含义：

• INF:普通信息
• DBG:调试信息
• ERR:错误信息
• LOG_LEVEL:当前日志输出阈值

②核心日志宏

#define LOG(level,format,...)do{\\
if(level < LOG_LEVEL) break; \\
...打印日志... \\
}while(0)

核心作用：根据等级决定打不打印日志

③二次封装

#define INF_LOG(format,...)
LOG(INF,format,##__VA_ARGS__)
#define DBG_LOG(format,...)
LOG(DBG,format,##__VA_ARGS__)
#define ERR_LOG(format,...)
LOG(ERR,format,##__VA_ARGS__)

总结：

• 用宏实现类似printf的日志接口
• 用__FILE__和__LINE__定位代码位置
• 用LOG_LEVEL控制日志的输出等级
• 用二次宏（INF_LOG /DBG_LOG）简化调用
• 用 ##__VA_ARGS__解决空参数问题
• 用do while(0)保证宏安全

Socket模块：套接字模块

Socket套接字类功能设计

Socket.hpp

class Socket
{
private:
int _sockfd;
public:
Socket():_sockfd(–1){}
//通过描述符构建socket对象
Socket(int fd):_sockfd(fd){}
~Socket(){Close();}
int Fd(){return _sockfd;}
//1.创建套接字
bool Creat()
{
//int socket(int domain,int type,int protocol)
int sockfd = socket(AF_INET,SOCK_STREAM,0);
if(sockfd < 0)
{
ERR_LOG("CREAT SOCKET FAILED!!!");
return false;
}
_sockfd = sockfd;
return true;
}
//2.绑定地址信息
bool Bind(const std::string &_ip,uint64_t port)
{
//组织一个IPV4的地址结构
struct sockaddr_in addr;
addr.sin_family = AF_INET;
addr.sin_port = htons(port);
addr.sin_addr.s_addr = inet_addr(_ip.c_str());
socklen_t len = sizeof(struct sockaddr_in);

//int bind(int sockfd,struct sockaddr* addr,socklen_t len)
int ret = bind(_sockfd,(struct sockaddr*)&addr,len);
if(ret < 0)
{
ERR_LOG("BIND ADDRESS FAILED");
return false;
}
return true;
}
//3.开始监听
bool Listen(int backlog = MAX_LISTEN)
{
int ret = listen(_sockfd,backlog);
if(ret < 0)
{
ERR_LOG("SOCKET LISTEN FAILED!");
return false;
}
return true;
}
//4.向服务器发起连接
bool Connect(const std::string &ip,uint64_t port)
{
//int connect(int sockfd,struct sockaddr* addr,socklen_t len);
struct sockaddr_in addr;
addr.sin_family = AF_INET;
addr.sin_port = htons(port);
addr.sin_addr.s_addr = inet_addr(ip.c_str());
socklen_t len = sizeof(struct sockaddr_in);

int ret = connect(_sockfd,(struct sockaddr*)&addr,len);
if(ret < 0)
{
ERR_LOG("CONNECT SERVER FAILED!");
return false;
}
return true;
}
//5.获取新连接
int Accept()
{
//int accept(int sockfd,struct sockaddr* addr,socklen_t *len);
int newfd = accept(_sockfd,nullptr,nullptr);
if(newfd < 0)
{
// 处理非阻塞模式下的EAGAIN错误
if(errno == EAGAIN || errno == EINTR)
{
return –2; // 特殊返回值，表示需要重试
}
ERR_LOG("SOCKET ACCEPT FAILED!");
return –1;
}
return newfd;
}
//6.接收数据 flag为标志位，表示非阻塞
ssize_t Recv(void* buf,size_t len,int flag = 0)
{
//ssize_t recv(int sockfd,void *buf,size_t len.int flage)
ssize_t ret = recv(_sockfd,buf,len,flag);
if(ret <= 0)
{
//EAGAIN表示当前socket的接收缓冲区没有数据了，只要在非阻塞的情况下才有这个错误
//EINTR表示当前socket的阻塞等待，被信号打断了
if(errno == EAGAIN || errno == EINTR)
{
return 0;//表示这次接收没有接收到数据
}
ERR_LOG("SOCKET RECV FAILED!");
return –1;
}
return ret;//实际接收到的数据长度
}

//非阻塞接收
ssize_t NonBlockRecv(void *buf,size_t len)
{
return Recv(buf,len,MSG_DONTWAIT); //MSG_DONTWAIT是一个非阻塞标志，表示当前接收是一个非阻塞
}
//7.发送数据
ssize_t Send(const void *buf,size_t len,int flag = 0)
{
//ssize_t send(int sockfd,void*data,size_t len,int flag)
ssize_t ret = send(_sockfd,buf,len,flag);
if(ret < 0)
{
// 处理非阻塞模式下的EAGAIN错误
if(errno == EAGAIN || errno == EINTR)
{
return 0; // 表示这次发送没有发送数据
}
ERR_LOG("SOCKET SEND FAILED!");
return –1;
}
return ret;//实际发送的数据长度
}
//非阻塞发送
ssize_t NonBlockSend(void *buf,size_t len)
{
if(len == 0) return 0;
return Send(buf,len,MSG_DONTWAIT);//MSG_DONTWAIT表示非阻塞发送
}
//8.关闭套接字
void Close()
{
if(_sockfd != –1)
{
close(_sockfd);
_sockfd = –1;
}
}
//9.创建一个服务端连接
bool CreatServer(uint64_t port,const std::string &ip = "0.0.0.0",bool block_flag = false)
{
//1.创建套接字
if(Creat() == false) return false;
if(block_flag) NonBlock();
//2.绑定地址
if(Bind(ip,port) == false) return false;
//3.开始监听
if(Listen() == false) return false;
//4.设置非阻塞
//NonBlock();
//5.启动地址复用
ReuseAddress();

return true;
}
//10.创建一个客户端连接
bool CreateClient(uint64_t port,const std::string &ip)
{
//1.创建套接字
if(Creat() == false) return false;
//2.连接服务器
if(Connect(ip,port) == false) return false;

return true;
}
//11.创建套接字选项，开启地址端口复用
void ReuseAddress()
{
//int setsockopt(int fd, int level,int optname,int void* value,int vallen)
int val = 1;
setsockopt(_sockfd,SOL_SOCKET,SO_REUSEADDR,(void*)&val,sizeof(int));
val = 1;
setsockopt(_sockfd,SOL_SOCKET,SO_REUSEPORT,(void*)&val,sizeof(int));
}
//12.设置套接字阻塞属性 — 设置为非阻塞
void NonBlock()
{
//int fcntl(int fd,int cmd, …/* arg */);
int flag = fcntl(_sockfd,F_GETFL,0);
fcntl(_sockfd,F_SETFL,flag | O_NONBLOCK);
}
};

Channel模块：事件管理模块

Channel类的设计

目的： 对描述符的监控事件管理

功能： （1）事件管理：

① 描述符是否可读 ② 描述符是否可写 ③ 对描述符监控可读 ④ 对描述符监控可写 ⑤ 解除可读事件监控 ⑥ 解除可写事件监控 ⑦ 解除所有事件监控

（2）事件触发后的处理管理

① 需要处理的事件：可读事件、可写事件、挂断事件、错误事件、任意事件 ② 事件处理的回调函数 Channel类的接口定义：

class Channel
{
private:
uint32_t _events; //当前需要监控的事件
uint32_t _revent; //当前连接触发的事件
using EventCallback = std::function<void()>;
EventCallback _read_callback; //可读事件触发后的回调
EventCallback _write_callback; //可写事件触发后的回调
EventCallback _err_callback; //错误事件触发后的回调
EventCallback _close_callback; //连接断开事件被触发后的回调
EventCallback _event_callback; //任意事件被触发后的回调

public:
Channel(){}

void SetReadCallback(const EventCallback &cb){}
void SetWrtieCallback(const EventCallback &cb){}
void SetErrorCallback(const EventCallback &cb){}
void SetCloseCallback(const EventCallback &cb){}
void SetEventCallback(const EventCallback &cb){}

bool ReadAble(){} //当前是否可读
bool WriteAble(){} //当前是否可写
void EnableRead(){} //启动读事件监控
void EnableWrite(){} //启动写事件监控
void DisableRead(){} //关闭读事件监控
void DisableWrtie(){} //关闭写事件监控
void DisableAll(){} //关闭所有事件监控
void Remove(){} //移除监控
void HandleEvent(){} //事件处理,一旦连接触发了事件，就调用这个函数，自己触发了什么事件，如何处理自己决定
};

Channel类的接口实现：

class Channel
{
private:
int _fd;
EventLoop *_loop;
uint32_t _events; //当前需要监控的事件
uint32_t _revents; //当前连接触发的事件
using EventCallback = std::function<void()>;
EventCallback _read_callback; //可读事件触发后的回调
EventCallback _write_callback; //可写事件触发后的回调
EventCallback _error_callback; //错误事件触发后的回调
EventCallback _close_callback; //连接断开事件被触发后的回调
EventCallback _event_callback; //任意事件被触发后的回调

public:
Channel(EventLoop *loop,int fd):_fd(fd),_events(0),_revents(0),_loop(loop){}
int Fd(){return _fd;}
uint32_t Events(){return _events;} //获取想要监控的事件
void SetRevents(uint32_t events){_revents = events;} //获取实际就绪的事件
void SetReadCallback(const EventCallback &cb){_read_callback = cb;}
void SetWriteCallback(const EventCallback &cb){_write_callback = cb;}
void SetErrorCallback(const EventCallback &cb){_error_callback = cb;}
void SetCloseCallback(const EventCallback &cb){_close_callback = cb;}
void SetEventCallback(const EventCallback &cb){_event_callback = cb;}

//是否监控了可读
bool ReadAble(){ return _events & EPOLLIN;}
//是否监控了可写
bool WriteAble(){ return _events & EPOLLOUT;}
//启动读事件监控
void EnableRead(){ _events |= EPOLLIN;Update();}
//启动写事件监控
void EnableWrite(){ _events |= EPOLLOUT;Update();}
//关闭读事件监控
void DisableRead(){ _events &= ~EPOLLIN;Update();}
//关闭写事件监控
void DisableWrite(){_events &= ~EPOLLOUT;Update();}
//关闭所有事件监控
void DisableAll(){_events = 0;Update();}
//移除监控
void Remove();
void Update();
//事件处理,一旦连接触发了事件，就调用这个函数，自己触发了什么事件，如何处理自己决定
void HandleEvent()
{
if((_revents & EPOLLIN) || (_revents & EPOLLRDHUP) /*连接断开自动触发可读事件*/ || (_revents & EPOLLPRI)/*优先带外数据*/)
{
if(_read_callback) _read_callback();
}
if(_revents & EPOLLOUT)
{
if(_write_callback) _write_callback();
} //有可能会释放连接的操作，一次只处理一个
else if(_revents & EPOLLERR)
{
if(_error_callback) _error_callback(); //一旦出错就会释放连接，没必要调用任意回调了，因此要放到前面调用任意事件回调
}
else if(_revents & EPOLLHUP)
{
if(_close_callback) _close_callback();
}
if(_event_callback) _event_callback(); //放到事件处理完毕后调用，刷新活跃度
}
};

Poller模块：描述符IO事件监控模块

意义： 通过epoll实现对描述符的IO事件监控

功能： （1）添加 / 修改描述符的事件监控 （不存在则添加，存在则更新） （2）移除描述符的事件监控

封装思想： （1）必须拥有一个epoll的操作句柄 （2）拥有一个 struct epoll_event结构数组,监控时保存所有的活跃事件 （3）使用一个哈希表，管理描述符和描述符对应的事件管理Channel对象 逻辑流程： ①对描述符进行监控，通过Channel才能知道描述符需要监控什么事件 ②当描述符就绪了，通过描述符在哈希表中给找到对应的Channel（得到了Channel才能知道什么事件该如何处理），当描述符就绪了返回就绪描述符对应的Channel

class Poller
{
private:
int _epfd;
struct epoll_event _evs[xxx];
std::unordered_map<int,Channel*>
private:
1.判断要更新事件的描述符是否存在
2.针对epoll直接操作（添加，修改，移除）
public：
1.添加或更新描述符所监控的事件
2.移除描述符的监控
3.开始监控，获取就绪的Channel
};

Poller模块类接口实现：

#define MAX_EPOLLEVENTS 1024
class Poller
{
private:
int _epfd;
struct epoll_event _evs[MAX_EPOLLEVENTS];
std::unordered_map<int,Channel *> _channels;
private:
//对epoll的直接操作
void Update(Channel* channel,int op)
{
//int epoll_ctll(int epfd,int op,int fd,struct_event *event)
int fd = channel->Fd();
struct epoll_event ev;
ev.data.fd = fd;
ev.events = channel->Events();
int ret = epoll_ctl(_epfd,op,fd,&ev);
if(ret < 0)
{
ERR_LOG("EPOLLCTL FAILED!");
}
return;
}
//判断一个Channel是否添加了事件监控
bool HasChannel(Channel* channel)
{
auto it = _channels.find(channel->Fd());
if(it == _channels.end())
{
return false;
}
return true;
}
public:
Poller()
{
_epfd = epoll_create(MAX_EPOLLEVENTS);
if(_epfd < 0)
{
ERR_LOG("EPOLLCREAT FAILED!");
abort(); //退出程序
}
}
//添加或者修改事件监控
void UpdateEvent(Channel* channel)
{
bool ret = HasChannel(channel);
if(ret == false)
{
//不存在则添加
_channels.insert(std::make_pair(channel->Fd(),channel));
return Update(channel,EPOLL_CTL_ADD);
}
//存在则修改
return Update(channel,EPOLL_CTL_MOD);
}
//移除监控
void RemoveEvent(Channel* channel)
{
auto it = _channels.find(channel->Fd());
if(it != _channels.end())
{
_channels.erase(it);
}
Update(channel,EPOLL_CTL_DEL);
}
//开始监控，返回活跃连接
void Poll(std::vector<Channel*> *active)
{
//int epoll_wait(int epfd,struct epoll_event *events,int maxevents,int timeout);
int nfds = epoll_wait(_epfd ,_evs,MAX_EPOLLEVENTS,–1);//-1表示阻塞式监控 nfds表示就绪的事件个数
if(nfds < 0)
{
if(errno == EINTR) //表示阻塞被信号打断了
{
return;
}
ERR_LOG("EPOLL WAIT ERROR:%s\\n",strerror(errno));
abort();//退出程序
}
for(int i = 0; i < nfds; i++)
{
auto it = _channels.find(_evs[i].data.fd);
assert(it != _channels.end());
it->second->SetRevents(_evs[i].events); //设置实际就绪事件
active->push_back(it->second);
}
return;
}
};

void Channel::Remove(){ return _poller->RemoveEvent(this);}
void Channel::Update(){ return _poller->UpdateEvent(this);}

eventfd 介绍 eventfd：创建一个描述符用于事件的通知

区分信号。信号是对进程进行事件通知的

eventfd本质在内核里面管理的就是一个计数器，创建evnetfd就会在内核中创建一个计数器，每当向eventfd中写入一个数值—用于表示事件通知次数，可以使用read进行数据读取，读取到的数据就是通知的次数。读取之后，计数清零。

用处： 在EventLoop模块实现事件通知功能。

#include <sys/eventfd.h>
int eventfd(unsigned int intval,int flags);
功能：创建一个eventfd对象，实现事件的通知
参数：
intval:计数初值
flags：
EFD_ELOEXEC – 禁止进程复制
EFD_NONBLOCK – 启动非阻塞属性
返回值：返回一个文件描述符用于操作
eventfd也是通过read/write/close进行操作的
注意点：read&write进行IO的时候，数据只能是一个8字节数据

优点：

（1）相比于pipe更轻量：

• 开销更低：pipe需要维护两个文件描述符和一个内核缓冲区，而eventfd只需要一个文件描述符和一个简单的64位整数计数器。
• 资源占用少：每一个eventfd只消耗一个文件描述符，这在需要大量通知通道场景下优势明显。

（2）原子性操作：eventfd的读写操作都是原子性的，这意味着多个线程可以并发的读写，而不需要额外的锁保护。

EventLoop模块：事件监控以及处理的模块

关键点：这个模块与线程是一一对应关联的。 监控了一个连接，而这个连接一旦就绪，就要进行事件处理。如果这个描述符在多个事件中都触发了事件，进行处理，就会存在线程安全问题。因此我们需要将一个连接的事件监控，以及连接事件处理，以及其他操作，都放在同一个线程中进行。

如何保证对一个连接的所有操作都在EventLoop线程中？ 解决方案：给EventLoop中添加一个队列，对连接的所有操作，都进行一次封装，将对连接的操作并不直接执行，而是当作任务添加到任务的队列中。这样就能保证对连接的所有操作，都是在一个线程中执行的，不涉及线程安全问题。

但是对任务队列的操作有线程安全问题。只需要给task的操作加一把锁。 EventLoop处理流程： ①在线程中，对描述符进行事件监控 ②有描述符就绪，则对描述符进行事件处理 ③所有的就绪事件处理完了，这时候再将任务队列中的任务一一执行。

功能： （1）事件监控 ①使用poller模块 ②有事件就绪则进行事件处理

（2）执行任务队列中的任务 ①一个线程的任务队列

注意点：因为有可能因为等待描述符IO事件就绪，导致执行流程阻塞，这时候任务队列中的任务将得不到执行。 因此得有一个任务通知的东西，能够唤醒事件通知的阻塞。

当时间就绪，需要处理的时候，在处理的过程中，如果要对连接进行某些操作，这些操作都应该保证在EventLoop对应的线程中执行，保证对连接的各项操作都是线程安全的。

EventLoop模块代码实现：

class EventLoop
{
private:
using Functor = std::function<void()>;
std::thread::id _thread_id; //线程ID
int _event_fd; //event_fd唤醒事件监控有可能导致的阻塞
std::unique_ptr<Channel> _event_channel; //当EventLoop释放的时候，unique_ptr也会释放
Poller _poller; //进行所有描述符的事件监控
std::vector<Functor> _tasks; //任务池
std::mutex _mutex; //任务池操作的线程安全
TimerWheel _timer_wheel; //定时器模块
public:
//执行任务池中的所有任务
void RunAllTask()
{
std::vector<Functor> functor;
{
//加锁
std::unique_lock<std::mutex> _lock(_mutex);//出了作用域就会被销毁，自动解锁
_tasks.swap(functor);
}
for(auto &f : functor)
{
f();
}
return;
}
static int CreateEventFd()
{
int efd = eventfd(0,EFD_CLOEXEC | EFD_NONBLOCK);
if(efd < 0)
{
ERR_LOG("CTREATE EVENTFD FAILED !");
abort();//让程序异常退出
}
return efd;
}
void ReadEventFd()
{
uint64_t res = 0;
int ret = read(_event_fd,&res,sizeof(res));
if(ret < 0)
{
//EINTR – 表示被信号打断 EAGAIN – 表示无数据可读
if(errno == EINTR || errno == EAGAIN)
{
return;
}
ERR_LOG("READ EVENTFD FAILED!");
abort();//异常退出
}
return;
}
void WeakUpEventFd()
{
uint64_t val = 1;
int ret = write(_event_fd,&val,sizeof(val));
if(ret < 0)
{
if(errno == EINTR)
{
return;
}
ERR_LOG("WRITE EVENTFD FAILED!");
abort();//异常退出
}
return;
}
public:
EventLoop()
:_thread_id(std::this_thread::get_id())
,_event_fd(CreateEventFd())
,_event_channel(new Channel(this,_event_fd))
,_timer_wheel(this)
{
//给eventfd添加可读事件回调函数，读取eventfd事件通知次数
_event_channel->SetReadCallback(std::bind(&EventLoop::ReadEventFd,this));
//启动eventfd读事件监控
_event_channel->EnableRead();
}
//启动EventLoop模块 三步走：事件监控 -> 就绪事件处理 -> 执行任务
void Start()
{
while(1)
{
//1.事件监控
std::vector<Channel*> actives; //活跃连接
_poller.Poll(&actives);
//2.事件处理
for(auto &channel :actives)
{
channel->HandleEvent();//进行事件回调
}
//3.任务执行
RunAllTask();//执行任务池中的任务
}
}
//用于判断当前线程是否是EventLoop对应的线程
bool IsInLoop()
{
return (_thread_id/*当前EventLoop的线程id*/ == std::this_thread::get_id()); /*获取当前运行线程的id*/
}
void AssertInLoop()
{
assert(_thread_id == std::this_thread::get_id());
}
//判断当前执行的任务是否在当前线程中，如果在就执行，如果不在，就加入到任务队列中
void RunInLoop(const Functor &cb)
{
if(IsInLoop())
{
return cb();
}
return QueueInLoop(cb);
}
//将操作压入任务池
void QueueInLoop(const Functor &cb)
{
{
std::unique_lock<std::mutex> _lock(_mutex);
_tasks.push_back(cb);
}
//唤醒有可能因为没有事件就绪而导致的epoll阻塞
//其实就是给eventfd写入一个数据，eventfd就会触发可读事件
WeakUpEventFd();
}
//添加或修改描述符的事件监控
void UpdateEvent(Channel* channel){ return _poller.UpdateEvent(channel);}
//移除描述符的事件监控
void RemoveEvent(Channel* channel){ return _poller.RemoveEvent(channel);}
//添加定时任务
void TimerAdd(uint64_t id,uint32_t delay,const TaskFunc &cb){return _timer_wheel.TimerAdd(id,delay,cb);}
//刷新定时任务
void TimerRefresh(uint64_t id){return _timer_wheel.TimerRefresh(id);}
//取消定时任务
void TimerCancel(uint64_t id){return _timer_wheel.TimerCancel(id);}
//是否有某个定时器任务的村子
bool HasTimer(uint64_t id){return _timer_wheel.HasTimer(id);}
};

TimerWheel定时器模块

定时器模块的整合： timefd:实现内核每隔一段时间，给进程一次超时事件（timefd可读事件）。 timerwheel:实现每次执行Runtimetask,都可以执行一波到期的定时任务。

要实现一个完整的秒级定时器，就要将这两个功能整合到一起。 timefd设置为每秒钟触发一次定时事件，当事件被触发，则运行一次timerwheeld的Runtimetask，执行一下所有的过期定时任务。

而timerfd的事件监控与触发，可以融合EventLoop来实现。

using TaskFunc = std::function<void()>; //定时器要执行的任务函数
using ReleaseFunc = std::function<void()>;
//定时器任务对象
class TimerTask
{
private:
uint64_t _id; //定时器任务对象id
uint32_t _timeout; //定时任务的超时时间
bool _canceled; //定时器任务是否取消 false-表示不取消
TaskFunc _task_cb; //定时器要执行的定时任务
ReleaseFunc _release; //用于删除TimerWheel中保存的定时器对象信息，删除_timers
public:
TimerTask(uint64_t id,uint32_t delay,const TaskFunc &cb)
:_id(id)
,_timeout(delay)
,_canceled(false)
,_task_cb(cb)
{}

~TimerTask()
{
//执行任务
if(_canceled == false) _task_cb();
_release();
}
void Cancel(){ _canceled = true;}
void SetRelease(const ReleaseFunc &cb) { _release = cb;}
uint32_t DelayTime(){return _timeout;}

};
//时间轮
class TimerWheel
{
private:
using WeakTask = std::weak_ptr<TimerTask>;
using PtrTask = std::shared_ptr<TimerTask>;
int _tick; //当前的秒针走到哪里释放哪里，释放哪里就相当于执行任务
int _capacity; //表盘的最大数量—-其实就是最大的延迟时间
std::vector<std::vector<PtrTask>> _wheel; //存shared_ptr(决定生死)
std::unordered_map<uint64_t,WeakTask> _timers; //存weak_ptr(只是索引)

EventLoop *_loop;
int _timerfd;//定时器描述符 — 可读事件回调就是读取计数器,执行定时任务
std::unique_ptr<Channel> _timer_channel;
private:
void RemoveTimer(uint64_t id)
{
auto it = _timers.find(id);
if(it != _timers.end())
{
_timers.erase(it);
}
}
static int CreateTimerFd()
{
int timerfd = timerfd_create(CLOCK_MONOTONIC,0);
if(timerfd < 0)
{
ERR_LOG("TIMERFD CREATE FAILED!");
abort();
}
//int timefd_settime(int fd,int flags,struct itimerspec *new,struct itimerspec *old);
struct itimerspec itime;
itime.it_value.tv_sec = 1;
itime.it_value.tv_nsec = 0; //第一次的超时时间为1秒后
itime.it_interval.tv_sec = 1;
itime.it_interval.tv_nsec = 0; //第一次超时后，每次的间隔为1秒
timerfd_settime(timerfd,0,&itime,nullptr);
return timerfd;
}
int ReadTimerFd()
{
uint64_t times;
//有可能因为其他描述符的事件处理事件比较长，然后在处理定时器描述符事件的时候，有可能就超时了很多次
//read读到的数据times就是从上一次read之后超时的次数
int ret = read(_timerfd,&times,8);
if(ret < 0)
{
ERR_LOG("READ TIMERFD FAILED!");
abort();
}
return times;
}
//这个函数应该每秒被执行一次，相当于秒针向后走了一步
void RunTimeTask()
{
_tick = (_tick + 1) % _capacity;
_wheel[_tick].clear();//清空指定位置数组，就会把数组中保存的所有管理定时器对象的shared_ptr释放掉
}
void OnTime()
{
//根据实际的超时次数，执行对应的超时任务
int times = ReadTimerFd();
for(int i = 0;i < times; i++)
{
RunTimeTask();
}
}
//添加定时任务
void TimerAddInLoop(uint64_t id,uint32_t delay,const TaskFunc &cb)
{
PtrTask pt(new TimerTask(id,delay,cb)); //创建任务对象
pt->SetRelease(std::bind(&TimerWheel::RemoveTimer,this,id)); //设置“销毁时清理索引”
int pos = (_tick + delay) % _capacity;
_wheel[pos].push_back(pt); //强引用
_timers[id] = WeakTask(pt); //弱引用
}
//修改定时任务(刷新，延时定时任务)
void TimerRefreshInLoop(uint64_t id)
{
//通过保存的定时器对象的weak_ptr构造一个shard_ptr出来，添加到轮子中
auto it = _timers.find(id);
if(it == _timers.end())
{
return;//没找到定时任务，没法刷新，没法延迟
}
PtrTask pt = it->second.lock(); ///lock获取weak_ptr管理的对象对应的shared_ptr,把weak_ptr转变为shared_ptr
if(!pt) return;
int delay = pt->DelayTime();
int pos = (_tick + delay) % _capacity;
_wheel[pos].push_back(pt);
}
//取消定时任务
void TimerCancelInLoop(uint32_t id)
{
auto it = _timers.find(id);
if(it == _timers.end())
{
return;
}
PtrTask pt = it->second.lock();
if(pt) pt->Cancel();
}
public:
TimerWheel(EventLoop *loop)
:_capacity(60)
,_tick(0)
,_wheel(_capacity)
,_loop(loop)
,_timerfd(CreateTimerFd())
,_timer_channel(new Channel(_loop,_timerfd))
{
_timer_channel->SetReadCallback(std::bind(&TimerWheel::OnTime,this));
_timer_channel->EnableRead(); //启动读事件监控
}
/*定时器中有个_timers成员，定时器信息的操作有可能在多个线程中进行，因此需要考虑线程安全问题*/
/*不想加锁，那就把对定时器的所有操作都放在一个线程中进行*/
void TimerAdd(uint64_t id,uint32_t delay,const TaskFunc &cb);
//修改定时任务(刷新，延时定时任务)
void TimerRefresh(uint64_t id);
//取消定时任务
void TimerCancel(uint32_t id);
/*这个接口存在线程安全问题 — 这个接口实际上不能被外界使用者调用，只能在模块内，在对应的EventLoop线程内执行*/
bool HasTimer(uint64_t id)
{
auto it = _timers.find(id);
if(it == _timers.end())
{
return false;
}
return true;
}
~TimerWheel()
{}
};

遇到的问题： 定义变量的时候 ，把_wheel定义在了_capacity的前面了，导致了后面初始化_wheel（_capacity)的时候_capacity还没有没初始化。导致的报错。

用gdb进行调试

Connection模块:对通信连接进行整体管理的模块

Connection模块是对于通信连接进行整体管理的模块，对通信连接的所有操作都是通过这么模块的功能实现的。

管理： （1）套接字的管理，能够进行套接字的操作。 （2）连接事件的管理，可读，可写，错误，关闭，任意。 （3）缓冲区的管理，便于Socket数据的接收和发送。 （4）协议上下文的管理，记录请求数据的处理过程。 （5）回调函数的管理

• 因为连接接收到数据之后如何处理需要由用户决定，因此必须有业务处理回调函数。
• 一个连接建立成功之后，该如何处理，要由用户决定，因此要必须有连接建立成功的回调函数
• 一个连接建立关闭前，该如何处理，由用户决定，因此要有关闭连接的回调函数。 任意事件的产生，有没有某些处理，由用户决定，因此也要由任意事件的回调函数

功能： （1）发送数据 — 给用户提供的数据发送接口，并不是真正的发送数据接口，而是把数据放到发送缓冲区，启动写事件监控。 （2）关闭连接 — 提供给用户，并不是真正的关闭接口，因该在实际释放连接之前，看一看输入输出缓冲区有没有数据要处理。 （3）启动非活跃连接的超时销毁功能。 （4）取消给非跃连接的超时销毁功能。 （5）协议切换，一个连接，接收数据之后，如何进行业务处理，取决于上下文，以及数据的业务处理回调函数。

Connection模块是连接的管理模块，对连接的所有操作都是通过这个模块完成的。 场景：对连接操作的时候，但是连接已经释放，导致内存访问错误，最终服务器崩溃。 解决方案：使用智能指针shared_ptr，对Connection对象来进行管理,只要任意一个使用Connection的地方，都保存了一份shared_ptr,其他地方就算是释放了shared_ptr也没关系，因为计数不为0，不会导致Connetion的实际释放。

代码实现：

class Connection;
//DISCONECTED-连接关闭状态 CONNECTING-连接建立成功-待处理状态
//CONNECTED-连接建立完成，各种设置已设置，可以通信的状态
//DISCONNECTING-待关闭状态
typedef enum{ DISCONNECTED,CONNECTING,CONNECTED,DISCONNECTING}Connstatu;
using PtrConnection = std::shared_ptr<Connection>;
class Connection : public std::enable_shared_from_this<Connection>
{
private:
uint64_t _conn_id; //连接的唯一id，便于连接的管理和查找
//uint64_t _time_id; //定时器ID,必须是唯一的，这一块为了简化操作，我们将_conn_id,作为我们的定时器id
int _sockfd; //连接关联的文件描述符
bool _enable_inactive_release; //连接是否启动非活跃连接销毁的判断标志，默认为false
EventLoop *_loop; //连接所关联的EventLoop
Connstatu _statu;
Socket _socket; //套接字操作管理
Channel _channel; //连接的事件管理
Buffer _in_buffer; //输入缓冲区 – 存放从socket中读取的数据
Buffer _out_buffer; //输出缓冲区 – 存放要发送给对端的数据
Any _context; //请求的接收处理上下文

/*这四个回调函数是让服务器模块设置的（其实服务器的处理回调也是组建的使用者来设置的）*/
/*换句话说，这几个回调都是我们的组件使用者使用的*/
//连接成功的回调
using ConnectedCallback = std::function<void(const PtrConnection&)>;
//业务处理的回调
using MessageCallback = std::function<void(const PtrConnection&,Buffer *)>; //对缓冲区的数据进行业务处理
//关闭阶段的处理回调
using ClosedCallback = std::function<void(const PtrConnection&)>;
//任意事件触发的回调
using AnyEventCallback = std::function<void(const PtrConnection&)>;
ConnectedCallback _connected_callback;
MessageCallback _message_callback;
ClosedCallback _closed_callback;
AnyEventCallback _event_callback;
/*组件内的连接关闭回调 — 组件内设置的，服务器组件内会把所有的连接管理起来，一旦某个连接要关闭，就应该从管理的地方移除自己的信息*/
ClosedCallback _server_closed_callback;
private:
/*五个Channel的事件回调函数*/
//描述符触发可读事件后回调的函数，接收socket数据放到接收缓冲区中，然后调用_message_callback进行业务处理
void HandleRead()
{
//1.接收Socket的数据，放到缓冲区
char buf[65536];
ssize_t ret = _socket.NonBlockRecv(buf,65535);//非阻塞接收
if(ret < 0)
{
//出错了，不能直接关闭
return ShutdownInLoop();
}
//rett == 0,表示没有读到数据，并不是连接断开了，连接断开返回的是-1
//将数据放入输入缓冲区，写入之后，将写偏移向后移动
_in_buffer.WriteAndPush(buf,ret);
//2.调用message_callback进行业务处理
if(_in_buffer.ReadAbleSize() > 0)
{
//shared_from_this() – 从当前对象自身获取自身的shared_ptr管理对象
return _message_callback(shared_from_this(),&_in_buffer);
}
}
//描述符触发可写事件后的回调函数，将发送缓冲区中的数据进行发送
void HandleWrite()
{
//_out_buffer中保存的数据，就是要发送的数据
ssize_t ret = _socket.NonBlockSend(_out_buffer.ReadPosition(),_out_buffer.ReadAbleSize());
if(ret < 0)
{
//发送错误就应该关闭连接了，但这时候需要检查一下接收缓冲区中有没有数据要处理
if(_in_buffer.ReadAbleSize() > 0)
{
_message_callback(shared_from_this(),&_in_buffer);
}
return Release();//实际的关闭释放操作
}
_out_buffer.MoveReadOffset(ret);
if(_out_buffer.ReadAbleSize() == 0)
{
_channel.DisableWrite(); //没有数据发送了，关闭写事件监控·
//如果当前是连接待关闭状态，有数据则发送发送完数据关闭连接，没有数据则直接释放
if(_statu == DISCONNECTING)
{
return Release();
}
}
return;
}
//描述符触发挂断事件的回调函数
void HandleClose()
{
/*一旦连接挂断了，套接字什么都干不了，因此有数据待处理就处理一下，完毕关闭连接*/
if(_in_buffer.ReadAbleSize() > 0)
{
_message_callback(shared_from_this(),&_in_buffer);
}
return Release();
}
//描述符触发出错事件
void HandleError()
{
return HandleClose();
}
//描述符触发任意事件 1.刷新连接的活跃度-延迟定时销毁任务 2.调用组件使用者的任意事件回调
void HandleEvent()
{
if(_enable_inactive_release == true){ _loop->TimerRefresh(_conn_id);}
if(_event_callback) { _event_callback(shared_from_this());}
}
//连接获取之后所处的状态下，要进行的各个设置（给Channel设置事件回调，启动读监控）
void EstablishedInLoop()
{
//1.修改连接状态; 2.启动读事件监控; 3.调用回调函数
assert(_statu == CONNECTING); //当前状态必须是上层的半连接状态
_statu = CONNECTED; //这条语句执行完毕，则连接就进入到了已连接状态
_channel.EnableRead();
if(_connected_callback) _connected_callback(shared_from_this());
}
//这个接口才是实际的释放接口
void ReleaseInLoop()
{
// 1.修改连接状态，将其设置为DISCONNECTED
_statu = DISCONNECTED ;
// 2.移除事件监控
_channel.Remove();
// 3.关闭描述符
_socket.Close();
// 4.如果当前任务队列中还有定时销毁任务，则取消任务
if(_loop->HasTimer(_conn_id)) CancelInactiveReleaseInLoop();
// 5.调用关闭回调函数 ，避免先移除服务器管理的连接信息，导致Connection被释放，再去处理会报错，因此先调用用户的回调函数
if(_closed_callback) _closed_callback(shared_from_this());
//移除服务器内部管理的连接信息
if(_server_closed_callback) _server_closed_callback(shared_from_this());
}
//这个接口并不是实际的发送数据接口，而是把数据放到了发送缓冲区 ，启动了可写事件监控
void SendInLoop(Buffer &buf)
{
if(_statu == DISCONNECTED) return; //如果连接是关闭状态，什么事都不做
_out_buffer.WriteBufferAndPush(buf);
if(_channel.WriteAble() == false)
{
_channel.EnableWrite();
}
}
//这个关闭操作并非实际的连接释放操作，需要判断有没有数据待处理，待发送
void ShutdownInLoop()
{
_statu = DISCONNECTING;//设置半关闭状态
if(_in_buffer.ReadAbleSize() > 0)
{
if(_message_callback)
{
_message_callback(shared_from_this(),&_in_buffer);
}
}
//要么是写入数据的时候出错关闭，要么是没有待发送数据直接关闭
if(_out_buffer.ReadAbleSize() > 0)
{
if(_channel.WriteAble() == false)
{
_channel.EnableWrite();
}

}
if(_out_buffer.ReadAbleSize() == 0)
{
Release();
}
}
//启动非活跃连接超时释放规则
void EnableInactiveReleaseInLoop(int sec)
{
//1.将判断标志 _enable_inactive_release 设置为 true
_enable_inactive_release = true;
//2.如果定时销毁任务存在，则刷新一下定时时间即可
if(_loop->HasTimer(_conn_id)){ return _loop->TimerRefresh(_conn_id);}
//3.如果不存在定时销毁任务，则新增
_loop->TimerAdd(_conn_id,sec,std::bind(&Connection::ReleaseInLoop,this));
}
void CancelInactiveReleaseInLoop()
{
_enable_inactive_release = false;
if(_loop->HasTimer(_conn_id))
{
_loop->TimerCancel(_conn_id);
}
}
//切换协议
void UpgradeInLoop(const Any &context
,const ConnectedCallback &conn
,const MessageCallback &msg
,const ClosedCallback &closed
,const AnyEventCallback &event)
{
_context = context; //设置上下文
_connected_callback = conn;
_message_callback = msg;
_closed_callback = closed;
_event_callback = event;
}
public:
Connection(EventLoop *loop,uint64_t conn_id,int sockfd)
:_conn_id(conn_id)
,_sockfd(sockfd)
,_enable_inactive_release(false)
,_loop(loop)
,_statu(CONNECTING)
,_socket(_sockfd)
,_channel(loop,sockfd)
{
_channel.SetCloseCallback(std::bind(&Connection::HandleClose,this));
_channel.SetEventCallback(std::bind(&Connection::HandleEvent,this));
_channel.SetReadCallback(std::bind(&Connection::HandleRead,this));
_channel.SetWriteCallback(std::bind(&Connection::HandleWrite,this));
_channel.SetErrorCallback(std::bind(&Connection::HandleError,this));
}
~Connection(){DBG_LOG("RELEASE CONNECTION:%p",this);}
//获取管理的文件描述符
int Fd() {return _sockfd; }
//获取连接ID
int Id() {return _conn_id;}
//返回状态
bool Connected(){return (_statu == CONNECTED);} //是否处于Connected状态
//设置上下文 – 连接建立完成时
void SetContext(const Any &context){_context = context;}
//获取行下文 – 返回的是一个指针
Any *GetContext(){ return &_context;}
void SetConnectedCallback(const ConnectedCallback &cb){ _connected_callback = cb;}
void SetMessageCallback(const MessageCallback &cb){_message_callback = cb;}
void SetClosedCallback(const ClosedCallback &cb){_closed_callback = cb;}
void SetAnyEventCallback(const AnyEventCallback &cb){_event_callback = cb;}
void SetSrvClosedCallback(const ClosedCallback &cb){_server_closed_callback = cb;}
//连接建立就绪后，进行Channel进行回调设置，启动读监控，调用_connected_callback
void Established()
{
_loop->RunInLoop(std::bind(&Connection::EstablishedInLoop,this));
}
//发送数据,将数据放到发送缓冲区，启动写事件监控
void Send(const char* data,size_t len)
{
//外界传入的 data 可能是一个临时的空间，我们现在只是把发送操作压如任务池，有可能没有立即被执行
//因此，有可能执行的时候，data指向的空间，有可能已经被释放了
Buffer buf;
buf.WriteAndPush(data,len);
_loop->RunInLoop(std::bind(&Connection::SendInLoop,this,std::move(buf)));
}
//提供给组件使用者的关闭接口 – 并不实际关闭，需要判断有没有数据待处理
void Shutdown()
{
_loop->RunInLoop(std::bind(&Connection::ShutdownInLoop,this));
}
void Release()
{
_loop->QueueInLoop(std::bind(&Connection::ReleaseInLoop, this));
}
//启动非活跃连接销毁 sec表示启动多长时间无通信算非活跃连接，添加定时任务
void EnableInactiveRelease(int sec)
{
_loop->RunInLoop(std::bind(&Connection::EnableInactiveReleaseInLoop,this,sec));
}
//取消非活跃连接销毁
void CancelInactiveRelease()
{
_loop->RunInLoop(std::bind(&Connection::CancelInactiveReleaseInLoop,this));
}
//协议切换 – 重置上下文，以及阶段性处理函数 – 这个进口必须在EventLoop线程中立即执行，防备新的事件触发后，处理的时候，切换任务还没有被执行，导致数据用原协议处理
void Upgrade(const Any &context,const ConnectedCallback &conn,const MessageCallback &msg,
const ClosedCallback &closed,const AnyEventCallback &event)
{
_loop->AssertInLoop();
_loop->RunInLoop(std::bind(&Connection::UpgradeInLoop,this,context,conn,msg,closed,event));
}
};

Acceptor模块：对监听套接字进行管理

（1）创建一个监听套接字 （2）启动读事件监控 （3）事件触发后，获取新连接 （4）调用新连接获取成功后的函数 而这个函数的功能是：

为新连接创建Connection进行管理（这一步不是accept模块操作的，而是服务器模块 ）

因为Accept模块只进行监听连接的管理，因此获取到新连接的描述符之后，对于新连接的描述符如何处理它并不关心。 对于新连接如何处理，应该是服务器模块来管理的。 服务器模块，实现了对新连接描述符的处理函数，将这个函数设置给Accept模块中的回调函数

class Acceptor
{
private:
Socket _socket; //监听套接字
EventLoop *_loop; //用于对监听套接字进行事件监控
Channel _channel; //用于对监听套接字进行事件管理

using AcceptCallback = std::function<void(int)>;
AcceptCallback _accept_callback;
private:
//监听套接字的读事件处理回调函数 – 获取新连接，调用AcceptCallback函数，进行新连接处理
void HandleRead()
{
int newfd = _socket.Accept();
if(newfd < 0)
{
return;
}
if(_accept_callback) _accept_callback(newfd);
}
int CreateServer(int port)
{
bool ret = _socket.CreatServer(port);
assert(ret == true);
return _socket.Fd();
}
public:
/*不能将启动读事件监控，放到构造函数中，必须在设置回调函数后，再去启动*/
/*否则有可能造成启动监控后，立即有事件，处理的时候，回调函数还没有设置，新连接得不到处理，且资源泄漏*/
Acceptor(EventLoop *loop,int port)
:_socket(CreateServer(port))
,_loop(loop)
,_channel(loop,_socket.Fd())
{
_channel.SetReadCallback(std::bind(&Acceptor::HandleRead,this));
}

//对回调函数的设置
void SetAcceptCallback(const AcceptCallback &cb){ _accept_callback = cb;}
void Listen(){_channel.EnableRead();}
};

//对回调函数的设置
void SetAcceptCallback(const AcceptCallback &cb){ _accept_callback = cb;}
void Listen(){_channel.EnableRead();}
};

LoopThread模块

目标： 将EventLoop模块与线程整和起来 EventLoop模块与线程是一一对应的。 EventLoop模块实例化的对象，在构造的时候就会初始化_thread_id,而后边运行的时候判断当前是否运行在EventLoop模块对应的线程中，就是将线程Id与EventLoop模块中的thread_id进行比较，相同就表示在同一个线程里，不同就表示当前运行的线程并不是EventLoop线程。

含义： EventLoop模块在实例化对象的时候，必须在线程内部 EventLoop在实例化对象的时候会设置自己的thread_id 如果我们先创建多个EventLoop对象，然后创建了多个线程，将各个线程的id，重新给EventLoop进行设置。 存在问题：在构造EventLoop对象，到设置新的thread_id期间将是不可控的。 因此我们必须先创建线程，然后在线程的入口函数中，去实例化EventLoop对象

构造一个新的模块：LoopThread 这个模块的功能：将EventLoop与thread整合到一起 思想： 1.创建线程 2.在线程中实例化EventLoop对象 功能： 可以向外部返回实例化的EventLoop

LoopThread模块的代码实现：

class LoopThread
{
private:
/*用于实现_loop的同步关系，避免线程创建了，在 _loop还没有实例化之前去获取_loop*/
std::mutex _mutex; //互斥锁
std::condition_variable _cond;//条件变量
EventLoop *_loop; //EventLoop指针变量，这个对象需要在线程内实例化
std::thread _thread; //EventLoop对应的线程
private:
/*实例化EvenLoop对象，唤醒_cond上有可能阻塞的线程，并且运行EventLoop模块对应的功能*/
void ThreadEntey()
{
EventLoop loop;
{
std::unique_lock<std::mutex> lock(_mutex); //加锁
_loop = &loop;
_cond.notify_all();
}
_loop->Start();
}
public:
/*创建线程，设定线程入口函数*/
LoopThread()
:_loop(nullptr)
,_thread(std::thread(&LoopThread::ThreadEntey,this)){}
/*返回当前线程关联的EventLoop对象指针*/
EventLoop *GetLoop()
{
EventLoop *loop = nullptr;
{
std::unique_lock<std::mutex> lock(_mutex); //加锁
_cond.wait(lock,[&](){return _loop != nullptr;}); //loop为空，就一直阻塞，等待被唤醒
loop = _loop;
}
return loop;
}
};

LoopThreadPoll模块

针对LoopThread设计一个线程池，便与使用者对线程的控制。 LoopThreadPoll模块：对所有的LoopThread进行管理和分配

功能：

注意事项：在服务器中，主从Reactor模型，主线程只负责新连接的获取，从属Reactor线程进行新连接的事件监控以及事件的处理，因此当前的线程池，从属线程的数量有可能为0，也就是实现单Reactor服务器。一个线程及负责获取连接，也负责连接的处理。

当主线程获取了一个新连接，需要将新连接挂到从属线程上进行事件监控及处理。 假设有0个从属线程，则直接分配给主线程的EventLoop进行处理 假设有多个从属线程，则采用RR轮转（轮转分配）思想，进行线程的分配(将对应线程的EventLoop获取到，设置给对应的Connection)\\

LoopThreadPoll模块代码实现：

class LoopThreadPool
{
private:
int _thread_count; //从属线程的数量
int _next_idx; //RR轮转
EventLoop* _baseloop; //主EventLoop.从属线程数量为0，则所有的操作都在_baseloop中进行
std::vector<LoopThread*> _threads; //保存所有的LoopThread对象
std::vector<EventLoop*> _loops; //从属线程的数量如果大于0，在从_loop中进行EventLoop线程分配
public:
LoopThreadPool(EventLoop *baseloop)
:_thread_count(0)
,_next_idx(0)
,_baseloop(baseloop)
{}
//设置线程数量
void SetThreadCount(int count){_thread_count = count;}
//创建所有的从属线程
void Create()
{
if(_thread_count > 0)
{
_threads.resize(_thread_count);
_loops.resize(_thread_count);
for(int i = 0;i < _thread_count; i++)
{
_threads[i] = new LoopThread();
_loops[i] = _threads[i]->GetLoop();
}
}
}
EventLoop *NextLoop()
{
if(_thread_count == 0)
{
return _baseloop;
}
_next_idx = (_next_idx + 1) % _thread_count;
return _loops[_next_idx];
}
};

TcpServer模块

TcpServer模块：对所有模块的整合，通过TcpServer实例化的对象，可以非常简单的完成一个服务器的搭建。

管理：

功能：

流程：

TcpServer模块的代码实现：

class TcpServer
{
private:
uint64_t _next_id; //这是一个自动增长的连接id
int _timeout; //这是非活跃连接的统计时间-多长时间没有通信就属于非活跃连接
int _port;
bool _enable_inactive_release;//是否启动了非活跃连接超时销毁的判断标志
EventLoop _baseloop; //这是主线程的EventLoop对象，负责监听事件的处理
Acceptor _acceptor; //这是监听套接字的管理对象
LoopThreadPool _pool; //这是从属的EventLoop线程池
std::unordered_map<uint64_t,PtrConnection> _conns; //保存所有连接对应的shared_ptr对象

using ConnectedCallback = std::function<void(const PtrConnection&)>;
//业务处理的回调
using MessageCallback = std::function<void(const PtrConnection&,Buffer *)>; //对缓冲区的数据进行业务处理
//关闭阶段的处理回调
using ClosedCallback = std::function<void(const PtrConnection&)>;
//任意事件触发的回调
using AnyEventCallback = std::function<void(const PtrConnection&)>;
using Functor = std::function<void()>;
ConnectedCallback _connected_callback;
MessageCallback _message_callback;
ClosedCallback _closed_callback;
AnyEventCallback _event_callback;
private:
void RunAfterInLoop(const Functor &task,int delay)
{
_next_id++;
_baseloop.TimerAdd(_next_id,delay,task);
}
//为新连接构造Connection进行管理
void NewConnectionn(int fd)
{
_next_id++;
PtrConnection conn(new Connection(_pool.NextLoop(),_next_id,fd));
conn->SetMessageCallback(_message_callback); //-1就是“占位符”，表示这个参数不是现在给，而是等以后调用时再传进来
conn->SetClosedCallback(_closed_callback);
conn->SetConnectedCallback(_connected_callback);
conn->SetAnyEventCallback(_event_callback);
conn->SetSrvClosedCallback(std::bind(&TcpServer::RemoveConnection,this,std::placeholders::_1));
if(_enable_inactive_release) conn->EnableInactiveRelease(_timeout); //启动非活跃超时销毁
conn->Established();//就绪初始化
_conns.insert(std::make_pair(_next_id,conn));
}
void RemoveConnectionInLoop(const PtrConnection &conn)
{
int id = conn->Id();
auto it = _conns.find(id);
if(it != _conns.end())
{
_conns.erase(id);
}
}
//从管理Connection的_conns中移除Connection的管理信息
void RemoveConnection(const PtrConnection &conn)
{
_baseloop.RunInLoop(std::bind(&TcpServer::RemoveConnectionInLoop,this,conn));
}
public:
TcpServer(int port)
:_port(port)
,_next_id(0)
,_enable_inactive_release(false)
,_acceptor(&_baseloop,port)
,_pool(&_baseloop)
{
_acceptor.SetAcceptCallback(std::bind(&TcpServer::NewConnectionn,this,std::placeholders::_1));
_acceptor.Listen();//将监听套接字挂到baseLoop上
}
void SetThreadCount(int count){ return _pool.SetThreadCount(count);}

void SetConnectedCallback(const ConnectedCallback &cb){ _connected_callback = cb;}
void SetMessageCallback(const MessageCallback &cb){_message_callback = cb;}
void SetClosedCallback(const ClosedCallback &cb){_closed_callback = cb;}
void SetAnyEventCallback(const AnyEventCallback &cb){_event_callback = cb;}
void EnableInactiveRelease(int timeout){_timeout = timeout;_enable_inactive_release = true;}
//用于添加一个定时任务
void RunAfter(const Functor &task,int delay)
{
_baseloop.RunInLoop(std::bind(&TcpServer::RunAfterInLoop,this,task,delay));
}
void Start()
{
_pool.Create();//创线程中的从属线程
_baseloop.Start();
}
};

void Channel::Remove(){ return _loop->RemoveEvent(this);}
void Channel::Update(){ return _loop->UpdateEvent(this);}
/*不想加锁，那就把对定时器的所有操作都放在一个线程中进行*/
void TimerWheel::TimerAdd(uint64_t id,uint32_t delay,const TaskFunc &cb)
{
_loop->RunInLoop(std::bind(&TimerWheel::TimerAddInLoop,this,id,delay,cb));
}
//修改定时任务(刷新，延时定时任务)
void TimerWheel::TimerRefresh(uint64_t id)
{
_loop->RunInLoop(std::bind(&TimerWheel::TimerRefreshInLoop,this,id));
}
//取消定时任务
void TimerWheel::TimerCancel(uint32_t id)
{
_loop->RunInLoop(std::bind(&TimerWheel::TimerCancelInLoop,this,id));
}

class NetWork
{
public:
NetWork()
{
DBG_LOG("SIGPIPE INIT");
signal(SIGPIPE,SIG_IGN); //IGPIPE 防止连接关闭的时候，继续发送数据导致send触发异常，导致程序退出
}
};