仿mudou库one thread oneloop式并发服务器

#include <iostream>
#include <typeinfo>
#include <string>
class Any
{
private:class holder{public:holder(){}virtual const std::type_info &type() = 0; // 返回保存数据的类型virtual holder *clone() = 0;              // 克隆自身};template <class T>class placeHolder : public holder{public:placeHolder(const T &val): _val(val){}virtual const std::type_info &type(){return typeid(_val);}virtual placeHolder *clone(){return new placeHolder(_val);}T _val;};holder *_content;public://无参构造Any() : _content(nullptr) {}//有参构造template <class T>Any(const T val): _content(new placeHolder<T>(val)){}//克隆函数Any(const Any &other): _content(other._content == nullptr ? nullptr : other._content->clone()) {} // 这个地方用到了多态，父类指针调用子类的clone函数//析构~Any(){delete _content;}//交换自身和传入对象保存的内容Any &Swap(Any &other){std::swap(_content, other._content);return *this;}template <class T>T *get(){if (typeid(T) != _content->type())return nullptr;elsereturn &(((placeHolder<T>*)_content)->_val);}Any &operator=(const Any &other){// 为val构造一个临时的通用容器，然后与当前容器自身进行指针交换，临时对象释放的时候，原先保存的数据也就被释放Any(other).Swap(*this);return *this;}template <class T>Any &operator=(const T val){Any(val).Swap(*this);return *this;}
};//测试用例
/*
class Test{public:Test() {std::cout << "构造" << std::endl;}Test(const Test &t) {std::cout << "拷贝" << std::endl;}~Test() {std::cout << "析构" << std::endl;}
};
int main()
{Any a;a = 10;int *pa = a.get<int>();std::cout << *pa << std::endl;a = std::string("nihao");std::string *ps = a.get<std::string>();std::cout << *ps << std::endl;return 0;
}*/

HTTP协议模块⽤于对⾼并发服务器模块进⾏协议⽀持，基于提供的协议⽀持能够更⽅便的完成指定协 议服务器的搭建。

⽽HTTP协议⽀持模块的实现，可以细分为以下⼏个模块。

这个模块是⼀个⼯具模块，主要提供HTTP协议模块所⽤到的⼀些⼯具函数，⽐如url编解码，⽂件读 写....等。 HttpRequest模块：

这个模块是HTTP请求数据模块，⽤于保存HTTP请求数据被解析后的各项请求元素信息。

这个模块是HTTP响应数据模块，⽤于业务处理后设置并保存HTTP响应数据的的各项元素信息，最终 会被按照HTTP协议响应格式组织成为响应信息发送给客⼾端。

HttpContext模块：

HttpServer模块：

Buffer类⽤于实现⽤⼾态缓冲区，提供数据缓冲，取出等功能。

class Buffer
{
#define BUFFER_DEFAULT_SIZE 1024
private:std::vector<char> _buffer;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();}// 获取写入位置char *WritePosition(){return Begin() + _writer_idx;}// 获取读取位置char *ReadPosition(){return Begin() + _reader_idx;}// 获取头剩余空间uint64_t HeadIdleSize(){return _reader_idx;}// 获取尾部剩余空间uint64_t TailIdleSize(){return _buffer.size() - _writer_idx;}// 获取可读数据大小uint64_t ReadAbleSize(){return _writer_idx - _reader_idx;}// 将读偏移向后移动void MoveReadOffset(uint64_t len){if (len == 0)return;// 向后移动的大小，必须小于可读数据大小if(len <= ReadAbleSize())_reader_idx += len;else ERR_LOG("MoveReadOffset ERROR");}// 将写偏移向后移动void MoveWriteOffset(uint64_t len){// 向后移动的大小，必须小于当前后边的空闲空间大小if(len <= TailIdleSize())_writer_idx += len;else ERR_LOG("MoveWriteOffset ERROR");}// 确保可以空间足够，否则就扩容void EnsureWriteSpace(uint64_t len){if (TailIdleSize() >= len){return;}else{if (len <= HeadIdleSize() + TailIdleSize()){// 如果头尾空间足够，就尽量节约空间不扩容，只将缓冲区重新整理即可// 将数据移动到头部，将头部和尾部剩余空间结合uint64_t rsz = ReadAbleSize();std::copy(ReadPosition(), ReadPosition() + rsz, Begin());_reader_idx = 0;_writer_idx = rsz;}else{// 总体空间不够就进行扩容_buffer.resize(_writer_idx + len);}}}// 写入数据void Write(const void *data, int len){// 1.保证有足够空间 2.将数据拷入缓冲区EnsureWriteSpace(len);const char *d = (const char *)data;std::copy(d, d + len, WritePosition());}void WriteAndPush(const void *data, int len){Write(data, len);MoveWriteOffset(len);}// 写入字符串void WriteString(std::string &data){Write(data.c_str(), data.size());}void WriteStringAndPush(std::string &data){WriteString(data);MoveWriteOffset(data.size());}// 从其他缓冲区写入void WriteBuffer(Buffer &buf){Write((const void *)buf.ReadPosition(), buf.ReadAbleSize());}void WriteBufferAndPush(Buffer &buf){Write((const void *)buf.ReadPosition(), buf.ReadAbleSize());MoveWriteOffset(buf.ReadAbleSize());}// 读取数据void Read(void *rec, int len){if (len <= ReadAbleSize()){// std::copy(Begin() + _reader_size, Begin() + _writer_size, (char *)rec);std::copy(ReadPosition(), ReadPosition() + len, (char*)rec);}else{ERR_LOG("读取数据出错");}}void ReadAndPop(void *rec, int len){if (len <= ReadAbleSize()){Read(rec, len);MoveReadOffset(len);}else{ERR_LOG("ReadAndPop ERROR");}}// 以字符串的形式读取数据std::string ReadAsString(int len){if (len <= ReadAbleSize()){std::string str;str.resize(len);Read(&str[0], len);return str;}return "";}std::string ReadAsStringAndpop(int len){if(len <= ReadAbleSize()){std::string ret = ReadAsString(len);MoveReadOffset(len);return ret;}else {ERR_LOG("ReadAsStringAndpop ERROR");return "";}}// 查找一行数据char *FindCRLF(){void *res = memchr((void *)ReadPosition(), '\n', ReadAbleSize());return (char *)res;}// 以字符串形式获取一行数据std::string GetLine(){char *pos = FindCRLF();if (pos == nullptr){return "";}//+1是把换行符也取出来，因为pos是下一个行的起始地址return ReadAsString(pos - ReadPosition() + 1);}std::string GetLineAndPop(){// 调用GetLine函数获取一行字符串std::string ret = GetLine();// 根据获取的字符串长度移动读取偏移量MoveReadOffset(ret.size());// 返回获取到的字符串return ret;}// 清空数据void Clear(){_writer_idx = 0;_reader_idx = 0;}
};

⽇志宏的实现：

#define INF 0
#define DBG 1
#define ERR 2
#define LOG_LEVEL DBG#define LOG(level, format, ...) do{\if (level < LOG_LEVEL) break;\time_t t = time(NULL);\struct tm *ltm = localtime(&t);\char tmp[32] = {0};\strftime(tmp, 31, "%H:%M:%S", ltm);\fprintf(stdout, "[%p %s %s:%d] " format "\n", (void*)pthread_self(), 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__)

套接字Socket类实现：

#define MAX_LISEN 1024
class Socket
{
private:int _socketfd;public:Socket() : _socketfd(-1) {}Socket(int fd) : _socketfd(fd) {}~Socket() {}int Fd(){return _socketfd;}// 创建套接字bool Create(){_socketfd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);if (_socketfd < 0){ERR_LOG("CREATED SOCKET ERROR");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 ret = bind(_socketfd, (struct sockaddr *)&addr, len);if (ret < 0){ERR_LOG("BIND ERROR");return false;}return true;}// 监听套接字bool Listen(int backLog = MAX_LISEN){int ret = listen(_socketfd, backLog);if (ret < 0){ERR_LOG("LISTEN ERROR");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 ret = connect(_socketfd, (struct sockaddr *)&addr, len);if (ret < 0){ERR_LOG("CONNECT FAILED");return false;}return true;}// 获取新连接int Accept(){int newfd = accept(_socketfd, nullptr, nullptr);if (newfd < 0){ERR_LOG("ACCEPT FAILED");return -1;}return newfd;}// 接收数据ssize_t Recv(void *buf, size_t len, int flag = 0){ssize_t ret = recv(_socketfd, buf, len, flag);if (ret <= 0){if (errno == EAGAIN || errno == EINTR){ERR_LOG("RECEIVE AGAIN");return 0;}ERR_LOG("RECEIVE FAILED:%d", errno);return -1;}return ret;}ssize_t NoBlockRecv(void *buf, size_t len){// 使用Recv函数接收数据，并设置标志位为MSG_DONTWAIT以实现非阻塞接收return Recv(buf, len, MSG_DONTWAIT);}// 发送数据int Send(const void *buf, size_t len, int flag = 0){DBG_LOG("SEND:%ld", len);ssize_t ret = send(_socketfd, buf, len, flag);if (ret <= 0){if (ret < 0){if (errno == EAGAIN || errno == EINTR){return 0;}}ERR_LOG("SEND FAILED");return -1;}return ret;}ssize_t NoBlockSend(void *buf, size_t len){return Send(buf, len, MSG_DONTWAIT);}// 关闭套接字void Close(){if (_socketfd != -1){close(_socketfd);_socketfd = -1;}}// 创建一个服务端连接bool CreateServer(uint16_t port, const std::string &ip = "0.0.0.0", bool block_flag = false){if (Create() == false)return false;if (block_flag)NoBlock();if (Bind(ip, port) == false)return false;if (Listen() == false)return false;ReuseAdress();return true;}// 创建一个客户端连接bool CreateClient(uint16_t port, const std::string &ip){if (Create() == false)return false;if (Connect(ip, port) == false)return false;return true;}// 开启地址端口重用void ReuseAdress(){int val = 1;// 设置套接字选项 SO_REUSEADDR，允许地址重用setsockopt(_socketfd, SOL_SOCKET, SO_REUSEADDR, (void*)&val, sizeof(int));val = 1;// 设置套接字选项 SO_REUSEPORT，允许端口重用setsockopt(_socketfd, SOL_SOCKET, SO_REUSEPORT, (void*)&val, sizeof(int));}// 套接字设置为非阻塞void NoBlock(){//获取原本的flag值int flag = fcntl(_socketfd, F_GETFL, 0);fcntl(_socketfd, F_SETFL, flag | O_NONBLOCK);//新增非阻塞}
};

事件管理Channel类实现：

/*
一个channel对应一个fd，去监管一个fd上发生的所有事件
poller负责监管所有channel，poller才具有实际的epoll，poller才是实际的监控者，负责监管所有channel
channel保存一个envent表和相应的回调函数,这也poller才可以根据channel进行相应的监控（根据event判断是否要监控读，写等）和正确的回调（读，写等触发怎么样的回调）
*/
class Poller;
class EventLoop;
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): _loop(Loop),_fd(fd),_events(0),_revents(0){}int Fd(){return _fd;}// 设置实际就绪的事件void SetRevents(uint32_t events){_revents = events;}// 获取想要监控的事件uint32_t Event(){return _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 DisableWrite(){_events &= ~EPOLLOUT;Update();}// 关闭读事件监控void DisableRead(){_events &= ~EPOLLIN;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 (_event_callback)_event_callback();}// 可写事件调用if (_revents & EPOLLOUT){if (_write_callback)_write_callback();// 任何事件都调用，刷新活跃度if (_event_callback)_event_callback();}//错误或者关闭事件由于可能会导致连接关闭，所以先刷新活跃度防止连接被释放以后再刷新导致错误//并且由于可能导致连接关闭，所以一次只处理一个// 错误事件调用if (_revents & EPOLLERR){// 任何事件都调用，刷新活跃度if (_event_callback)_event_callback();if (_error_callback)_error_callback();}// 关闭事件调用else if (_revents & EPOLLHUP){// 任何事件都调用，刷新活跃度if (_event_callback)_event_callback();DBG_LOG("_CLOSE_CALLBACK");if (_close_callback){_close_callback();}}}
};#define MAX_EPOLLEVENTS 1024
//实际的监控者，负责监控所有事件（并根据channel进行事件管理）
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){// 获取Channel对象的文件描述符int fd = channel->Fd();// 创建epoll事件结构体struct epoll_event ev;// 设置事件结构体中的文件描述符ev.data.fd = fd;// 设置事件结构体中的事件类型ev.events = channel->Event();// 调用epoll_ctl函数更新epoll事件int ret = epoll_ctl(_epfd, op, fd, &ev);// 如果epoll_ctl函数执行失败if (ret < 0){// 打印错误信息DBG_LOG("EPOLL_CTL FAILED");}}// 判断channel是否已经被poller管理bool HasChannel(Channel *channel){auto it = _channels.find(channel->Fd());if (it != _channels.end())return true;elsereturn false;}public:Poller(){_epfd = epoll_create(MAX_EPOLLEVENTS);if (_epfd < 0){DBG_LOG("EPOLL_CREATE ERROR");abort();}}// 添加，修改监控事件void UpdateEvent(Channel *channel){bool ret = HasChannel(channel);if (ret){// 存在就修改Update(channel, EPOLL_CTL_MOD);}else{_channels[channel->Fd()] = channel;Update(channel, EPOLL_CTL_ADD);}}// 移除监控void RemoveEvent(Channel *channel){auto it = _channels.find(channel->Fd());if (it != _channels.end()){_channels.erase(it);Update(channel, EPOLL_CTL_DEL);return;}DBG_LOG("Remove Event Error");}// 开始监控，获取活跃连接// epoll会返回就绪事件的fd，然后根据fd再查找对应的channel,再由EventLoop调用HandleEventvoid Poll(std::vector<Channel *> *active){int _nfds = epoll_wait(_epfd, _evs, MAX_EPOLLEVENTS, -1);if (_nfds < 0){if (errno == EINTR){return;}ERR_LOG("EPOLL_WAIT FAILED");abort();}for (int i = 0; i < _nfds; i++){auto it = _channels.find(_evs[i].data.fd);if (it != _channels.end()){it->second->SetRevents(_evs[i].events); // 设置实际就绪事件类型active->push_back(it->second);}else{DBG_LOG("_CHANNELS FIND ERROR");}}}
};

描述符事件监控Poller类实现：

#define MAX_EPOLLEVENTS 1024
//实际的监控者，负责监控所有事件（并根据channel进行事件管理）
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){// 获取Channel对象的文件描述符int fd = channel->Fd();// 创建epoll事件结构体struct epoll_event ev;// 设置事件结构体中的文件描述符ev.data.fd = fd;// 设置事件结构体中的事件类型ev.events = channel->Event();// 调用epoll_ctl函数更新epoll事件int ret = epoll_ctl(_epfd, op, fd, &ev);// 如果epoll_ctl函数执行失败if (ret < 0){// 打印错误信息DBG_LOG("EPOLL_CTL FAILED");}}// 判断channel是否已经被poller管理bool HasChannel(Channel *channel){auto it = _channels.find(channel->Fd());if (it != _channels.end())return true;elsereturn false;}public:Poller(){_epfd = epoll_create(MAX_EPOLLEVENTS);if (_epfd < 0){DBG_LOG("EPOLL_CREATE ERROR");abort();}}// 添加，修改监控事件void UpdateEvent(Channel *channel){bool ret = HasChannel(channel);if (ret){// 存在就修改Update(channel, EPOLL_CTL_MOD);}else{_channels[channel->Fd()] = channel;Update(channel, EPOLL_CTL_ADD);}}// 移除监控void RemoveEvent(Channel *channel){auto it = _channels.find(channel->Fd());if (it != _channels.end()){_channels.erase(it);Update(channel, EPOLL_CTL_DEL);return;}DBG_LOG("Remove Event Error");}// 开始监控，获取活跃连接// epoll会返回就绪事件的fd，然后根据fd再查找对应的channel,再由EventLoop调用HandleEventvoid Poll(std::vector<Channel *> *active){int _nfds = epoll_wait(_epfd, _evs, MAX_EPOLLEVENTS, -1);if (_nfds < 0){if (errno == EINTR){return;}ERR_LOG("EPOLL_WAIT FAILED");abort();}for (int i = 0; i < _nfds; i++){auto it = _channels.find(_evs[i].data.fd);if (it != _channels.end()){it->second->SetRevents(_evs[i].events); // 设置实际就绪事件类型active->push_back(it->second);}else{DBG_LOG("_CHANNELS FIND ERROR");}}}
};

Reactor-EventLoop实现：

class EventLoop
{using Functor = std::function<void()>;std::thread::id _thread_id; // 线程Idint _event_fd;              // 唤醒由于IO事件可能导致的阻塞std::unique_ptr<Channel> _event_channel;Poller _poller;              // 对描述符进行监控std::vector<Functor> _tasks; // 任务池std::mutex _mutex;           // 保障对任务池操作的安全性TimeWheel _timer_wheel;public:// 执行任务池中的所有操作void RunAllTask(){std::vector<Functor> tasks;{//执行任务的时候直接将任务池中的任务置换出来，再处理，避免执行任务的时候阻塞任务池std::unique_lock<std::mutex> _lock(_mutex);tasks.swap(_tasks);}for (auto f : tasks){f();}return;}static int CreateEventFd(){int efd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);if (efd < 0){ERR_LOG("EVENT FD CREATE FAILED");abort();}return efd;}void ReadEventFd(){uint64_t res = 0;res = read(_event_fd, &res, sizeof(res));if (res < 0){if (errno == EINTR){return;}ERR_LOG("ReadEventFd ERROR");}}/*** @brief 唤醒事件文件描述符** 通过向事件文件描述符写入数据来唤醒等待该描述符的线程。** @return 无返回值*/void WakeUpEventFd(){// 设置写入的值uint64_t val = 1;// 将值写入到事件文件描述符int ret = write(_event_fd, &val, sizeof(val));if (ret < 0){// 判断错误码是否为EINTR或EAGAIN// EINTR:被信号打断 EAGAIN:无数据可读if (errno == EINTR || errno == EAGAIN){// 如果是这两种错误码之一，则直接返回return;}// 如果不是这两种错误码，则记录错误日志ERR_LOG("ReadEventFd ERROR");}}public:EventLoop(): _thread_id(std::this_thread::get_id()),_event_fd(CreateEventFd()),_event_channel(new Channel(this, _event_fd)),_timer_wheel(this){// 给Event_fd添加回调函数读取eventfd，通过readEventfd来打断IO阻塞//由于epoll的事件监控，有可能会因为没有事件到来⽽持续阻塞，导致任务队列中的任务不能及时得//到执⾏，因此创建了eventfd，添加到Poller的事件监控中，⽤于实现每次向任务队列添加任务的时//候，通过向eventfd写⼊数据来唤醒epoll的阻塞。_event_channel->SetReadCallBack(std::bind(&EventLoop::ReadEventFd, this));_event_channel->EnableRead();}// 判断要执行的任务是否在当前线程中，如果是就直接执行，否则就压入队列void RunInLoop(const Functor &cb){if (IsInLoop()){cb();}else{QueueInLoop(cb);}}// 开始运行三步走--事件监控=》就绪事件处理=》执行任务void Start(){while (1){std::vector<Channel *> actives;_poller.Poll(&actives);//先处理事件触发的回调（读事件回调，写事件回调.....）for (auto &channel : actives){channel->HandleEvent();}//然后再执行任务池中的任务RunAllTask();}}// 将任务压入队列void QueueInLoop(const Functor &cb){{std::unique_lock<std::mutex> _lock(_mutex);_tasks.push_back(cb);}// 唤醒因为没有事件就绪从而导致的epoll阻塞// 其实就是给eventfd写入一个数据，从而触发可读事件打断阻塞// 因为任务池的处理是在有IO事件被触发以后才会被处理的，所有如果完全没有IO事件导致阻塞就会导致任务池的任务也一直不被执行WakeUpEventFd();}// 判断当前线程是否是EventLoop对应的线程bool IsInLoop(){return (_thread_id == std::this_thread::get_id());}// 更新事件监控void UpdateEvent(Channel *channel){_poller.UpdateEvent(channel);}// 移除事件监控void RemoveEvent(Channel *channel){_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);}
};

定时器模块的整合

using TaskFunc = std::function<void()>;
using ReleaseFunc = std::function<void()>;
class TimerTask
{uint64_t _id;         // 定时器的IDuint64_t _timeout;    // 超时时间bool _canceled;       // 取消定时任务TaskFunc _task_cb;    // 回调函数的任务ReleaseFunc _release; // 清除时间轮中的记录public:TimerTask(uint64_t id, uint64_t timeout, TaskFunc cb): _id(id),_timeout(timeout),_task_cb(cb),_canceled(false){}uint64_t ID(){return _id;}void SetRelease(ReleaseFunc release){_release = release;}~TimerTask(){if (!_canceled)_task_cb();_release();}void Cancel(){DBG_LOG("TIMER CANCEL");_canceled = true;}//获取超时时间uint32_t DelayTime(){return _timeout;}
};class TimeWheel
{
private:using WeakTask = std::weak_ptr<TimerTask>;using PtrTask = std::shared_ptr<TimerTask>;int _tick;                                      // 秒针int _capacity;                                  // 最大定时时间std::vector<std::vector<PtrTask>> _timeWheel;   // 时间轮std::unordered_map<uint32_t, WeakTask> _timers; // 用来保存定时器的信息EventLoop *_loop;int _timerfd; // 定时器描述符--可读事件回调就是读取计时器执行定时任务std::unique_ptr<Channel> _timer_channel;private:void RemoveTimer(uint64_t id){DBG_LOG("REMOVE TIMER");auto it = _timers.find(id);if (it != _timers.end()){_timers.erase(it);}}//通过定时器来进行定时，触发一次读IO事件，保证一秒触发一次时间轮的走动static int CreateTimerfd(){int timerfd = timerfd_create(CLOCK_MONOTONIC, 0);if (timerfd < 0){ERR_LOG("CREATE TIMERFD FAILED");abort();}struct itimerspec itime;itime.it_interval.tv_sec = 1;itime.it_interval.tv_nsec = 0;itime.it_value.tv_nsec = 0;itime.it_value.tv_sec = 1;timerfd_settime(timerfd, 0, &itime, nullptr);DBG_LOG("CREATE TIMERFD SUCCESS :%d",timerfd);return timerfd;}int ReadTimeFd(){uint64_t times;int ret = read(_timerfd, &times, 8);if (ret < 0){ERR_LOG("READ TIME FD:%d FAILED %s",_timerfd, strerror(errno));abort();}return times;}void RunTimerTask(){// DBG_LOG("时间轮被驱动");_tick = (_tick + 1) % _capacity;_timeWheel[_tick].clear();}//定时器的回调读事件，读事件回调会被IO事件触发引发读回调（设置为该OnTime函数），根据超时时间，移动时间轮去执行定时任务void OnTime(){int ret = ReadTimeFd();for (int i = 0; i < ret; i++)RunTimerTask();}public:TimeWheel(EventLoop *loop): _tick(0),_capacity(60),_timeWheel(_capacity),_loop(loop),_timerfd(CreateTimerfd()),_timer_channel(new Channel(_loop, _timerfd)){_timer_channel->SetReadCallBack(std::bind(&TimeWheel::OnTime, this));_timer_channel->EnableRead();}void TimerAdd(uint64_t id, uint64_t delay, 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 true;}return false;}void TimerAddInLoop(uint64_t id, uint64_t timeout, TaskFunc cb){PtrTask pt(new TimerTask(id, timeout, cb)); // 创建任务pt->SetRelease(std::bind(&TimeWheel::RemoveTimer, this, id)); // 设置删除函数回调int pos = (_tick + pt->DelayTime()) % _capacity;              // 找到要插入到时间轮的位置_timeWheel[pos].push_back(pt);                                // 插入时间轮_timers[id] = WeakTask(pt);                                   // 保存定时器的信息}void TimerRefreshInLoop(uint64_t id){auto it = _timers.find(id);if (it == _timers.end()){return; // 没找着定时任务，没法刷新，没法延迟}PtrTask pt = it->second.lock();int pos = (_tick + pt->DelayTime()) % _capacity; // 找到要插入到时间轮的位置_timeWheel[pos].push_back(pt);                   // 插入时间轮// DBG_LOG("时间轮被刷新");}void TimerCancelInLoop(uint64_t id){auto it = _timers.find(id);if (it == _timers.end()){return; // 没找着定时任务，没法刷新，没法延迟}PtrTask pt = it->second.lock();if(pt)pt->Cancel();else DBG_LOG("TIMER CANCEL ERROR! PT IS NULL");DBG_LOG("TIMER CANCEL IN LOOP");}
};

Connection模块

// DISCONNECTED 已经关闭
// CONNECTING 连接建立成功-待处理状态
// CONNECTED 连接已经建立完成，各项设置已就绪，可以通信
// DISCONNECTING 准备断开连接
typedef enum
{DISCONNECTED,CONNECTING,CONNECTED,DISCONNECTING
} ConnStatu;class Connection;
using PtrConnection = std::shared_ptr<Connection>;
class Connection : public std::enable_shared_from_this<Connection>//继承自std::enable_shared_from_this<Connection>，这样就可以在Connection内部获取自己的shared_ptr管理对象
{int _conn_id; // Connection的id// _tiemr_id 定时器id，这里使用conn_id代替int _sockfd;                           // 连接关联的描述符ConnStatu _statu;                      // 连接状态EventLoop *_loop;                      // 连接关联一个EventLoopbool _enable_inactive_release = false; // 是否启动非活跃自动销毁，默认关闭Socket _socket;                        // 对套接字的管理Channel _channel;                      // 对连接事件的管理Buffer _in_buffer;                     // 输入缓冲区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:// 监控可读事件的回调函数，读取缓冲区的数据调用_read_callbackvoid HandleRead(){DBG_LOG("HANDLE READ");// 1. 接收socket的数据，放到缓冲区char buf[65536];ssize_t ret = _socket.NoBlockRecv(buf, 65535);if (ret < 0){// 出错了,不能直接关闭连接DBG_LOG("RECV ERROR AND SHUT DOWN");return ShutDownInLoop();}// 这里的等于0表示的是没有读取到数据，而并不是连接断开了，连接断开返回的是-1// 将数据放入输入缓冲区,写入之后顺便将写偏移向后移动_in_buffer.WriteAndPush(buf, ret);// 2. 调用message_callback进行业务处理if (_in_buffer.ReadAbleSize() > 0){DBG_LOG("_message_callback()");// shared_from_this--从当前对象自身获取自身的shared_ptr管理对象if (_message_callback)return _message_callback(shared_from_this(), &_in_buffer);DBG_LOG("_message_callback()");}DBG_LOG("HANDLE READ END");}// 监控可写事件的回调函数，发送缓冲区中的数据void HandleWrite(){DBG_LOG("HANDLE WRITE");int ret = _socket.NoBlockSend(_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); // 这里应该是将读取数据位置向后偏移，而不是可写，可写是写入数据时移动的DBG_LOG("剩余数据量：%ld,status:%d", _out_buffer.ReadAbleSize(), _statu);if (_out_buffer.ReadAbleSize() == 0){_channel.DisableWrite(); // 没有数据要发送，关闭写事件监控，避免被重复触发// 如果当前是连接待关闭状态，则有数据，发送完数据释放连接，没有数据则直接释放if (_statu == DISCONNECTING){return Release();}}// 如果是待关闭连接状态，那么有数据就先处理数据，没有则直接释放}// 监控关闭事件的回调函数void HandleClose(){DBG_LOG("HANDLE CLOSE");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.调用回调函数// 当前步骤一定是处于半连接状态，否则必然是出错了if (_statu != CONNECTING){ERR_LOG("ESTABLISHED IN LOOP ERROR");abort();}_statu = CONNECTED;_channel.EnableRead();if (_connected_callback)_connected_callback(shared_from_this());}// 实际的释放接口void ReleaseInLoop(){DBG_LOG("REALEASE IN LOOP");// 1.修改连接状态，设为DISCONNECTED_statu = DISCONNECTED;// 2.移除连接事件监控_channel.Remove();// 3.关闭描述符_socket.Close();// DBG_LOG("REALEASE IN LOOP MIDDLE");// 4.如果当前定时器队列中还有任务，则取消任务if (_loop->HasTimer(_conn_id)){CancelInactiveReleaseInLoop();}DBG_LOG("REALEASE IN LOOP MIDDLE");// 5.先调用关闭回调函数，避免先移除服务器内部信息导致Connection被销毁，再处理会出错if (_closed_callback)_closed_callback(shared_from_this());// 移除服务器内部的连接信息if (_server_closed_callback)_server_closed_callback(shared_from_this());DBG_LOG("REALEASE IN LOOP END");}// 并非实际的数据发送接口，只是将数据写入缓冲区，启动写事件void SendInLoop(Buffer &buf){if (_statu == DISCONNECTED)return;_out_buffer.WriteBufferAndPush(buf);if (_channel.WriteAble() == false){_channel.EnableWrite();}}// 并非实际的关闭连接接口，还需要判断数据是否处理完毕void ShutDownInLoop(){DBG_LOG("SHUT DWON IN LOOP");_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){DBG_LOG("BUFFER EMPTY AND RELEASE IN LOOP");Release();}DBG_LOG("SHUT DOWN IN LOOP:%ld", _out_buffer.ReadAbleSize());}// 启动非活跃连接超时void EnableInactiveReleaseInLoop(int sec){_enable_inactive_release = true;// 如果定时销毁任务已经存在，则需要先刷新任务一下if (_loop->HasTimer(_conn_id)){return _loop->TimerRefresh(_conn_id);}// 如果不存在定时销毁任务则新增DBG_LOG("ENABLE INACTIVE RELEASE IN LOOP");_loop->TimerAdd(_conn_id, sec, std::bind(&Connection::Release, this));}void CancelInactiveReleaseInLoop(){_enable_inactive_release = false;if (_loop->HasTimer(_conn_id)){return _loop->TimerCancel(_conn_id);}}void UpgradeInLoop(const Any &context, const ConnectedCallBack &con, const MessageCallBack &msg, const ClosedCallBack &clo, const AnyEventCallBack &evt){_context = context;_connected_callback = con;_message_callback = msg;_closed_callback = clo;_event_callback = evt;}public:Connection(EventLoop *loop, uint64_t coon_id, int sockfd) : _conn_id(coon_id), _sockfd(sockfd), _enable_inactive_release(false), _loop(loop),_statu(CONNECTING), _socket(sockfd), _channel(loop, _sockfd){_channel.SetReadCallBack(std::bind(&Connection::HandleRead, this));_channel.SetWriteCallBack(std::bind(&Connection::HandleWrite, this));_channel.SetCloseCallBack(std::bind(&Connection::HandleClose, this));_channel.SetErrorCallBack(std::bind(&Connection::HandleError, this));_channel.SetEventCallBack(std::bind(&Connection::HandleEvent, this));}~Connection() { DBG_LOG("RELEASE CONNECTION :%p", this); }// 发送数据，将数据放到缓冲区，启动写事件监控void Send(const char *data, size_t len){Buffer buf;buf.WriteAndPush(data, len);_loop->RunInLoop(std::bind(&Connection::SendInLoop, this, std::move(buf)));}int Fd(){return _sockfd;}int Id(){return _conn_id;}// 进行channel回调，启动读监控，调用_connected_callbackvoid Establish(){_loop->RunInLoop(std::bind(&Connection::EstablishedInLoop, this));}// 是否处于连接状态bool Connected(){return _statu;}// 设置上下文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;}// 提供给组件使用者的关闭接口，并不实际关闭，还需要判断数据实际是否处理完毕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));}// 切换协议-重置上下文及处理阶段性数据 非线程安全void Upgrade(const Any &context, const ConnectedCallBack &con, const MessageCallBack &msg, const ClosedCallBack &clo, const AnyEventCallBack &evt){_loop->RunInLoop(std::bind(&Connection::UpgradeInLoop, this, context, con, msg, clo, evt));}
};

Accepter模块

class Accepter
{
private:Socket _socket;   // 创建监听套接字EventLoop *_loop; // 对监听套接字进行事件监控Channel _channel; // 对监听事件进行事件管理using AcceptCallBack = std::function<void(int)>;AcceptCallBack _accept_callback;private:// 监听套接字读事件的回调--获取新连接 回调_accept_callbackvoid HandleRead(){int newfd = _socket.Accept();if (newfd < 0){return;}if (_accept_callback)_accept_callback(newfd);}int CreateServer(int port){bool ret = _socket.CreateServer(port);if (ret == false){abort();}return _socket.Fd();}public:Accepter(EventLoop *loop, int port): _socket(CreateServer(port)),_loop(loop),_channel(loop, _socket.Fd()){_channel.SetReadCallBack(std::bind(&Accepter::HandleRead, this));}void SetAcceptCallBack(const AcceptCallBack &cb){_accept_callback = cb;}/*不能将启动读事件监控，放到构造函数中，必须在设置回调函数后，再去启动否则有可能造成启动监控后，立即有事件，处理的时候，回调函数还没设置：新连接得不到处理，且资源泄漏*/void Listen(){_channel.EnableRead();}
};

LoopThread模块

class LoopThread
{
private:// 用于实现获取Loop的同步,避免线程已经被创建了，但是EventLoop没有被实例化之前就被获取导致的问题std::mutex _mutex;             // 互斥锁std::condition_variable _cond; // 条件变量EventLoop *_loop;              // 实例对象需要在线程内再实例化std::thread _thread;           // EventLoop对应的线程private:// 实例化EventLoop对象，开始运行EventLoop模块的功能void ThreadEntry(){// 实例化对象，唤醒所有被_cond.wait阻塞的线程EventLoop loop;{std::unique_lock<std::mutex> lock(_mutex);_loop = &loop;_cond.notify_all();}loop.Start();}public:// 创建线程，设定线程入口函数LoopThread(): _loop(nullptr),_thread(std::thread(&LoopThread::ThreadEntry, this)){}// 获取EventLoopEventLoop *GetLoop(){// 等待_loop不为空，也就是线程已经被创建并且EventLoop对象已经实例化// 例如主Reactor线程在创建了LoopThread之后就立刻获取EventLoop对象，这时候对象还没有实例化就会导致问题EventLoop *loop = nullptr;{std::unique_lock<std::mutex> lock(_mutex);_cond.wait(lock, [&](){ return _loop != nullptr; });loop = _loop;}return loop;}
};

结合上述的分析所以我们要构造一个新模块

LoopThreadPool模块

// 线程池，用于分发任务到不同的IO线程
// baseloop是主Reactor所在的线程的EventLoop对象，用于分发到其他IO线程
// _threads是所有从Reactor线程的LoopThread对象
// _loops是所有从Reactor线程的EventLoop对象
class LoopThreadPool
{
private:int _thread_count;int _next_idx;EventLoop *_baseloop;std::vector<LoopThread *> _threads;std::vector<EventLoop *> _loops;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();}}}//采用公平轮转分配(RR轮转)EventLoop *NextLoop(){if (_thread_count == 0){return _baseloop;}_next_idx = (_next_idx + 1) % _thread_count;return _loops[_next_idx];}
};

TcpServer模块

class TcpServer
{using Functor = std::function<void()>;private:uint64_t _next_id; // 自增idint _port;int _timeout;                                       // 非活跃连接的超时时间bool _Enable_Inactive_Release;                      // 是否启动非活跃消耗标志Accepter _accepter;                                 // 监听套接字的管理对象EventLoop _baseloop;                                // 主线程的EventLoopLoopThreadPool _pool;                               // 从线程loop的线程池std::unordered_map<uint64_t, PtrConnection> _coons; // 保存管理所有对应连接的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 &)>;ConnectedCallBack _connected_callback;MessageCallBack _message_callback;ClosedCallBack _closed_callback;AnyEventCallBack _event_callback;private:// 为新连接构建一个Connection进行管理void NewConnection(int fd){_next_id++;PtrConnection coon(new Connection(_pool.NextLoop(), _next_id, fd));coon->SetMessageCallBack(_message_callback);coon->SetClosedCallBack(_closed_callback);coon->SetConnectedCallBack(_connected_callback);coon->SetAnyEventCallBack(_event_callback);coon->SetSrvClosedCallBack(std::bind(&TcpServer::RemoveConnection, this, std::placeholders::_1));if (_Enable_Inactive_Release)coon->EnableInactiveRelease(_timeout);coon->Establish();_coons[_next_id] = coon;}void RemoveConnectionInLoop(const PtrConnection &coon){int id = coon->Id();auto it = _coons.find(id);if (it != _coons.end()){_coons.erase(it);}}// 从管理Connection的_conns种移除连接信息void RemoveConnection(const PtrConnection &coon){_baseloop.RunInLoop(std::bind(&TcpServer::RemoveConnectionInLoop, this, coon));}void RunAfterInLoop(const Functor &task, int delay){_next_id++;_baseloop.TimerAdd(_next_id, delay, task);}public:TcpServer(int port): _port(port),_next_id(0),_Enable_Inactive_Release(false),_accepter(&_baseloop, port),_pool(&_baseloop){_accepter.SetAcceptCallBack(std::bind(&TcpServer::NewConnection, this, std::placeholders::_1));_accepter.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();}
};

Http协议模块实现

Util模块

class Util
{
public:// 字符串分割函数 将src按照sep进行分割，将结果放入array，返回分割之后的字符串数量static size_t Split(const std::string &src, const std::string &sep, std::vector<std::string> *array){size_t offset = 0;while (offset < src.size()){// 在src偏移量offset处向后查找字符sep，返回查到的位置size_t pos = src.find(sep, offset);if (pos == std::string::npos){if (pos == src.size())break;array->push_back(src.substr(offset));return array->size();}if (pos == offset){offset = pos + sep.size();continue; // 空串不添加}array->push_back(src.substr(offset, pos - offset));offset = pos + sep.size();}return array->size();}// 读取文件内容static bool ReadFile(const std::string filename, std::string *buf){std::ifstream ifs(filename, std::ios::binary);if (ifs.is_open() == false){ERR_LOG("FILE %s OPEN FAILED", filename.c_str());return false;}size_t fsize = 0;ifs.seekg(0, ifs.end);fsize = ifs.tellg(); // 获取读写位置与文件开头的偏移量，就是文件大小ifs.seekg(0, ifs.beg);buf->resize(fsize);ifs.read(&((*buf)[0]), fsize);if (ifs.good() == false){ERR_LOG("READ %s FILE FAILED", filename.c_str());ifs.close();return false;}ifs.close();return true;}// 写入文件内容static bool WriteFile(const std::string &filename, const std::string &buf){std::ofstream ofs(filename, std::ios::binary | std::ios::trunc);if (ofs.is_open() == false){ERR_LOG("FILE %s OPEN FAILED", filename.c_str());return false;}ofs.write(buf.c_str(), buf.size());if (ofs.good() == false){ERR_LOG("WRITE %s FILE FAILED", filename.c_str());ofs.close();return false;}ofs.close();return true;}// Url编码 避免URL中资源路径与查询字符串中的特殊字符与http请求中的特殊字符产生歧义// RFC3986文档规定，编码格式为%HH,其中也规定 . - _ ~ 数字，字母为绝对不编码字符 如C++ -> C%2B%2B// W3C文档规定，查询字符串中的空格，需要编码为+，查询则是+转空格static std::string UrlEncode(const std::string &url, bool convert_space_to_plus){std::string ret;for (auto c : url){if (c == '.' || c == '-' || c == '_' || c == '~' || isalnum(c)){ret += c;continue;}if (c == ' '){if (convert_space_to_plus == true){ret += '+';continue;}else{ret += ' ';continue;}}char tmp[4] = {0};snprintf(tmp, 4, "%%%02X", c);ret += tmp;}return ret;}static char HEXTOI(char c){if (c >= '0' && c <= '9'){return c - '0';}if (c >= 'a' && c <= 'z'){return c - 'a' + 10;}if (c >= 'A' && c <= 'Z'){return c - 'A' + 10;}return -1;}// Url解码static std::string UrlDecode(const std::string &url, bool convert_space_to_plus){std::string ret;// 遇到百分号就将其后的两个字符重新转化为数字，第一个字符向左移4位+第二个字符for (int i = 0; i < url.size(); i++){if (url[i] == '+' && convert_space_to_plus == true){ret += ' ';continue;}if (url[i] == '%' && i + 2 < url.size()){char val1 = url[i + 1];char val2 = url[i + 2];char v = (val1 << 4) + val2;ret += v;i += 2;continue;}ret += url[i];}return ret;}// 响应状态码的描述信息获取static std::string StatuDesc(int statu){auto it = stu_msg.find(statu);if (it == stu_msg.end()){return "UnKnow";}return it->second;}// 根据文件后缀名获取mime/*** @brief 根据文件名获取文件的MIME类型** 根据提供的文件名，查找并返回该文件对应的MIME类型。** @param filename 文件名* @return 文件的MIME类型，如果无法确定则返回 "application/octet-stream"*/static std::string ExtMime(const std::string &filename){size_t pos = filename.find_last_of('.');if (pos == std::string::npos){return "application/ocet-stream";}std::string ext = filename.substr(pos);auto it = mime_msg.find(ext);if (it == mime_msg.end()){return "application/ocet-stream";}return it->second;}// 判断一个文件是否是目录static bool IsDirectory(const std::string &filename){// 定义结构体stat变量ststruct stat st;// 调用stat函数获取文件状态信息，将结果存储在st中int ret = stat(filename.c_str(), &st);// 如果stat函数返回小于0的值，说明获取文件状态信息失败if (ret < 0){// 返回false，表示该文件不是一个目录return false;}// 返回S_ISDIR(st.st_mode)的结果，如果文件是一个目录，则返回true，否则返回falsereturn S_ISDIR(st.st_mode);}// 判断一个文件是否是普通文件static bool IsRegular(const std::string &filename){struct stat st;int ret = stat(filename.c_str(), &st);if (ret < 0){return false;}return S_ISREG(st.st_mode);}// 判断http请求的资源路径有效性// 防止用户访问到相对根目录之外// 主要思路是控制目录深度static bool ValidPath(const std::string &path){int level = 0;std::vector<std::string> subdir;Split(path, "/", &subdir);for (auto dir : subdir){if (dir == ".."){level--;if (level < 0){return false;}}else{level++;}}return true;}
};class HttpRequest
{
public:std::string _method;                                   // 请求方法std::string _path;                                     // 请求路径std::string _version;                                  // 协议版本std::string _body;                                     // 正文std::smatch _matches;                                  // 资源路径的正则提取路径std::unordered_map<std::string, std::string> _headers; // 头部字段std::unordered_map<std::string, std::string> _params;  // 查询字符串
public:HttpRequest() : _version("HTTP/1.1") {}void ReSet(){_method.clear();_path.clear();_version="HTTP/1.1";_body.clear();std::smatch tmp;_matches.swap(tmp);_headers.clear();_params.clear();}// 插入头部字段void SetHeader(const std::string &key, const std::string &val){DBG_LOG("REQ SET HEADER KEY:%s,Val:%s",key.c_str(),val.c_str());_headers[key] = val;}// 判断是否有头部字段bool HasHeader(const std::string &key) const {auto it = _headers.find(key);if (it == _headers.end()){return false;}return true;}// 获取头部字段std::string GetHeader(const std::string &key) const {auto it = _headers.find(key.c_str());if (it == _headers.end()){DBG_LOG("NOT FIND:%s",key.c_str());return "";}return it->second;}// 插入查询字符串void SetParam(const std::string &key, const std::string &val){_params[key] = val;}// 判断是否有某个查询字符串bool HasParam(const std::string &key){auto it = _params.find(key);if (it == _params.end()){return false;}return true;}// 获取指定查询字符串std::string GetParam(const std::string &key){auto it = _params.find(key);if (it == _params.end()){return "";}return it->second;}// 获取正文长度size_t ContentLenth(){bool ret = HasHeader("Content-Length");if (ret == false){return 0;}std::string len = GetHeader("Content-Length");return std::stol(len);}// 判断是否是短连接bool Close () const{// 没有Connection 或者Connection的值为false都是短连接if (HasHeader("Connection") == true && GetHeader("Connection") == "keep-alive"){DBG_LOG("CLOSE RETURN FALSE");return false;}return true;}
};

HttpResquest模块

class HttpRequest
{
public:std::string _method;                                   // 请求方法std::string _path;                                     // 请求路径std::string _version;                                  // 协议版本std::string _body;                                     // 正文std::smatch _matches;                                  // 资源路径的正则提取路径std::unordered_map<std::string, std::string> _headers; // 头部字段std::unordered_map<std::string, std::string> _params;  // 查询字符串
public:HttpRequest() : _version("HTTP/1.1") {}void ReSet(){_method.clear();_path.clear();_version="HTTP/1.1";_body.clear();std::smatch tmp;_matches.swap(tmp);_headers.clear();_params.clear();}// 插入头部字段void SetHeader(const std::string &key, const std::string &val){DBG_LOG("REQ SET HEADER KEY:%s,Val:%s",key.c_str(),val.c_str());_headers[key] = val;}// 判断是否有头部字段bool HasHeader(const std::string &key) const {auto it = _headers.find(key);if (it == _headers.end()){return false;}return true;}// 获取头部字段std::string GetHeader(const std::string &key) const {auto it = _headers.find(key.c_str());if (it == _headers.end()){DBG_LOG("NOT FIND:%s",key.c_str());return "";}return it->second;}// 插入查询字符串void SetParam(const std::string &key, const std::string &val){_params[key] = val;}// 判断是否有某个查询字符串bool HasParam(const std::string &key){auto it = _params.find(key);if (it == _params.end()){return false;}return true;}// 获取指定查询字符串std::string GetParam(const std::string &key){auto it = _params.find(key);if (it == _params.end()){return "";}return it->second;}// 获取正文长度size_t ContentLenth(){bool ret = HasHeader("Content-Length");if (ret == false){return 0;}std::string len = GetHeader("Content-Length");return std::stol(len);}// 判断是否是短连接bool Close () const{// 没有Connection 或者Connection的值为false都是短连接if (HasHeader("Connection") == true && GetHeader("Connection") == "keep-alive"){DBG_LOG("CLOSE RETURN FALSE");return false;}return true;}
};

HttpResponse模块

class HttpResponse
{
public:int _statu;bool _redirect;std::string _body;std::string _redirect_url;std::unordered_map<std::string, std::string> _headers;public:HttpResponse(): _statu(200),_redirect(false){}HttpResponse(int statu): _statu(statu),_redirect(false){}void ReSet(){_statu = 200;_redirect = false;_body.clear();_redirect_url.clear();_headers.clear();}void SetHeader(const std::string &key, const std::string &val){DBG_LOG("RES SET HEADER KEY:%s,Val:%s",key.c_str(),val.c_str());_headers[key] = val;}bool HasHeader(const std::string &key){auto it = _headers.find(key);if (it == _headers.end()){return false;}return true;}// 获取指定报头std::string GetHeader(const std::string &key){auto it = _headers.find(key);if (it == _headers.end()){return "";}return it->second;}// 设置类型void SetContent(const std::string &body, const std::string &type){_body = body;SetHeader("Content-Type", type);}// 设置重定向void SetRedirect(const std::string &url, int statu = 302){_statu = statu;_redirect = true;_redirect_url = url;}// 判断是否为短连接bool Close(){DBG_LOG("JUDGE CLOSE");// 没有Connection 或者Connection的值为false都是短连接if (HasHeader("Connection") == true && GetHeader("Connection") == "keep-alive"){return false;//长连接}DBG_LOG("TRUE:%s",GetHeader("Connection").c_str());return true;//短连接}
};

HttpContext模块

typedef enum
{RECV_HTTP_ERROR,RECV_HTTP_LINE,RECV_HTTP_HEAD,RECV_HTTP_BODY,RECV_HTTP_OVER
} HttpRecvStatu;#define MAX_LINE 8 * 1024
class HttpContext
{
private:int _resp_statu;           // 响应状态码HttpRecvStatu _recv_statu; // 接收及解析的状态HttpRequest _request;      // 已经解析到的请求信息
private:// 首行接收bool RecvHttpLine(Buffer *buf){DBG_LOG("RecvHttpLine");if (_recv_statu != RECV_HTTP_LINE) return false;//1. 获取一行数据，带有末尾的换行 std::string line = buf->GetLineAndPop();//2. 需要考虑的一些要素：缓冲区中的数据不足一行， 获取的一行数据超大if (line.size() == 0) {//缓冲区中的数据不足一行，则需要判断缓冲区的可读数据长度，如果很长了都不足一行，这是有问题的if (buf->ReadAbleSize() > MAX_LINE) {_recv_statu = RECV_HTTP_ERROR;_resp_statu = 414;//URI TOO LONGreturn false;}//缓冲区中数据不足一行，但是也不多，就等等新数据的到来return true;}if (line.size() > MAX_LINE) {_recv_statu = RECV_HTTP_ERROR;_resp_statu = 414;//URI TOO LONGreturn false;}bool ret = ParseHttpLine(line);if (ret == false) {return false;}//首行处理完毕，进入头部获取阶段_recv_statu = RECV_HTTP_HEAD;return true;}// 首行解析bool ParseHttpLine(const std::string &line){DBG_LOG("ParseHttpLine 1 %s",line.c_str());std::smatch matches;std::regex e("(GET|HEAD|POST|PUT|DELETE) ([^?]*)(?:\\?(.*))? (HTTP/1\\.[01])(?:\n|\r\n)?", std::regex::icase);ERR_LOG("LINE=%s",line.c_str());bool ret = std::regex_match(line, matches, e);if (ret == false){_recv_statu = RECV_HTTP_ERROR;_resp_statu = 400; // BAD REQUESTreturn false;}// 0 : GET /juhuo/login?user=xiaoming&pass=123123 HTTP/1.1// 1 : GET// 2 : /juhuo/login// 3 : user=xiaoming&pass=123123// 4 : HTTP/1.1// 请求方法的获取_request._method = matches[1];std::transform(_request._method.begin(), _request._method.end(), _request._method.begin(), ::toupper);// 资源路径的获取，需要进行URL解码操作，但是不需要+转空格_request._path = Util::UrlDecode(matches[2], false);// 协议版本的获取_request._version = matches[4];// 查询字符串的获取与处理std::vector<std::string> query_string_arry;std::string query_string = matches[3];// 查询字符串的格式 key=val&key=val....., 先以 & 符号进行分割，得到各个字串Util::Split(query_string, "&", &query_string_arry);// 针对各个字串，以 = 符号进行分割，得到key 和val， 得到之后也需要进行URL解码for (auto &str : query_string_arry){size_t pos = str.find("=");if (pos == std::string::npos){_recv_statu = RECV_HTTP_ERROR;_resp_statu = 400; // BAD REQUESTreturn false;}std::string key = Util::UrlDecode(str.substr(0, pos), true);std::string val = Util::UrlDecode(str.substr(pos + 1), true);_request.SetParam(key, val);}DBG_LOG("ParseHttpLine 4");return true;}// 解析报头bool RecvHttpHead(Buffer *buf){if (_recv_statu != RECV_HTTP_HEAD)//if (_recv_statu != RECV_HTTP_LINE)return false;// 一行一行解析数据，遇到空行结束while (1){std::string line = buf->GetLineAndPop();if (line.size() == 0){// 如果缓冲区数据不足一行，且缓冲区数据过大，则url可能存在问题if (buf->ReadAbleSize() > MAX_LINE){_recv_statu = RECV_HTTP_ERROR;_resp_statu = 414; // url too longreturn false;}// 数据不足一行，但是数据不多则返回等待新数据return true;}if (line.size() > MAX_LINE){_recv_statu = RECV_HTTP_ERROR;_resp_statu = 414; // url too longreturn false;}// 解析完毕，结束解析if (line == "\n" || line == "\r\n"){break;}bool ret = ParseHttpHead(line);if (ret = false){return false;}}// 头部处理完毕进入正文阶段_recv_statu = RECV_HTTP_BODY;return true;}//解析报头bool ParseHttpHead(std::string &line){if (line.back() == '\n') line.pop_back();//末尾是换行则去掉换行字符if (line.back() == '\r') line.pop_back();//末尾是回车则去掉回车字符size_t pos = line.find(": ");if (pos == std::string::npos){_recv_statu = RECV_HTTP_ERROR;_resp_statu = 400; // BAD REQUESTreturn false;}std::string key = Util::UrlDecode(line.substr(0, pos), true);std::string val = Util::UrlDecode(line.substr(pos + 2), true);DBG_LOG("Parse Http Head:KEY:%s,Val:%s",key.c_str(),val.c_str());_request.SetHeader(key, val);return true;}bool RecvHttpBody(Buffer *buf){if (_recv_statu != RECV_HTTP_BODY)return false;size_t content_length = _request.ContentLenth();if (content_length == 0){// 没有正文则请求结束_recv_statu = RECV_HTTP_OVER;return true;}// 获取正文长度// 当前已经接收多少正文size_t real_length = content_length - _request._body.size(); // 实际还要接收的数据长度// 接收正文到body中，也要考虑当前的缓冲区数据是否是一个正文// 若恰好包含则直接提取if (buf->ReadAbleSize() >= real_length){_request._body.append(buf->ReadPosition(), real_length);buf->MoveReadOffset(real_length);_recv_statu = RECV_HTTP_OVER;return true;}// 若不足则等待新数据_request._body.append(buf->ReadPosition(), buf->ReadAbleSize());buf->MoveReadOffset(buf->ReadAbleSize());return true;}public:HttpContext(): _resp_statu(200),_recv_statu(RECV_HTTP_LINE){}void ReSet(){_resp_statu = 200;_recv_statu = RECV_HTTP_LINE;_request.ReSet();}int ResponseStatu(){return _resp_statu;}HttpRecvStatu RecvStatu(){return _recv_statu;}HttpRequest &Request(){return _request;}void RecvHttpRequest(Buffer *buf){DBG_LOG("RecvHttpRequest");switch (_recv_statu){// 此处不需要break,本身就需要按照line,head,body依次接收解析！case RECV_HTTP_LINE:RecvHttpLine(buf);case RECV_HTTP_HEAD:DBG_LOG("RECV_HTTP_HEAD");RecvHttpHead(buf);case RECV_HTTP_BODY:RecvHttpBody(buf);}}
};

HttpServer模块

class HttpServer {private:using Handler = std::function<void(const HttpRequest &, HttpResponse *)>;using Handlers = std::vector<std::pair<std::regex, Handler>>;Handlers _get_route;Handlers _post_route;Handlers _put_route;Handlers _delete_route;std::string _basedir; //静态资源根目录TcpServer _server;private:void ErrorHandler(const HttpRequest &req, HttpResponse *rsp) {//1. 组织一个错误展示页面std::string body;body += "<html>";body += "<head>";body += "<meta http-equiv='Content-Type' content='text/html;charset=utf-8'>";body += "</head>";body += "<body>";body += "<h1>";body += std::to_string(rsp->_statu);body += " ";body += Util::StatuDesc(rsp->_statu);body += "</h1>";body += "</body>";body += "</html>";//2. 将页面数据，当作响应正文，放入rsp中rsp->SetContent(body, "text/html");}//将HttpResponse中的要素按照http协议格式进行组织，发送void WriteResponse(const PtrConnection &conn, const HttpRequest &req, HttpResponse &rsp) {//1. 先完善头部字段DBG_LOG("WRITE RESPONSE!!!");if (req.Close() == true) {rsp.SetHeader("Connection", "close");}else {rsp.SetHeader("Connection", "keep-alive");}if (rsp._body.empty() == false && rsp.HasHeader("Content-Length") == false) {rsp.SetHeader("Content-Length", std::to_string(rsp._body.size()));}if (rsp._body.empty() == false && rsp.HasHeader("Content-Type") == false) {rsp.SetHeader("Content-Type", "application/octet-stream");}if (rsp._redirect == true) {rsp.SetHeader("Location", rsp._redirect_url);}//2. 将rsp中的要素，按照http协议格式进行组织std::stringstream rsp_str;rsp_str << req._version << " " << std::to_string(rsp._statu) << " " << Util::StatuDesc(rsp._statu) << "\r\n";for (auto &head : rsp._headers) {rsp_str << head.first << ": " << head.second << "\r\n";}rsp_str << "\r\n";rsp_str << rsp._body;//3. 发送数据conn->Send(rsp_str.str().c_str(), rsp_str.str().size());}/*** @brief 判断请求是否是文件处理请求** 该函数用于判断传入的 HTTP 请求是否是一个文件处理请求。** @param req HTTP 请求对象* @return 如果请求是文件处理请求，则返回 true；否则返回 false。*/bool IsFileHandler(const HttpRequest &req) {// 1. 必须设置了静态资源根目录if (_basedir.empty()) {return false;}// 2. 请求方法，必须是GET / HEAD请求方法if (req._method != "GET" && req._method != "HEAD") {return false;}// 3. 请求的资源路径必须是一个合法路径DBG_LOG("hahah: %s",req._path.c_str());if (Util::ValidPath(req._path) == false) {return false;}// 4. 请求的资源必须存在,且是一个普通文件//    有一种请求比较特殊 -- 目录：/, /image/， 这种情况给后边默认追加一个 index.html// index.html    /image/a.png// 不要忘了前缀的相对根目录,也就是将请求路径转换为实际存在的路径  /image/a.png  ->   ./wwwroot/image/a.pngstd::string req_path = _basedir + req._path;//为了避免直接修改请求的资源路径，因此定义一个临时对象if (req._path.back() == '/')  {req_path += "index.html";}if (Util::IsRegular(req_path) == false) {return false;}return true;}//静态资源的请求处理 --- 将静态资源文件的数据读取出来，放到rsp的_body中, 并设置mimevoid FileHandler(const HttpRequest &req, HttpResponse *rsp) {std::string req_path = _basedir + req._path;if (req._path.back() == '/')  {req_path += "index.html";}bool ret = Util::ReadFile(req_path, &rsp->_body);if (ret == false) {return;}std::string mime = Util::ExtMime(req_path);rsp->SetHeader("Content-Type", mime);return;}//功能性请求的分类处理void Dispatcher(HttpRequest &req, HttpResponse *rsp, Handlers &handlers) {// 在对应请求方法的路由表中，查找是否含有对应资源请求的处理函数，有则调用，没有则发挥404// 思想：路由表存储的时键值对 -- 正则表达式 & 处理函数// 使用正则表达式，对请求的资源路径进行正则匹配，匹配成功就使用对应函数进行处理//   /numbers/(\d+)       /numbers/12345DBG_LOG("ON Dispatcher");// 遍历所有的处理器for (auto &handler : handlers) {// 获取正则表达式const std::regex &re = handler.first;// 获取处理函数const Handler &functor = handler.second;// 使用正则表达式对请求的资源路径进行正则匹配bool ret = std::regex_match(req._path, req._matches, re);// 如果匹配失败，则继续遍历下一个处理器if (ret == false) {continue;}// 匹配成功，调用处理函数，传入请求信息和空的响应对象return functor(req, rsp);//传入请求信息，和空的rsp，执行处理函数}// 如果遍历完所有处理器都没有匹配到对应的处理函数，则设置响应状态为404rsp->_statu = 404;}void Route(HttpRequest &req, HttpResponse *rsp) {//1. 对请求进行分辨，是一个静态资源请求，还是一个功能性请求//   静态资源请求，则进行静态资源的处理//   功能性请求，则需要通过几个请求路由表来确定是否有处理函数//   既不是静态资源请求，也没有设置对应的功能性请求处理函数，就返回405DBG_LOG("Route");if (IsFileHandler(req) == true) {//是一个静态资源请求, 则进行静态资源请求的处理return FileHandler(req, rsp);}if (req._method == "GET" || req._method == "HEAD") {return Dispatcher(req, rsp, _get_route);}else if (req._method == "POST") {return Dispatcher(req, rsp, _post_route);}else if (req._method == "PUT") {return Dispatcher(req, rsp, _put_route);}else if (req._method == "DELETE") {return Dispatcher(req, rsp, _delete_route);}rsp->_statu = 405;// Method Not Allowedreturn ;}//设置上下文void OnConnected(const PtrConnection &conn) {conn->SetContext(HttpContext());DBG_LOG("NEW CONNECTION %p", conn.get());}//缓冲区数据解析+处理void OnMessage(const PtrConnection &conn, Buffer *buffer) {while(buffer->ReadAbleSize() > 0){//1. 获取上下文DBG_LOG("ON MESSAGE 1");HttpContext *context = conn->GetContext()->get<HttpContext>();//2. 通过上下文对缓冲区数据进行解析，得到HttpRequest对象//  1. 如果缓冲区的数据解析出错，就直接回复出错响应//  2. 如果解析正常，且请求已经获取完毕，才开始去进行处理context->RecvHttpRequest(buffer);HttpRequest &req = context->Request();HttpResponse rsp(context->ResponseStatu());if (context->ResponseStatu() >= 400) {//进行错误响应，关闭连接DBG_LOG("ON MESSAGE ERROR");ErrorHandler(req, &rsp);//填充一个错误显示页面数据到rsp中WriteResponse(conn, req, rsp);//组织响应发送给客户端context->ReSet();buffer->MoveReadOffset(buffer->ReadAbleSize());//出错了就把缓冲区数据清空conn->ShutDown();//关闭连接return;}if (context->RecvStatu() != RECV_HTTP_OVER) {//当前请求还没有接收完整,则退出，等新数据到来再重新继续处理return;}//3. 请求路由 + 业务处理Route(req, &rsp);//4. 对HttpResponse进行组织发送WriteResponse(conn, req, rsp);//5. 重置上下文context->ReSet();//6. 根据长短连接判断是否关闭连接或者继续处理if (rsp.Close() == true) {DBG_LOG("rsp.close()");conn->ShutDown();//短链接则直接关闭}}return;}public:HttpServer(int port, int timeout = DEFALT_TIMEOUT):_server(port) {_server.EnableInactiveRelease(timeout);_server.SetConnectedCallBack(std::bind(&HttpServer::OnConnected, this, std::placeholders::_1));_server.SetMessageCallBack(std::bind(&HttpServer::OnMessage, this, std::placeholders::_1, std::placeholders::_2));}void SetBaseDir(const std::string &path) {if(Util::IsDirectory(path) == true)_basedir = path;else abort();}/*设置/添加，请求（请求的正则表达）与处理函数的映射关系*/void Get(const std::string &pattern, const Handler &handler) {_get_route.push_back(std::make_pair(std::regex(pattern), handler));}void Post(const std::string &pattern, const Handler &handler) {_post_route.push_back(std::make_pair(std::regex(pattern), handler));}void Put(const std::string &pattern, const Handler &handler) {_put_route.push_back(std::make_pair(std::regex(pattern), handler));}void Delete(const std::string &pattern, const Handler &handler) {_delete_route.push_back(std::make_pair(std::regex(pattern), handler));}void SetThreadCount(int count) {_server.SetThreadCount(count);}void Listen() {_server.Start();}
};

简单服务器搭建使用样例

#include "http.hpp"#define WWWROOT "./wwwroot/"std::string RequestStr(const HttpRequest &req)
{std::stringstream ss;ss << req._method << " " << req._path << " " << req._version << "\r\n";for (auto &it : req._params){ss << it.first << ": " << it.second << "\r\n";}for (auto &it : req._headers){ss << it.first << ": " << it.second << "\r\n";}ss << "\r\n";ss << req._body;return ss.str();
}
void Hello(const HttpRequest &req, HttpResponse *rsp)
{rsp->SetContent(RequestStr(req), "text/plain");//sleep(15);
}
void Login(const HttpRequest &req, HttpResponse *rsp)
{rsp->SetContent(RequestStr(req), "text/plain");
}
void PutFile(const HttpRequest &req, HttpResponse *rsp)
{rsp->SetContent(RequestStr(req), "text/plain");std::string pathname = WWWROOT + req._path;Util::WriteFile(pathname, req._body);
}
void DelFile(const HttpRequest &req, HttpResponse *rsp)
{rsp->SetContent(RequestStr(req), "text/plain");
}
int main()
{HttpServer server(8085);server.SetThreadCount(3);server.SetBaseDir(WWWROOT); // 设置静态资源根目录，告诉服务器有静态资源请求到来，需要到哪里去找资源文件server.Get("/hello", Hello);server.Post("/login", Login);server.Put("/1234.txt", PutFile);server.Delete("/1234.txt", DelFile);INF_LOG("服务器设置完毕");server.Listen();return 0;
}