仿modou庫one thread one loop式并發服務器

源碼：田某super/moduo

SERVER模塊：

Buffer模塊：

Socket模塊：

Channel模塊：

Connection模塊：

Acceptor模塊：

TimerQueue模塊：

Poller模塊：

EventLoop模塊：

TcpServer模塊：

HTTP協議組件模塊：

Util模塊：

HttpRequest模塊：

HttpContext模塊：

HttpServer模塊：

通過咱們實現的?并發服務器組件，可以簡潔快速的完成?個?性能的服務器搭建。
并且，通過組件內提供的不同應?層協議?持，也可以快速完成?個?性能應?服務器的搭建（當前，為了便于項?的演?，項?中提供HTTP協議組件的?持）。在這?，要明確的是咱們要實現的是?個?并發服務器組件，因此當前的項?中并不包含實際的業務內容。

圖片來源：?圖解one loop per thread：使用muduo網絡庫實現web服務器_znzxc的博客-CSDN博客

本項目總共分為兩大模塊分別為Server和HTTP模塊。

SERVER模塊：

SERVER模塊就是對所有的連接以及線程進?管理，讓它們各司其職，在合適的時候做合適的事，最終完成?性能服務器組件的實現。

管理??：
監聽連接管理：對監聽連接進?管理
通信連接管理：對通信連接進?管理
超時連接管理：對超時連接進?管理
基于以上可以將其分為多個子模塊：

Buffer模塊：

?于實現用戶態緩沖區，提供數據緩沖，取出等功能。

class Buffer {private:std::vector<char> _buffer; //使用vector進行內存空間管理uint64_t _reader_idx; //讀偏移uint64_t _writer_idx; //寫偏移public:Buffer():_reader_idx(0), _writer_idx(0), _buffer(BUFFER_DEFAULT_SIZE){}char *Begin() { return &*_buffer.begin(); }//獲取當前寫入起始地址, _buffer的空間起始地址，加上寫偏移量char *WritePosition() { return Begin() + _writer_idx; }//獲取當前讀取起始地址char *ReadPosition() { return Begin() + _reader_idx; }//獲取緩沖區末尾空閑空間大小--寫偏移之后的空閑空間, 總體空間大小減去寫偏移uint64_t TailIdleSize() { return _buffer.size() - _writer_idx; }//獲取緩沖區起始空閑空間大小--讀偏移之前的空閑空間uint64_t HeadIdleSize() { return _reader_idx; }//獲取可讀數據大小 = 寫偏移 - 讀偏移uint64_t ReadAbleSize() { return _writer_idx - _reader_idx; }//將讀偏移向后移動void MoveReadOffset(uint64_t len) { if (len == 0) return; //向后移動的大小，必須小于可讀數據大小assert(len <= ReadAbleSize());_reader_idx += len;}//將寫偏移向后移動 void MoveWriteOffset(uint64_t len) {//向后移動的大小，必須小于當前后邊的空閑空間大小assert(len <= TailIdleSize());_writer_idx += len;}//確保可寫空間足夠（整體空閑空間夠了就移動數據，否則就擴容）void EnsureWriteSpace(uint64_t len) {//如果末尾空閑空間大小足夠，直接返回if (TailIdleSize() >= len) { return; }//末尾空閑空間不夠，則判斷加上起始位置的空閑空間大小是否足夠, 夠了就將數據移動到起始位置if (len <= TailIdleSize() + HeadIdleSize()) {//將數據移動到起始位置uint64_t rsz = ReadAbleSize();//把當前數據大小先保存起來std::copy(ReadPosition(), ReadPosition() + rsz, Begin());//把可讀數據拷貝到起始位置_reader_idx = 0;    //將讀偏移歸0_writer_idx = rsz;  //將寫位置置為可讀數據大小， 因為當前的可讀數據大小就是寫偏移量}else {//總體空間不夠，則需要擴容，不移動數據，直接給寫偏移之后擴容足夠空間即可DBG_LOG("RESIZE %ld", _writer_idx + len);_buffer.resize(_writer_idx + len);}} //寫入數據void Write(const void *data, uint64_t len) {//1. 保證有足夠空間，2. 拷貝數據進去if (len == 0) return;EnsureWriteSpace(len);const char *d = (const char *)data;std::copy(d, d + len, WritePosition());}void WriteAndPush(const void *data, uint64_t len) {Write(data, len);MoveWriteOffset(len);}void WriteString(const std::string &data) {return Write(data.c_str(), data.size());}void WriteStringAndPush(const std::string &data) {WriteString(data);MoveWriteOffset(data.size());}void WriteBuffer(Buffer &data) {return Write(data.ReadPosition(), data.ReadAbleSize());}void WriteBufferAndPush(Buffer &data) { WriteBuffer(data);MoveWriteOffset(data.ReadAbleSize());}//讀取數據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);}std::string ReadAsString(uint64_t len) {//要求要獲取的數據大小必須小于可讀數據大小assert(len <= ReadAbleSize());std::string str;str.resize(len);Read(&str[0], len);return str;}std::string ReadAsStringAndPop(uint64_t len) {assert(len <= ReadAbleSize());std::string str = ReadAsString(len);MoveReadOffset(len);return str;}char *FindCRLF() {char *res = (char*)memchr(ReadPosition(), '\n', ReadAbleSize());return res;}/*通常獲取一行數據，這種情況針對是*/std::string GetLine() {char *pos = FindCRLF();if (pos == NULL) {return "";}// +1是為了把換行字符也取出來。return ReadAsString(pos - ReadPosition() + 1);}std::string GetLineAndPop() {std::string str = GetLine();MoveReadOffset(str.size());return str;}//清空緩沖區void Clear() {//只需要將偏移量歸0即可_reader_idx = 0;_writer_idx = 0;}
};

Socket模塊：

封裝套接字

class Socket {private:int _sockfd;public:Socket():_sockfd(-1) {}Socket(int fd): _sockfd(fd) {}~Socket() { Close(); }int Fd() { return _sockfd; }//創建套接字bool Create() {// int socket(int domain, int type, int protocol)_sockfd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);if (_sockfd < 0) {ERR_LOG("CREATE SOCKET FAILED!!");return false;}return true;}//綁定地址信息bool Bind(const std::string &ip, uint16_t port) {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;}//開始監聽bool Listen(int backlog = MAX_LISTEN) {// int listen(int backlog)int ret = listen(_sockfd, backlog);if (ret < 0) {ERR_LOG("SOCKET LISTEN FAILED!");return false;}return true;}//向服務器發起連接bool Connect(const std::string &ip, uint16_t port) {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 connect(int sockfd, struct sockaddr*addr, socklen_t len);int ret = connect(_sockfd, (struct sockaddr*)&addr, len);if (ret < 0) {ERR_LOG("CONNECT SERVER FAILED!");return false;}return true;}//獲取新連接int Accept() {// int accept(int sockfd, struct sockaddr *addr, socklen_t *len);int newfd = accept(_sockfd, NULL, NULL);if (newfd < 0) {ERR_LOG("SOCKET ACCEPT FAILED!");return -1;}return newfd;}//接收數據ssize_t Recv(void *buf, size_t len, int flag = 0) {// ssize_t recv(int sockfd, void *buf, size_t len, int flag);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 表示當前接收為非阻塞。}//發送數據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) {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 表示當前發送為非阻塞。}//關閉套接字void Close() {if (_sockfd != -1) {close(_sockfd);_sockfd = -1;}}//創建一個服務端連接bool CreateServer(uint16_t port, const std::string &ip = "0.0.0.0", bool block_flag = false) {//1. 創建套接字，2. 綁定地址，3. 開始監聽，4. 設置非阻塞， 5. 啟動地址重用if (Create() == false) return false;if (block_flag) NonBlock();if (Bind(ip, port) == false) return false;if (Listen() == false) return false;ReuseAddress();return true;}//創建一個客戶端連接bool CreateClient(uint16_t port, const std::string &ip) {//1. 創建套接字，2.指向連接服務器if (Create() == false) return false;if (Connect(ip, port) == false) return false;return true;}//設置套接字選項---開啟地址端口重用void ReuseAddress() {// int setsockopt(int fd, int leve, int optname, void *val, 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));}//設置套接字阻塞屬性-- 設置為非阻塞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模塊是對?個描述符需要進?的IO事件管理的模塊，實現對描述符可讀，可寫，錯誤...事件的管理操作，以及Poller模塊對描述符進?IO事件監控就緒后，根據不同的事件，回調不同的處理函數功能。

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();}
};

Connection模塊：

Connection模塊是對Buffer模塊，Socket模塊，Channel模塊的?個整體封裝，實現了對?個通信套接字的整體的管理，每?個進?數據通信的套接字（也就是accept獲取到的新連接）都會使? Connection進?管理。

class Connection : public std::enable_shared_from_this<Connection> {private:uint64_t _conn_id;  // 連接的唯一ID，便于連接的管理和查找//uint64_t _timer_id;   //定時器ID，必須是唯一的，這塊為了簡化操作使用conn_id作為定時器IDint _sockfd;        // 連接關聯的文件描述符bool _enable_inactive_release;  // 連接是否啟動非活躍銷毀的判斷標志，默認為falseEventLoop *_loop;   // 連接所關聯的一個EventLoopConnStatu _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_callbackvoid HandleRead() {//1. 接收socket的數據，放到緩沖區char buf[65536];ssize_t ret = _socket.NonBlockRecv(buf, 65535);if (ret < 0) {//出錯了,不能直接關閉連接return ShutdownInLoop();}//這里的等于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()); }}//連接獲取之后，所處的狀態下要進行各種設置（啟動讀監控,調用回調函數）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::Release, 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; }//獲取連接IDint Id() { return _conn_id; }//是否處于CONNECTED狀態bool Connected() { return (_statu == 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_callbackvoid 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));}//啟動非活躍銷毀，并定義多長時間無通信就是非活躍，添加定時任務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模塊：

Acceptor模塊是對Socket模塊，Channel模塊的?個整體封裝，實現了對?個監聽套接字的整體的管理。

class Acceptor {private:Socket _socket;//用于創建監聽套接字EventLoop *_loop; //用于對監聽套接字進行事件監控Channel _channel; //用于對監聽套接字進行事件管理using AcceptCallback = std::function<void(int)>;AcceptCallback _accept_callback;private:/*監聽套接字的讀事件回調處理函數---獲取新連接，調用_accept_callback函數進行新連接處理*/void HandleRead() {int newfd = _socket.Accept();if (newfd < 0) {return ;}if (_accept_callback) _accept_callback(newfd);}int CreateServer(int port) {bool ret = _socket.CreateServer(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(); }
};

TimerQueue模塊：

TimerQueue模塊是實現固定時間定時任務的模塊，可以理解就是要給定時任務管理器，向定時任務管理器中添加?個任務，任務將在固定時間后被執?，同時也可以通過刷新定時任務來延遲任務的執?。

using TaskFunc = std::function<void()>;
using ReleaseFunc = std::function<void()>;
class TimerTask{private:uint64_t _id;       // 定時器任務對象IDuint32_t _timeout;  //定時任務的超時時間bool _canceled;     // false-表示沒有被取消， true-表示被取消TaskFunc _task_cb;  //定時器對象要執行的定時任務ReleaseFunc _release; //用于刪除TimerWheel中保存的定時器對象信息public:TimerTask(uint64_t id, uint32_t delay, const TaskFunc &cb): _id(id), _timeout(delay), _task_cb(cb), _canceled(false) {}~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;std::unordered_map<uint64_t, WeakTask> _timers;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 timerfd_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;//第一次超時時間為1s后itime.it_interval.tv_sec = 1; itime.it_interval.tv_nsec = 0; //第一次超時后，每次超時的間隔時timerfd_settime(timerfd, 0, &itime, NULL);return timerfd;}int ReadTimefd() {uint64_t times;//有可能因為其他描述符的事件處理花費事件比較長，然后在處理定時器描述符事件的時候，有可能就已經超時了很多次//read讀取到的數據times就是從上一次read之后超時的次數int ret = read(_timerfd, &times, 8);if (ret < 0) {ERR_LOG("READ TIMEFD FAILED!");abort();}return times;}//這個函數應該每秒鐘被執行一次，相當于秒針向后走了一步void RunTimerTask() {_tick = (_tick + 1) % _capacity;_wheel[_tick].clear();//清空指定位置的數組，就會把數組中保存的所有管理定時器對象的shared_ptr釋放掉}void OnTime() {//根據實際超時的次數，執行對應的超時任務int times = ReadTimefd();for (int i = 0; i < times; i++) {RunTimerTask();}}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構造一個shared_ptr出來，添加到輪子中auto it = _timers.find(id);if (it == _timers.end()) {return;//沒找著定時任務，沒法刷新，沒法延遲}PtrTask pt = it->second.lock();//lock獲取weak_ptr管理的對象對應的shared_ptrint delay = pt->DelayTime();int pos = (_tick + delay) % _capacity;_wheel[pos].push_back(pt);}void TimerCancelInLoop(uint64_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(uint64_t id);/*這個接口存在線程安全問題--這個接口實際上不能被外界使用者調用，只能在模塊內，在對應的EventLoop線程內執行*/bool HasTimer(uint64_t id) {auto it = _timers.find(id);if (it == _timers.end()) {return false;}return true;}
};