C项目实战C项目实战

Protocol超⽂本传输协议是应⽤层协议是⼀种简单的请求-响应协议客⼾端根据⾃⼰的需要向服务器发送请求服务器针对请求提供服务完毕后通信结束。

实现⼀个HTTP服务器很简单但是实现⼀个⾼性能的服务器并不简单这个单元中将讲解基于Reactor模式的⾼性能服务器实现。

当然准确来说因为我们要实现的服务器本⾝并不存在业务咱们要实现的应该算是⼀个⾼性能服务器基础库是⼀个基础组件。

二、Reactor模型

模式是指通过⼀个或多个输⼊同时传递给服务器进⾏请求处理时的事件驱动处理模式。

服务端程序处理传⼊多路请求并将它们同步分派给请求对应的处理线程Reactor

模式也叫Dispatcher

统⼀监听事件收到事件后分发给处理进程或线程是编写⾼性能⽹络服务器的必备技术之⼀。

触发事件后进⾏事件处理

如果是新建连接请求则获取新建连接并添加⾄多路复⽤模型进⾏事件监控。

如果是数据通信请求则进⾏对应数据处理接收数据处理数据发送响应。

优点所有操作均在同⼀线程中完成思想流程较为简单不涉及进程/线程间通信及资源争抢问题。

适⽤场景适⽤于客⼾端数量较少且处理速度较为快速的场景。

处理较慢或活跃连接较多会导致串⾏处理的情况下后处理的连接⻓时间⽆法得到响应).

触发事件后进⾏事件处理

如果是新建连接请求则获取新建连接并添加⾄多路复⽤模型进⾏事件监控。

如果是数据通信请求则接收数据后分发给Worker线程池进⾏业务处理。

⼯作线程处理完毕后将响应交给Reactor线程进⾏数据响应

优点充分利⽤CPU多核资源

承担所有事件的监听和响应在单线程中运⾏⾼并发场景下容易成为性能瓶颈。

在主Reactor中处理新连接请求事件有新连接到来则分发到⼦Reactor中监控

在⼦Reactor中进⾏客⼾端通信监控有事件触发则接收数据分发给Worker线程池

我要实现的是主从Reactor模型服务器也就是主Reactor线程仅仅监控监听描述符获取新建连接保证获取新连接的⾼效性提⾼服务器的并发性能。

主Reactor获取到新连接后分发给⼦Reactor进⾏通信事件监控。

⽽⼦Reactor线程监控各⾃的描述符的读写事件进⾏数据读写以及业务处理。

该项目从Reactor主要作用IO事件监控IO操作业务处理比较轻量。

当前实现中因为并不确定组件使⽤者的使⽤意向因此并不提供业务层⼯作线程池的实现只实现主从Reactor⽽Worker⼯作线程池可由组件库的使⽤者的需要⾃⾏决定是否使⽤和实现。

三、功能模块划分

我要实现的是⼀个带有协议⽀持的Reactor模型⾼性能服务器因此将整个项⽬的实现划分为两个⼤的模块

SERVER模块

SERVER模块就是对所有的连接以及线程进⾏管理让它们各司其职在合适的时候做合适的事最终

基于以上的管理思想将这个模块进⾏细致的划分⼜可以划分为以下多个⼦模块

Buffer子模块

Buffer模块是⼀个缓冲区模块⽤于实现通信中⽤⼾态的接收缓冲区和发送缓冲区功能

意义1、防止接收到的数据不是一条完整的数据因此对接收的数据进行缓冲

功能设计

Socket模块是对套接字操作封装的⼀个模块主要实现的socket的各项操作。

功能设计

Channel模块是对⼀个描述符需要进⾏的IO事件管理的模块实现对描述符可读可写错误…事件的管理操作以及Poller模块对描述符进⾏IO事件监控就绪后根据不同的事件回调不同的处理函数功能。

意义对于描述符的监控事件在用户态更容易维护以及触发事件后的操作流程更加的清晰

功能设计

设置对于不同事件的回调处理函数明确触发了某个事件之后应该怎么处理

Connection模块

这是一个对于通信连接进行整体管理的一个模块对一个连接的操作都是通过这个模块进行的​

Connection模块一个连接有任何的事件该怎么处理都是由这个模块来进行处理的因为组件的设计也不知道使用者要如何处理事

Connection模块内部包含有三个由组件使⽤者传⼊的回调函数连接建⽴完成回调事件回调新数据回调关闭回调。

Connection模块内部包含有两个组件使⽤者提供的接⼝数据发送接⼝连接关闭接⼝

Connection模块内部包含有两个⽤⼾态缓冲区⽤⼾态接收缓冲区⽤⼾态发送缓冲区

Connection模块内部包含有⼀个Socket对象完成描述符⾯向系统的IO操作

Connection模块内部包含有⼀个Channel对象完成描述符IO事件就绪的处理

具体处理流程如下

实现向Channel提供可读可写错误等不同事件的IO事件回调函数然后将Channel和对应的描述符添加到Poller事件监控中。

当描述符在Poller模块中就绪了IO可读事件则调⽤描述符对应Channel中保存的读事件处理函数进⾏数据读取将socket接收缓冲区全部读取到Connection管理的⽤⼾态接收缓冲区中。

然后调⽤由组件使⽤者传⼊的新数据到来回调函数进⾏处理。

组件使⽤者进⾏数据的业务处理完毕后通过Connection向使⽤者提供的数据发送接⼝将数据写⼊Connection的发送缓冲区中。

启动描述符在Poll模块中的IO写事件监控就绪后调⽤Channel中保存的写事件处理函数将发送缓冲区中的数据通过Socket进⾏⾯向系统的实际数据发送。

Acceptor模块

Acceptor模块是对Socket模块Channel模块的⼀个整体封装实现了对⼀个监听套接字的整体的管理。

Acceptor模块内部包含有⼀个Socket对象实现监听套接字的操作

Acceptor模块内部包含有⼀个Channel对象实现监听套接字IO事件就绪的处理

具体处理流程如下

实现向Channel提供可读事件的IO事件处理回调函数函数的功能其实也就是获取新连接为新连接构建⼀个Connection对象出来。

当获取了一个新建连接的描述符之后需要为这个通信连接封装一个Connection对象设置各种不同回调

因为Acceptor模块本身并不知道一个连接产生了某个事件该如何处理因此获取一个通信连接后Connection的封装以及事件回调的设置都应该由服务器模块来进行

TimerQueue模块

TimerQueue模块是实现固定时间定时任务的模块可以理解就是要给定时任务管理器向定时任务管理器中添加⼀个任务任务将在固定时间后被执⾏同时也可以通过刷新定时任务来延迟任务的执⾏

这个模块主要是对Connection对象的⽣命周期管理对⾮活跃连接进⾏超时后的释放功能。

TimerQueue模块内部包含有⼀个timerfdlinux系统提供的定时器。

TimerQueue模块内部包含有⼀个Channel对象实现对timerfd的IO时间就绪回调处理

功能设计添加定时任务、刷新定时任务、希望一个定时任务重新开始计时、取消定时任务

Poller模块

Poller模块是对epoll进⾏封装的⼀个模块主要实现epoll的IO事件添加修改移除获取活跃连接功能。

意义对epoll进行的封装让对描述符进行事件监控的操作更加简单

功能接口添加事件监控、Channel模块、修改事件监控、移除事件监控

EventLoop模块

EventLoop模块可以理解就是我们上边所说的Reactor模块它是对Poller模块TimerQueue模块Socket模块的⼀个整体封装进⾏所有描述符的事件监控。

意义对于服务器中的所有的事件都是由EventLoop模块来完成

每一个Connection连接都会绑定一个EventLoop模块和线程因为外界对于连接的所有操作都是要放到同一个线程中进行的

2、连接触发事件后调用回调进行处理

功能设计将对连接的操作任务添加到任务队列、定时任务的添加、定时任务的刷新、定时任务的取消

具体操作流程

1、通过Poller模块对当前模块管理内的所有描述符进⾏IO事件监控有描述符事件就绪后通过描述符对应的Channel进⾏事件

2、所有就绪的描述符IO事件处理完毕后对任务队列中的所有操作顺序进⾏执⾏。

由于epoll的事件监控有可能会因为没有事件到来⽽持续阻塞导致任务队列中的任务不能及时得到执⾏因此创建了

eventfd添加到Poller的事件监控中⽤于实现每次向任务队列添加任务的时候通过向eventfd写⼊数据来唤醒epoll的阻塞

TcpServer模块

这个模块是⼀个整体Tcp服务器模块的封装内部封装了Acceptor模块EventLoopThreadPool模块。

TcpServer中包含有⼀个EventLoop对象在超轻量使⽤场景中不需要EventLoop线程池只需要在主线程中完成所有操作的情况。

TcpServer模块内部包含有⼀个EventLoopThreadPool对象其实就是EventLoop线程池也就是⼦Reactor线程池

TcpServer模块内部包含有⼀个Acceptor对象⼀个TcpServer服务器必然对应有⼀个监听套接字能够完成获取客⼾端新连接并处理的任务。

TcpServer模块内部包含有⼀个std::shared_ptr的hash表保存了所有的新建连接对应的Connection注意所有的Connection使⽤shared_ptr进⾏管理这样能够保证在hash表中删除了Connection信息后在shared_ptr计数器为0的情况下完成对Connection资源的释放操作。

功能对于监听连接的管理获取一个新连接之后如何处理由Server模块设置

对于通信连接的管理连接产生的某个事件如何处理由Server模块设置

对于超时连接的管理连接非活跃超时是否关闭由Server模块设置

对于事件监控的管理启动多少个线程有多少个EventLoop由Server设置

事件回调函数的设置一个连接产生了一个事件对于这个事件如何处理只有组件使用者知道因此一个事件的处理回调一定

是组件使用者设置给TcpServerTcpServer设置给各个Connection连接

具体流程如下

在实例化TcpServer对象过程中完成BaseLoop的设置Acceptor对象的实例化以及EventLoop线程池的实例化以及std::shared_ptr的hash表的实例化。

为Acceptor对象设置回调函数获取到新连接后为新连接构建Connection对象设置Connection的各项回调并使⽤shared_ptr进⾏管理并添加到hash表中进⾏管理并为Connection选择⼀个EventLoop线程为Connection添加⼀个定时销毁任务为Connection添加事件监控

四、Server代码实现

args);我们可以将bind接⼝看作是⼀个通⽤的函数适配器它接受⼀个函数对象以及函数的各项参数然后返回⼀个新的函数对象但是这个函数对象的参数已经被绑定为设置的参数。

运⾏的时候相当于总是调⽤传⼊固定参数的原函数。

_2...

在当前的⾼并发服务器中我们不得不考虑⼀个问题那就是连接的超时关闭问题。

我们需要避免⼀个连接⻓时间不通信但是也不关闭空耗资源的情况。

这时候我们就需要⼀个定时任务定时的将超时过期的连接进⾏释放。

#include

CLOCK_REALTIME-系统实时时间如果修改了系统时间就会出问题

CLOCK_MONOTONIC-从开机到现在的时间是⼀种相对时间flags:

0-默认阻塞属性返回值小于0则为错误

定时器会在每次超时时⾃动给fd中写⼊8字节的数据表⽰在上⼀次读取数据到当前读取数据期间超时了多少次。

示例:

#include

timerfd_create(CLOCK_MONOTONIC,

0);if(timerfd

第一次超时时间为1s后itime.it_interval.tv_sec

//第一次超时后每次超时的间隔timerfd_settime(timerfd,

itime,

-1;}printf(超时了距离上一次超时了%d次\n,times);}close(timerfd);return

}上边例⼦是⼀个定时器的使⽤⽰例是每隔3s钟触发⼀次定时器超时否则就会阻塞在read读取数据这⾥

时间轮思想

上述的例⼦存在⼀个很⼤的问题每次超时都要将所有的连接遍历⼀遍如果有上万个连接效率⽆疑是较为低下的。

时间轮的思想来源于钟表如果我们定了⼀个3点钟的闹铃则当时针⾛到3的时候就代表时间到了。

同样的道理如果我们定义了⼀个数组并且有⼀个指针指向数组起始位置这个指针每秒钟向后⾛动⼀步⾛到哪⾥则代表哪⾥的任务该被执⾏了那么如果我们想要定⼀个3s后的任务则只需要将任务添加到tick3位置则每秒中⾛⼀步三秒钟后tick⾛到对应位置这时候执⾏对应位置的任务即可。

但是同⼀时间可能会有⼤批量的定时任务因此我们可以给数组对应位置下拉⼀个数组这样就可

上述操作也有⼀些缺陷⽐如我们如果要定义⼀个60s后的任务则需要将数组的元素个数设置为60才可以如果设置⼀⼩时后的定时任务则需要定义3600个元素的数组这样⽆疑是⽐较⿇烦的。

因此可以采⽤多层级的时间轮有秒针轮分针轮时针轮当指针指向了时针轮所对应的位置的时候那么就会像分针轮进行移动当指针指向了分针轮所对应的位置的时候指针就会向秒针轮进行移动。

因为当前我们的应⽤中倒是不⽤设计的这么⿇烦因为我们的定时任务通常设置的30s以内所以简单的单层时间轮就够⽤了。

但是我们也得考虑⼀个问题当前的设计是时间到了则主动去执⾏定时任务释放连接那能不能在时间到了后⾃动执⾏定时任务呢

作为一个时间轮定时器本身并不关注任务类型只要是时间到了就需要被执行。

智能指针share_ptr,

1、使用一个类对定时任务进行封装类实例化的每一个对象就是一个定时任务对象当对象被销毁的时候再去执行定时任务

2、shared_ptr用于对new的对象进行空间管理当shared_ptr对一个对象进行管理的时候内部有一个计数器计数器为0的时候

int

--当针对pi又构建了一个shared_ptr对象则pi和pi1计数器为2

当pi和pi1中任意一个被释放的时候只是计数器-1因此他们管理的a对象并没有被释放

只有当pi和pi1都被释放了计数器为0了这时候才会释放管理的a

示例代码

cb):_id(id),_timeout(delay),_task_cb(cb),_canceled(false){}void

SetRelease(const

true;}~TimerTask(){if(_canceled

false)

_task_cb();//定时任务没有被取消才会执行_release();}

private:uint64_t

_task_cb;//定时器对象要执行的定时任务ReleaseFunc

_release;//用于删除TimerWheel中保存的定时器对象信息bool

_canceled;

public:TimerWheel():_capacity(60),_tick(0),_wheel(_capacity){}void

TimerAdd(u_int64_t

TimerTask(id,delay,cb));pt-SetRelease(std::bind(TimerWheel::RemoveTimer,this,id));_timers[id]

WeakTask(pt);int

_capacity;_wheel[pos].push_back(pt);}//刷新/延迟定时任务void

id){//通过保存我的定时器对象的weak_ptr构造一个shared_ptr出来添加到轮子中auto

_timers.end()){return;//没有找到定时任务没法刷新没法延迟}PtrTask

it-second.lock();//lock获取weak_ptr管理的对象对应的shared_ptrint

delay

_capacity;_wheel[pos].push_back(pt);}//这个函数应该每秒钟执行一次相当于秒针向后走了一步void

RunTimerTask(){_tick

_capacity;_wheel[_tick].clear();}//取消定时任务void

TimerCancel(uint64_t

_timers.end()){return;//没有找到定时任务没法刷新没法延迟}PtrTask

pt-Cancel();}

_timers.end()){_timers.erase(it);}}

private:using

//表盘最大数量---其实就是最大延迟时间std::vectorstd::vectorPtrTask

_wheel;std::unordered_mapuint64_t,

WeakTask

Test();tw.TimerAdd(888,5,std::bind(DelTest,

t));for(int

i){sleep(1);tw.TimerRefresh(888);//刷新定时任务tw.RunTimerTask();cout

endl;}tw.TimerCancel(888);//取消定时任务就不会被销毁了while(1){sleep(1);cout

-----------------

expression)描述了⼀种字符串匹配的模式pattern可以⽤来检查⼀个串是否含有某种⼦串、将匹配的⼦串替换或者从某个串中取出符合某个条件的⼦串等。

std::string

正则表达式可以从原始字符串中匹配并提取符合某种规则的数据提取的数据就放在matches中是一个类似于数组的容器e正则表达式的匹配规则返回值用于确定匹配是否成功示例代码

#include

/numbers/1234;//需要提取数字字符串//匹配以/number/起始后边跟了一个或多个数字字符的字符串并且在匹配过程中提取这个匹配到的数字字符串std::regex

matches;bool

std::regex_match(str,matches,e);if(ret

false){std::cout

/bitejiuyeke/login?userxiaomingpass123123

str

/bitejiuyeke/login?userxiaomingpass123123

HTTP/1.1;std::smatch

匹配除“\n”和\r之外的任何单个字符。

要匹配包括“\n”和\r在内的任何字符请使用像“[\s\S]”的模式。

//星

匹配前面的子表达式任意次。

例如zo*能匹配“z”也能匹配“zo”以及“zoo”。

*等价于{0,}。

//\\?(.*)

\\?

非获取匹配匹配pattern但不获取匹配结果不进行存储供以后使用。

最后的表示的是匹配前边的表达式0次或1次std::regex

([^?]*)(?:\\?(.*))?

(HTTP/1\\.[01])(?:\n|\r\n)?);bool

ret

每⼀个Connection对连接进⾏管理最终都不可避免需要涉及到应⽤层协议的处理因此在Connection中需要设置协议处理的上下⽂来控制处理节奏。

但是应⽤层协议千千万为了降低耦合度这个协议接收解析上下⽂就不能有明显的协议倾向它可以是任意协议的上下⽂信息因此就需要⼀个通⽤的类型来保存各种不同的数据结构。

在C语⾔中通⽤类型可以使⽤void*来管理但是在C中boost库和C17给我们提供了⼀个通⽤类型any来灵活使⽤如果考虑增加代码的移植性尽量减少第三⽅库的依赖则可以使⽤C17特性中的any或者⾃⼰来实现。

2、上下文的类型或者说结构不能固定因为服务器支持的协议有可能会不断增多不同的协议可能都会有不同的上下文结构

解决方案

Any类主要是实现⼀个通⽤类型出来在c17和boost库中都有现成的可以使⽤但是这⾥实现⼀下了解其思想这样也就避免了第三⽅库的使⽤了⾸先Any类肯定不能是⼀个模板类否则编译的时候

Any

Anyb,需要传类型作为模板参数也就是说在使⽤的时候就要确定其类型这是⾏不通的因为保存在Content中的协议上下⽂我们在定义any对象的时候是不知道他们的协议类型的因此⽆法传递类型作为模板参数因此考虑Any内部设计⼀个模板容器holder类可以保存各种类型数据⽽因为在Any类中⽆法定义这个holder对象或指针因为any也不知道这个类要保存什么类型的数据因此⽆法传递类型参数所以定义⼀个基类placehoder让holder继承于placeholde⽽Any类保存⽗类指针即可当需要保存数据时则new⼀个带有模板参数的⼦类holder对象出来保存数据然后让Any类中的⽗类指针指向这个⼦类对象就搞定了

示例代码

holder{public:placeholder(const

Tval):_val(val)

typeid(T);}//针对当前的对象自身克隆出一个新的子类对象virtual

holder*

other){std::swap(_content,other._content);return

*this;}templateclass

get()//返回子类对象保存的数据的指针{//想要获取的数据类型必须和保存的数据类型一致assert(typeid(T)

((placeholderT*)_content)-_val;}templateclass

TAny

val)//赋值运算符的重载函数{Any(val).Swap(*this);return

*this;}Any

other){Any(other).Swap(*this);return

*this;}

1、实现缓冲区得有一块内存空间采用vector,vector底层其实使用的就是一个线性的内存空间

2、要素

(1)写入数据当前写入位置指向哪里就从哪里开始写入如果后续剩余空闲不够了

考虑整体缓冲区空闲空间是否足够(因为都位置也会向后偏移前边有可能会有空闲空间)

不够扩容从当前写位置开始扩容足够大小

public:Buffer():_reader_idx(0),_writer_idx(0),_buffer(BUFFER_DEFAULT_SIZE){}//获取起始地址char*

Begin()

len){//末尾空间足够直接返回if(TailIdleSize()

len)

{return;}//末尾空闲空间不够则判断加上起始位置的空闲空间按大小是否足够,狗了就将数据移到起始位置if(len

TailIdleSize()

HeadIdleSize()){//将数据移动到起始位置uint64_t

rsz

ReadAbleSize();//把当前数据大小先保存起来std::copy(ReadPosition(),ReadPosition()

rsz,

rsz;}else{//总体空间不够则需要扩容不移动数据_buffer.resize(_writer_idx

len);}}//

{return;}//确保空间足够EnsureWriteSpace(len);const

char*

char*)data;std::copy(d,dlen,WritePosition());}//写入数据并移动位置void

WriteAndPush(void

len){Write(data,len);MoveReadOffset(len);}//

写入字符串并移动位置void

data){WriteString(data);MoveWriteOffset(data.size());}//写入缓冲区并移动位置void

data){WriteBuffer(data);MoveWriteOffset(data.ReadAbleSize());}//

写入字符串不移动数据void

data){Write((void*)data.c_str(),

data.size());}

Write((void*)data.ReadPosition(),

data.ReadAbleSize());}//读取数据void

Read(void*

len){Read(buf,len);MoveReadOffset(len);}//读取字符串并移动读取位置std::string

len);std::string

ReadString(len);MoveReadOffset(len);return

ReadString(uint64_t

str;str.resize(len);Read(str[0],len);return

str;}//获取换行符char*

(char*)res;}//获取一行数据std::string

GetLine(){char*

GetLine();MoveReadOffset(str.size());return

str;}//

_buffer;//使用vector进行内存空间管理uint64_t

_reader_idx;

\n;buffer.WriteStringAndPush(str);}while(buffer.ReadAbleSize()

0){std::string

buffer.ReadStringAndPop(buffer.ReadAbleSize());cout

tmp

buffer.WriteStringAndPush(str);//

Buffer

buffer1.WriteBufferAndPush(buffer);//

std::string

buffer1.ReadStringAndPop(buffer1.ReadAbleSize());//

cout

主要功能创建套接字、绑定地址信息、开始监听、向服务器发起连接、获取新连接、接收数据、发送数据

关闭套接字、创建一个服务端连接、创建一个客户端连接、设置套接字选项—开启地址端口重用、

设置套接字阻塞属性—设置为非阻塞

选项定义的层次。

或为特定协议的代码如IPv4IPv6TCP或为通用套接字代码SOL_SOCKET//地址重用

setsockopt(_sockfd,SOL_SOCKET,SO_REUSEADDR,

(void*)val,

setsockopt(_sockfd,SOL_SOCKET,SO_REUSEPORT,

(void*)val,

public:Socket():_sockfd(-1){}Socket(int

fd):_sockfd(fd){}~Socket(){Close();}int

Fd(){return

socket(AF_INET,SOCK_STREAM,IPPROTO_TCP);if(_sockfd

SOCKET

htons(port);addr.sin_addr.s_addr

inet_addr(ip.c_str());socklen_t

len

htons(port);addr.sin_addr.s_addr

inet_addr(ip.c_str());socklen_t

len

accept(_sockfd,nullptr,nullptr);if(newfd

ACCEPT

当前socket的接收缓冲区没有数据量在非阻塞的情况下才会有这个错误//EINTR

EAGAIN

0;//表示这次接收没有数据但是可以被原谅}logger-fatal(SOCKET

RECV

Recv(buf,len,MSG_DONTWAIT);//MSG_DONTWAIT

Send(const

1;//地址重用setsockopt(_sockfd,SOL_SOCKET,SO_REUSEADDR,

(void*)val,

1;//端口重用setsockopt(_sockfd,SOL_SOCKET,SO_REUSEPORT,

(void*)val,

lst_socket;lst_socket.CreateServer(8888);while(1){int

newfd

0){cli_sock.Close();continue;}cli_sock.Send(buf,ret);cli_sock.Close();}lst_socket.Close();

}tcp_client.cc

cli_sock;cli_sock.CreateClient(8888,

str

sqy;cli_sock.Send(str.c_str(),str.size());char

buf[1024]

fd):_fd(fd),_events(0),_revents(0),_loop(loop)

{}int

DisableWrite()//关闭写事件监控{_events

~EPOLLOUT;

EPOLLPRI)){if(_event_callback)_event_callback();if(_read_callback)

EPOLLOUT){if(_event_callback)_event_callback();//放到事件处理完毕后调用刷新活跃度if(_write_callback)

if(_revents

EPOLLERR){if(_event_callback)_event_callback();//不管任何事件都会调用的回调函数if(_error_callback)

if(_revents

EPOLLHUP){if(_event_callback)_event_callback();if(_close_callback)

_close_callback();}}

_events;}//获取想要监控的事件private:EventLoop*

_loop;int

_revents;//当前连接触发的事件EventCallback

_read_callback;

意义对epoll进行的封装让对描述符进行事件监控的操作更加简单

封装思想

3、使用hash表管理描述符与描述符对应事件管理Channel对象

逻辑流程

1、对描述符进行监控通过Channel才能知道描述符需要监控什么事件

2、当描述符就绪了通过描述符在hash表中找到对应的Channel得到了Channel才能知道什么事件如何处理

#define

epoll_create(1);//这个数字随便给if(_epfd

CREATE

FAILED);return;}}//添加或修改监控事件void

UpdateEvent(Channel*

false){//不存在则添加_channels.insert(std::make_pair(channel-Fd(),

channel));Update(channel,EPOLL_CTL_ADD);return;}Update(channel,EPOLL_CTL_MOD);}//移除监控void

RemoveEvent(Channel*

_channels.find(Channel-Fd());if(it

_channels.end()){_channels.erase(it);}Update(Channel,EPOLL_CTL_DEL);}//开始监控返回活跃连接void

active){//

EINTR){return;//这是被信号打断的直接返回}logger-fatal(EPOLL

WAIT

%s\n,strerror(errno));abort();//退出程序}for(int

nfds;

_channels.find(_evs[i].data.fd);assert(it

_channels.end());it-second-SetREvents(_evs[i].events);//设置实际就绪的事件active-push_back(it-second);}}

private://内层封装void

FAILED);abort();//退出程序}}//判断一个Channel是否已经添加了事件监控bool

HashChannel(Channel*

_channels.find(channel-Fd());if(it

false;}return

_evs[MAX_EPOLLEVENTS];std::unordered_mapint,Channel*

_channels;

_poller-RemoveEvent(this);}//移除监控

void

_poller-UpdateEvent(this);}Poller与Channel的联合调试代码

#include

std::endl;channel-Remove();//移除监控delete

channel;

HandleClose(channel);}channel-EnableWrite();//启动可写事件cout

buf

0){HandleClose(channel);//出错移除监控}channel-DisableWrite();//关闭写监控

}void

accept(fd,nullptr,nullptr);if(newfd

{return

(std::bind(HandleRead,channel));//为通信套接字设置可读事件的回调函数channel-SetWriteCallback(std::bind(HandleWrite,channel));//可写事件的回调函数channel-SetCloseCallback(std::bind(HandleClose,channel));//关闭事件的回调函数channel-SetErrorCallback(std::bind(HandleError,channel));//错误事件的回调函数channel-SetEventCallback(std::bind(HandleEvent,channel));//任意事件的回调函数channel-EnableRead();

int

lst_socket;lst_socket.CreateServer(8888);//为监听套接字创建一个Channel进行事件的管理以及事件的处理Channel

channel(poller,lst_socket.Fd());channel.SetReadCallback(std::bind(Acceptor,poller,

channel));//回调中获取新连接为新连接创建Channel并且添加监控channel.EnableRead();while(1){std::vectorChannel*actives;poller.Poll(actives);for(auto

actives){a-HandleEvent();}}lst_socket.Close();

}10、EventLoop模块

eventfd本质在内核里边管理的就是一个计数器创建eventfd就会在内核中创建一个计数器结构

例如假设每次给eventfd中写入一个1就表示通知了一次连续写了三次之后再去read读取出来的数字就是3读取之后计数清零

eventfd

eventfd也是通过read/write/close进行操作的

注意readwrite进行IO的时候数据只能是一个8字节的数据基本使用

#include

sizeof(res));printf(%ld\n,res);write(efd,

val,

sizeof(res));printf(%ld\n,res);close(efd);return

}Eventloop

监控了一个连接而这个连接一旦就绪就要进行事件处理但是如果这个描述符在多个线程中触发了事件进行处理

就会存在线程安全问题。

因此我们需要将一个连接的事件监控以及连接事件处理以及其他操作都放在同一个线程中进行。

解决方案给eventloop模块中添加一个任务队列,对连接的所有操作都进行一次封装将对

eventloop处理流程

2、有描述符就绪则对描述符进行事件处理如何保证处理回调函数中的操作都在线程中

3、所有的就绪事件处理完了这时候再去将任务队列中的所有任务一一执行

class

public:EventLoop():_thread_id(std::this_thread::get_id()),_event_fd(CreatEventFd()),_event_channel(new

Channel(this,_event_fd)),_poller(),_timer_wheel(this){//给eventfd添加可读事件回调函数读取eventfd事件通知次数_event_channel-SetReadCallback(std::bind(EventLoop::ReadEventFd,this));_event_channel-EnableRead();//启动可读事件监控}void

res

EINTR){return;}logger-fatal(READ

EVENTFD

EINTR){return;}logger-fatal(READ

EVENTFD

FAILED!!);abort();}}//判断将要执行的任务是否处于当前线程中如果是则执行不是则压入队列void

RunInLoop(const

cb)//将操作压入任务池{{std::unique_lockstd::mutex

_lock(_mutex);_tasks.push_back(cb);}//唤醒又肯因为没有事件就绪而导致的epoll阻塞//其实就是给eventfd写入一个数据eventfd就会触发可读事件WeakUpEventFd();}//用于判断当前线程是否是EventLoop对应的线程bool

IsInLoop(){return

std::this_thread::get_id();}void

AssertInLoop(){assert(_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

_timer_wheel.TimerRefresh(id);}void

TimerCancel(uint64_t

_timer_wheel.TimerCancel(id);}void

Start()//事件监控-》就绪事件处理-》执行任务{//1、事件监控std::vectorChannel*

actives;_poller.Poll(actives);//2、事件处理for(auto

channel

actives){channel-HandleEvent();}//3、执行任务//

RunAllTask();}//这个接口存在线程安全问题--这个接口不能被外界使用者调用只能在模块内在对应的eventloop线程内执行bool

HasTimer(uint64_t

RunAllTask()//执行所有任务池的任务{std::vectorFunctor

functor;{std::unique_lockstd::mutex

_lock(_mutex);//加锁保护交换操作交换操作不上线程安全的_tasks.swap(functor);}for(auto

functor){f();}}

_event_fd;//eventfd唤醒IO事件监控有可能导致的阻塞std::thread::id

_poller;std::unique_ptrChannel_event_channel;//在eventloop释放的时候他也要释放所以用智能指针std::vectorFunctor

_mutex;//实现任务池操作的线程安全TimerWheel

_timer_wheel;

timefd实现内核每隔一段事件给进程一次超时事件timefd可读

timewheel实现每次执行Runtimetask都可以执行一波到期的定时任务

timefd设置为每秒钟触发一次定时事件当事件被触发则运行一次timewheel的runtimetask执行一下所有的过期定时任务

using

cb):_id(id),_timeout(delay),_task_cb(cb),_canceled(false){}void

SetRelease(const

true;}~TimerTask(){if(_canceled

false)

_task_cb();//定时任务没有被取消才会执行_release();}

private:uint64_t

_task_cb;//定时器对象要执行的定时任务ReleaseFunc

_release;//用于删除TimerWheel中保存的定时器对象信息bool

_canceled;

loop):_capacity(60),_tick(0),_wheel(_capacity),_timefd(CreateTimerfd()),_timer_channel(new

Channel(_loop,

_timefd)),_loop(loop){_timer_channel-SetReadCallback(std::bind(TimerWheel::OnTime,

this));_timer_channel-EnableRead();//启动读事件监控}void

times;int

timerfd_create(CLOCK_MONOTONIC,

0);if(timerfd

第一次超时时间为1s后itime.it_interval.tv_sec

//第一次超时后每次超时的间隔timerfd_settime(timerfd,

itime,

timerfd;}//因为很多定时任务都涉及线程安全问题如果不在同一线程先加入任务队列void

TimerAdd(u_int64_t

TimerTask(id,delay,cb));pt-SetRelease(std::bind(TimerWheel::RemoveTimer,this,id));_timers[id]

WeakTask(pt);int

_capacity;_wheel[pos].push_back(pt);}//刷新/延迟定时任务void

id);void

id){//通过保存我的定时器对象的weak_ptr构造一个shared_ptr出来添加到轮子中auto

_timers.end()){return;//没有找到定时任务没法刷新没法延迟}PtrTask

it-second.lock();//lock获取weak_ptr管理的对象对应的shared_ptrint

delay

_capacity;_wheel[pos].push_back(pt);}//这个函数应该每秒钟执行一次相当于秒针向后走了一步void

RunTimerTask(){_tick

_capacity;_wheel[_tick].clear();}void

OnTime(){ReadTimefd();RunTimerTask();}//取消定时任务void

TimerCancel(uint64_t

_timers.end()){return;//没有找到定时任务没法刷新没法延迟}PtrTask

pt-Cancel();}

_timers.end()){_timers.erase(it);}}

private:using

//表盘最大数量---其实就是最大延迟时间std::vectorstd::vectorPtrTask

_wheel;std::unordered_mapuint64_t,

WeakTask

目的对连接进行全方位的管理对通信连接的所有操作都是通过这个模块提供的功能完成

功能设计

因为连接接收到数据之后该如何处理由用户决定因此必须有业务处理回调函数

一个连接建立成功后该如何处理由用户决定因此必须有连接建立成功的回调函数

一个连接关闭前该如何处理由用户决定因此必须有关闭连接回调函数。

任意事件的产生有没有某些处理由用户决定因此必须有任意事件的回调函数

1、发送数据

给用户提供的发送数据接口并不是真正的发送接口而是把数据放到发送缓冲区然后启动写事件监控

2、关闭连接

给用户提供的关闭连接接口应该在实际释放连接之前看看输入输出缓冲区是否有数据待处理

5、协议切换

一个连接接收数据后如何进行业务处理取决上下文以及数据的业务处理回调函数

Connection模块是对连接的管理模块对于连接的所有操作都是通过这个模块完成的

场景对连接进行操作的时候但是连接已经被释放导致内存访问错误最终程序崩溃

解决方案使用只能指针shared_ptr对Connection对象进行管理这样就能保证任意一个地方对Connection对象进行操作的时候保

存了一份shared_ptr因此就算其他地方进行释放操作也只是对shared_ptr的计数器-1而不会导致Connection的实际释放

typedef

{DISCONNECTED,//连接关闭状态CONNECTING,

//连接建立完成--各种设置已完成可以通信的状态DISCONNECTING//待关闭状态}ConnStatu;

class

std::enable_shared_from_thisConnection

{private:uint64_t

//定时器ID必须是唯一的这块为了简化操作使用conn_id作为定时器IDint

_sockfd;

连接是否启动非活跃销毁的判断标志默认为falseEventLoop

*_loop;

请求的接收处理上下文/*这四个回调函数是让服务器模块来设置的其实服务器模块的处理回调也是组件使用者设置的*//*换句话说这几个回调都是组件使用者使用的*/using

ConnectedCallback

PtrConnection);ConnectedCallback

_connected_callback;MessageCallback

_message_callback;ClosedCallback

_closed_callback;AnyEventCallback

_event_callback;/*组件内的连接关闭回调--组件内设置的因为服务器组件内会把所有的连接管理起来一旦某个连接要关闭*//*就应该从管理的地方移除掉自己的信息*/ClosedCallback

_server_closed_callback;private:/*五个channel的事件回调函数*///描述符可读事件触发后调用的函数接收socket数据放到接收缓冲区中然后调用_message_callbackvoid

HandleRead()

ShutdownInLoop();}//这里的等于0表示的是没有读取到数据而并不是连接断开了连接断开返回的是-1//将数据放入输入缓冲区,写入之后顺便将写偏移向后移动_in_buffer.WriteAndPush(buf,

ret);//2.

{//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(),

(ret

{_message_callback(shared_from_this(),

_in_buffer);}return

Release();//这时候就是实际的关闭释放操作了。

}_out_buffer.MoveReadOffset(ret);//千万不要忘了将读偏移向后移动if

没有数据待发送了关闭写事件监控//如果当前是连接待关闭状态则有数据发送完数据释放连接没有数据则直接释放if

(_statu

Release();}}return;}//描述符触发挂断事件void

HandleClose()

{/*一旦连接挂断了套接字就什么都干不了了因此有数据待处理就处理一下完毕关闭连接*/if

{_message_callback(shared_from_this(),

_in_buffer);}return

_event_callback(shared_from_this());

}}//连接获取之后所处的状态下要进行各种设置启动读监控,调用回调函数void

EstablishedInLoop()

CONNECTING);//当前的状态必须一定是上层的半连接状态_statu

CONNECTED;//当前函数执行完毕则连接进入已完成连接状态//

一旦启动读事件监控就有可能会立即触发读事件如果这时候启动了非活跃连接销毁_channel.EnableRead();if

_connected_callback(shared_from_this());}//这个接口才是实际的释放接口void

ReleaseInLoop()

移除连接的事件监控_channel.Remove();//3.

CancelInactiveReleaseInLoop();//5.

调用关闭回调函数避免先移除服务器管理的连接信息导致Connection被释放再去处理会出错因此先调用用户的回调函数if

(_closed_callback)

_closed_callback(shared_from_this());//移除服务器内部管理的连接信息if

_server_closed_callback(shared_from_this());}//这个接口并不是实际的发送接口而只是把数据放到了发送缓冲区启动了可写事件监控void

SendInLoop(Buffer

;_out_buffer.WriteBufferAndPush(buf);if

false)

{_channel.EnableWrite();}}//这个关闭操作并非实际的连接释放操作需要判断还有没有数据待处理待发送void

ShutdownInLoop()

_message_callback(shared_from_this(),

_in_buffer);}//要么就是写入数据的时候出错关闭要么就是没有待发送数据直接关闭if

{if

{Release();}}//启动非活跃连接超时释放规则void

EnableInactiveReleaseInLoop(int

sec)

_loop-TimerRefresh(_conn_id);}//3.

如果不存在定时销毁任务则新增_loop-TimerAdd(_conn_id,

sec,

event;}public:Connection(EventLoop

*loop,

_sockfd(sockfd),_enable_inactive_release(false),

_loop(loop),

_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,

logger-debug(RELEASE

}//连接建立就绪后进行channel回调设置启动读监控调用_connected_callbackvoid

Established()

{_loop-RunInLoop(std::bind(Connection::EstablishedInLoop,

this));}//发送数据将数据放到发送缓冲区启动写事件监控void

Send(const

{//外界传入的data可能是个临时的空间我们现在只是把发送操作压入了任务池有可能并没有被立即执行//因此有可能执行的时候data指向的空间有可能已经被释放了。

Buffer

len);_loop-RunInLoop(std::bind(Connection::SendInLoop,

this,

std::move(buf)));}//提供给组件使用者的关闭接口--并不实际关闭需要判断有没有数据待处理void

Shutdown()

{_loop-RunInLoop(std::bind(Connection::ShutdownInLoop,

this));}void

{_loop-QueueInLoop(std::bind(Connection::ReleaseInLoop,

this));}//启动非活跃销毁并定义多长时间无通信就是非活跃添加定时任务void

sec)

{_loop-RunInLoop(std::bind(Connection::EnableInactiveReleaseInLoop,

this,

{_loop-RunInLoop(std::bind(Connection::CancelInactiveReleaseInLoop,

this));}//切换协议---重置上下文以及阶段性回调处理函数

而是这个接口必须在EventLoop线程中立即执行//防备新的事件触发后处理的时候切换任务还没有被执行--会导致数据使用原协议处理了。

void

Upgrade(const

{_loop-AssertInLoop();_loop-RunInLoop(std::bind(Connection::UpgradeInLoop,

this,

Acceptor模块是对Socket模块Channel模块的⼀个整体封装实现了对⼀个监听套接字的整体的管理。

Acceptor模块内部包含有⼀个Socket对象实现监听套接字的操作

Acceptor模块内部包含有⼀个Channel对象实现监听套接字IO事件就绪的处理

具体处理流程如下

实现向Channel提供可读事件的IO事件处理回调函数函数的功能其实也就是获取新连接为新连接构建⼀个Connection对象出来。

当获取了一个新建连接的描述符之后需要为这个通信连接封装一个Connection对象设置各种不同回调

因为Acceptor模块本身并不知道一个连接产生了某个事件该如何处理因此获取一个通信连接后Connection的封装以及事件回调的设置都应该由服务器模块来进行

class

public://不能将启动读事件监控放到构造函数中必须在设置回调函数后再去启动//否则有可能造成启动监控后立即有事件回调函数还没设置新连接得不到处理Acceptor(EventLoop*

loop,

port):_socket(CreateServer(port)),_loop(loop),_channel(loop,_socket.Fd()){}void

AcceptCallback

Listen(){_channel.SetReadCallback(std::bind(Acceptor::HandleRead,this));_channel.EnableRead();}~Acceptor(){if(_socket.Fd()

private://监听套接字的读事件回调处理函数---获取新连接调用_accept_callback函数进行新连接处理void

HandleRead(){int

0){return;}if(_accept_callback)

CreateServer(int

_socket.CreateServer(port);(void)ret;assert(ret

true);return

EventLoop模块实例化的对象在构造的时候就会初始化_thread_id,而后边当运行一个操作的时候判断当前是否

运行在eventLoop模块对应的线程中就是将线程ID与EventLoop模块中的thread_id进行一个比较相同就表示

如果我们先创建了多个EventLoop对象然后创建了多个线程将各个线程的id重新给EventLoop进行设置

存在问题在构造EventLoop对象到设置新的thread_id期间将是不可控的

因此我们必须先创建线程然后在线程的入口函数中去实例化EventLoop对象

1、创建线程

public://创建线程设定线程入口函数LoopThread():_loop(nullptr),_thread(std::thread(LoopThread::ThreadEntry,this)){}//返回当前线程关联的EventLoop对象指针EventLoop*

GetLoop(){EventLoop*

nullptr;{std::unique_lockstd::mutex

lock(_mutex);//加锁_cond.wait(lock,[](){return

_loop

private://实例化EventLoop对象唤醒_cond上有可能阻塞的线程并且开始运行EventLoop模块的功能void

loop;{std::unique_lockstd::mutex

loop;_cond.notify_all();}loop.Start();}

private://用于实现_loop获取的同步关系避免线程创建了但是_loop还没有实例化之前去获取_loopstd::mutex

_mutex;//互斥锁std::condition_variable

_thread;//EventLoop对应的线程EventLoop*

_loop;//EventLoop指针变量这个对象需要在线程内实例化

设计一个线程池

LoopThreadPool模块对所有的LoopThread进行管理及分配

注意事项在服务器中主从Reactor模型是主线程只负责连接获取从属线程负责新连接的事件监控及处理

因此当前的线程池有可能从属线程会数量为0也就是实现单Reactor服务器一个线程及辅助获取连接也负责连接的处理

2、对所有的线程进行管理其实就是管理0个或多个LoopThread对象

当主线程获取了一个新连接需要将新连接挂到从属线程上进行事件监控及处理

假设有0个从属线程则直接分配给主线程的EventLoop进行处理

假设有多个从属线程则采用轮转思想进行线程的分配将对应线程的EventLoop获取到设置给对应的Connection

class

public:LoopThreadPool(EventLoop*

baseloop):_thread_count(0),_next_loop_idx(0),_baseloop(baseloop){}//设置线程数量void

SetThreadCount(int

Create()//创建所有的从属线程{if(_thread_count

0){_threads.resize(_thread_count);_loops.resize(_thread_count);for(int

_thread_count;

_threads[i]-GetLoop();}}}EventLoop*

return

_baseloop;//住EvnetLoop运行在主线程从属线程数量为0则所有操作都在baseloop中进行std::vectorLoopThread*

_threads;//保存所有的LoopThread对象std::vectorEventLoop*

_loops;//从属线程数量大于0则从_loops进行线程EventLoop分配

};16、TcpServer模块

对前边所有子模块的整合模块是提供给用户用于搭建一个高性能服务器的模块

1、Acceptor对象创建一个监听字

2、EventLoop对象baseloop对象实现对监听套接字的事件监控

PtrConnection

4、LoopThreadPool对象创建loop线程池对新建连接进行事件监控及处理

2、启动服务器

3、设置各种回调函数连接建立完成消息关闭任意用户设置给TcpServerTcpServer设置给获取的新连接

5、添加定时任务功能

1、在TcpServer中实例化一个Acceptor对象以及一个EventLoop对象baseloop

5、对连接对应的ConnEction设置功能回调连接完成回调消息回调关闭回调任意事件回调

7、将新连接对应的Connection挂到LoopThreadPool中的从属线程对应的EventLoop中进行事件监控

8、一旦Connection对应的连接就绪了可读事件则这时候执行读事件回调函数读取数据

class

port):_next_id(0),_port(port),_enable_inactive_release(false),_acceptor(_baseloop,_port),_pool(_baseloop){_acceptor.SetAcceptCallback(std::bind(TcpServer::NewConnection,this,std::placeholders::_1));_acceptor.Listen();//将监听套接字挂到baseloop上开始监听事件}void

SetThreadCount(int

count){_pool.SetThreadCount(count);}void

ConnectedCallbackcb)

//创建线程池的从属线程_baseloop.Start();}void

timeout){_timeout

timeout;_enable_inactive_release

true;}void

delay)//用于添加定时任务{_baseloop.RunInLoop(std::bind(TcpServer::RunAfterInLoop,this,task,delay));}

private:void

delay){_next_id;_baseloop.TimerAdd(_next_id,delay,task);}void

NewConnection(int

fd)//为新连接构造一个Connection进行管理{_next_id;PtrConnection

conn(new

Connection(_pool.NextLoop(),_next_id,

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

_conns.end()){_conns.erase(it);}}void

PtrConnection

conn)//从管理Connection的_conns中移除连接信息{_baseloop.RunInLoop(std::bind(TcpServer::RemoveConnectionInLoop,this,conn));}

private:uint64_t

//非活跃练级的统计事件---多长时间不通信是非活跃连接bool

_enable_inactive_release;//是否启动了非活跃连接超时销毁的判断标志Acceptor

_acceptor;

_baseloop;//主线程的eventloop对象负责监听事件的处理LoopThreadPool

_pool;//从属EventLoop线程池std::unordered_mapuint64_t,

PtrConnection

保存管理所有连接对应的shared_ptr对象ConnectedCallback

_connected_callback;MessageCallback

_message_callback;ClosedCallback

_closed_callback;AnyEventCallback

void

_loop-RemoveEvent(this);}//移除监控

void

{_loop-RunInLoop(std::bind(TimerWheel::TimerAddInLoop,

this,

TimerWheel::TimerRefresh(u_int64_t

id)

{_loop-RunInLoop(std::bind(TimerWheel::TimerRefreshInLoop,

this,

TimerWheel::TimerCancel(uint64_t

id)

{_loop-RunInLoop(std::bind(TimerWheel::TimerCancelInLoop,

this,

public:NetWork(){logger-debug(SIGPIPE

SIG_IGN);}

port):_server(port){_server.SetThreadCount(2);_server.EnableInactiveRelease(10);_server.SetClosedCallback(std::bind(EchoServer::OnClosed,this,std::placeholders::_1));_server.SetConnectedCallback(std::bind(EchoServer::OnConnected,this,std::placeholders::_1));_server.SetMessageCallback(std::bind(EchoServer::OnMessage,this,std::placeholders::_1,std::placeholders::_2));}void

private:void

CONNECTION:%p,conn.get());}void

OnConnected(const

CONNECTION:%p,conn.get());}void

OnMessage(const

*buf){conn-Send(buf-ReadPosition(),buf-ReadAbleSize());buf-MoveReadOffset(buf-ReadAbleSize());}

private:TcpServer

用于对高并发服务器模块进行协议支持基于提供的协议支持能够更方便的完成指定协议服务器的搭建。

Util模块

这个模块是一个根据模块主要提供HTTP协议模块所用到的一些工具函数比如url编码文件读写…等

HttpRequest模块

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

HttpResponse模块:

这个模块是HTTP响应数据模块用于业务处理后设置并保存HTTP响应数据的各项元素信息最终会被按照HTTP协议响应格式

HttpContext模块

这个模块是一个HTTP请求接收的上下文模块主要是为了防止再一次接收的数据中不是一个完整的HTTP请求则解析过程并未完成无法进行完整的请求处理需要在下次接收到新数据后根据上下文进行解析最终得到一个HttpRequest请求信息对象因此在请求数据的接收以及解析部分需要一个上下文来进行控制接收和处理节奏。

HttpServer模块

这个模块是最终给组件使用者提供的HTTP服务器模块了用于以简单的接口实现HTTP服务器的搭建。

HttpServer模块内容包含一个TcpServer对象TcpServer对象实现服务器的搭建

HttpServer模块内部包含有两个提供给TcpServer对象的接口连接建立成功设置上下文接口数据处理接口

HttpServer模块内部包含有一个hash-map表存储请求与处理函数的映射表组件使用者向HttpServer设置那些请求

应该使用那些函数进行处理等TcpServer收到对应的请求就会使用对应的函数进行处理。

1、Util模块代码实现

public://字符串分割函数,将src字符串按照sep字符进行分割得到的各个子串放到arry中最终返回子串的数量static

size_t

std::string::npos){//将剩余的部分当作一个子串arry-push_back(src.substr(idx));return

arry-size();}if(pos

sep.size();continue;//当字串为空没必要添加}arry-push_back(src.substr(idx,

pos

std::ios::binary);if(ifs.is_open()

FILE

ifs.tellg();ifs.seekg(0,ifs.beg);buf-resize(fsize);ifs.read((*buf)[0],fsize);if(ifs.good()

FILE

FAILED!!,filename.c_str());ifs.close();return

bool

ofs(filename,std::ios::binary);if(ofs.is_open()

false)

FAILED!!,filename.c_str());return

FILE

filename.c_str());ofs.close();return

true;}//URL编码,避免URL中子源路径与查询字符串中的特殊字符与HTTP请求中特殊字符产生歧义//编码格式将特殊字符的ascii值转换为两个16禁止字符前缀%//不编码字符RFC3986文档规定

~以及字母和数字属于绝对不编码字符//W3C标准中规定param中的空格必须被编码为//RFC2396规定URI中的保留字符需要转换为%HH格式static

std::string

convert_space_to_plus){std::string

res;for(auto

;continue;}//剩下的字符都是需要编码成为%HH格式char

tmp[4]

convert_space_to_plus){std::string

res;//遇到了%,则要将紧随其后的额2个字符转换为数字第一个数字左移4位然后加上第二个数字

%2b-2

statu){std::unordered_mapint,std::string

_statu_msg

Unknow;}//根据文件后缀名获取文件mimestatic

std::string

filename){std::unordered_mapstd::string,

std::string

application/vnd.amazon.ebook},{.bin,

application/octet-stream},{.bmp,

image/bmp},{.bz,

application/vnd.openxmlformats-officedocument.wordprocessingml.document},{.eot,

application/vnd.ms-fontobject},{.epub,

image/gif},{.htm,

image/vnd.microsoft.icon},{.ics,

application/java-archive},{.jpeg,

image/jpeg},{.jpg,

application/vnd.apple.installerxml},{.odp,

application/vnd.oasis.opendocument.presentation},{.ods,

application/vnd.oasis.opendocument.spreadsheet},{.odt,

application/vnd.oasis.opendocument.text},{.oga,

audio/ogg},{.ogv,

application/vnd.ms-powerpoint},{.pptx,

application/vnd.openxmlformats-officedocument.presentationml.presentation},{.rar,

application/x-rar-compressed},{.rtf,

image/svgxml},{.swf,

application/x-shockwave-flash},{.tar,

image/tiff},{.tiff,

application/vnd.ms-excel},{.xlsx,

application/vnd.openxmlformats-officedocument.spreadsheetml.sheet},{.xml,

application/vnd.mozilla.xulxml},{.zip,

video/3gpp},{.3g2,

application/x-7z-compressed}};size_t

pos

filename.find_last_of(.);if(pos

ext

application/octet-stream;}return

it-second;}return

application/octet-stream;}//判断一个文件是否是一个

/../login这个路径中的..会让路径的查找跑到相对根计算目录深度有多少层深度不能小于0std::vectorstd::string

subdir;Split(path,

http请求信息模块存储HTTP请求信息要素提供简单的功能性接口

请求信息要素

//资源路径的正则提取数据std::unordered_mapstd::string,

std::string

//头部字段std::unordered_mapstd::string,

std::string

//查询字符串public:HttpRequest():_version(HTTP/1.1)

{}void

{_method.clear();_path.clear();_version

HTTP/1.1;_body.clear();std::smatch

match;_matches.swap(match);_headers.clear();_params.clear();}//插入头部字段void

SetHeader(const

{_headers.insert(std::make_pair(key,

HasHeader(const

{_params.insert(std::make_pair(key,

HasParam(const

GetHeader(Content-Length);return

std::stol(clen);}//判断是否是短链接bool

Close()

没有Connection字段或者有Connection但是值是close则都是短链接否则就是长连接if

true

_redirect_url;std::unordered_mapstd::string,

std::string

_headers;public:HttpResponse():_redirect_flag(false),

_statu(200)

false;_body.clear();_redirect_url.clear();_headers.clear();}//插入头部字段void

SetHeader(const

{_headers.insert(std::make_pair(key,

HasHeader(const

没有Connection字段或者有Connection但是值是close则都是短链接否则就是长连接if

true

有可能出现接收的数据并不是一条完整的HTTP请求数据也就是请求的处理需要在多次收到数据后才能处理完成因此在每次处理的时候就需要将处理进度记录起来以便于下次从当前进度继续向下处理

接收状态

{RECV_HTTP_ERROR,RECV_HTTP_LINE,RECV_HTTP_HEAD,RECV_HTTP_BODY,RECV_HTTP_OVER

MAX_LINE

/bitejiuyeke/login?userxiaomingpass123123

HTTP/1.1//1

HTTP/1.1//请求方法的获取_request._method

matches[1];std::transform(_request._method.begin(),

::toupper);//资源路径的获取需要进行URL解码操作但是不需要转空格_request._path

false);//协议版本的获取_request._version

matches[4];//查询字符串的获取与处理std::vectorstd::string

query_string

符号进行分割得到各个字串Util::Split(query_string,

符号进行分割得到key

{//缓冲区中的数据不足一行则需要判断缓冲区的可读数据长度如果很长了都不足一行这是有问题的if

(buf-ReadAbleSize()

false;}//缓冲区中数据不足一行但是也不多就等等新数据的到来return

true;}if

false;}//首行处理完毕进入头部获取阶段_recv_statu

true;}bool

val\r\n....while(1){std::string

line

{//缓冲区中的数据不足一行则需要判断缓冲区的可读数据长度如果很长了都不足一行这是有问题的if

(buf-ReadAbleSize()

false;}//缓冲区中数据不足一行但是也不多就等等新数据的到来return

true;}if

false;}}//头部处理完毕进入正文获取阶段_recv_statu

true;}bool

line.pop_back();//末尾是换行则去掉换行字符if

(line.back()

line.pop_back();//末尾是回车则去掉回车字符size_t

pos

_request._body.size();//实际还需要接收的正文长度//3.

接收正文放到body中但是也要考虑当前缓冲区中的数据是否是全部的正文//

3.1

{_request._body.append(buf-ReadPosition(),

real_len);buf-MoveReadOffset(real_len);_recv_statu

true;}//

缓冲区中数据无法满足当前正文的需要数据不足取出数据然后等待新数据到来_request._body.append(buf-ReadPosition(),

buf-ReadAbleSize());buf-MoveReadOffset(buf-ReadAbleSize());return

true;}public:HttpContext():_resp_statu(200),

{}void

RECV_HTTP_LINE;_request.ReSet();}int

RespStatu()

因为处理完请求行后应该立即处理头部而不是退出等新数据switch(_recv_statu)

{case

表中记录了针对哪个请求应该使用哪个函数来进行业务处理的映射关系

当服务器收到了一个请求就在请求路由表中查找有没有对应请求的处理函数如果有则执行对应的处理函数即可

什么请求怎么处理由用户来设定服务器收到了请求只需要执行函数即可。

好处用户只需要实现业务处理函数然后将请求与处理函数的映射关系添加到服务器中而服务器只需要接收数据解析数据查找

路由表映射关系执行业务处理函数

3、对请求进行解析得到一个HttpRequest结构包含了所有的请求要素

将静态资源文件的数据读取处理填充到HttpResponse结构中

2.功能性请求—在请求路由映射表中查找处理函数找到了则执行函数

5、对静态资源请求/功能性请求进行处理完毕后得到了一个填充了响应信息的HttpResponse对象组织http格式响应进行发送

1、添加请求-处理函数映射信息(GET/POST/PUT/DELETE)

2、设置静态资源根目录

std::vectorstd::pairstd::regex,

Handler;Handlers

./wwwroot/image/a.pngstd::string

req_path

req._path;//为了避免直接修改请求的资源路径因此定义一个临时对象if

(req._path.back()

Util::ExtMime(req_path);rsp-SetHeader(Content-Type,

req,

{//在对应请求方法的路由表中查找是否含有对应资源请求的处理函数有则调用没有则发挥404//思想路由表存储的时键值对

正则表达式

处理函数//使用正则表达式对请求的资源路径进行正则匹配匹配成功就使用对应函数进行处理//

/numbers/(\d)

rsp);//传入请求信息和空的rsp执行处理函数}rsp-_statu

404;}void

既不是静态资源请求也没有设置对应的功能性请求处理函数就返回405if

(IsFileHandler(req)

{conn-SetContext(HttpContext());logger-debug(NEW

CONNECTION

conn-GetContext()-getHttpContext();//2.

通过上下文对缓冲区数据进行解析得到HttpRequest对象//

如果解析正常且请求已经获取完毕才开始去进行处理context-RecvHttpRequest(buffer);HttpRequest

req

rsp);//填充一个错误显示页面数据到rsp中WriteReponse(conn,

req,

rsp);//组织响应发送给客户端context-ReSet();buffer-MoveReadOffset(buffer-ReadAbleSize());//出错了就把缓冲区数据清空conn-Shutdown();//关闭连接return;}if

(context-RecvStatu()

{//当前请求还没有接收完整,则退出等新数据到来再重新继续处理return;}//3.

请求路由

对HttpResponse进行组织发送WriteReponse(conn,

req,

conn-Shutdown();//短链接则直接关闭}return;}public:HttpServer(int

port,

{_server.EnableInactiveRelease(timeout);_server.SetConnectedCallback(std::bind(HttpServer::OnConnected,

this,

std::placeholders::_1));_server.SetMessageCallback(std::bind(HttpServer::OnMessage,

this,

{assert(Util::IsDirectory(path)

true);_basedir

path;}/*设置/添加请求请求的正则表达与处理函数的映射关系*/void

Get(const

{_get_route.push_back(std::make_pair(std::regex(pattern),

handler));}void

{_post_route.push_back(std::make_pair(std::regex(pattern),

handler));}void

{_put_route.push_back(std::make_pair(std::regex(pattern),

handler));}void

{_delete_route.push_back(std::make_pair(std::regex(pattern),

handler));}void

{_server.SetThreadCount(count);}void

Listen()

{rsp-SetContent(RequestStr(req),

text/plain);

{rsp-SetContent(RequestStr(req),

text/plain);

req._path;Util::WriteFile(pathname,

req._body);

{rsp-SetContent(RequestStr(req),

text/plain);

server(7777);server.SetThreadCount(3);server.SetBaseDir(WWWROOT);//设置静态资源根目录告诉服务器有静态资源请求到来需要到哪里去找资源文件server.Get(/hello,

PutFile);server.Delete(/1234.txt,

DelFile);server.Listen();return

}六、服务器测试

cli_sock;cli_sock.CreateClient(7777,

req

{assert(cli_sock.Send(req.c_str(),

req.size())

buf);sleep(3);}cli_sock.Close();return

2、超时连接测试

/*超时连接测试1创建一个客户端给服务器发送一次数据后不动了查看服务器是否会正常的超时关闭连接*/#include

../server.hppint

cli_sock;cli_sock.CreateClient(7777,

req

{assert(cli_sock.Send(req.c_str(),

req.size())

buf);sleep(15);}cli_sock.Close();return

}这里测试的超时连接关闭我设置的超时时间是10秒服务端和客户端运行后客户端发送了一次数据后睡眠15秒

10秒后服务器超时释放而客户端到了15秒发送数据这时服务端已经释放连接发送数据失败所以超时连接测试成功。

3、Http服务器错误请求测试

/*给服务器发送一个数据告诉服务器要发送1024字节的数据但是实际发送的数据不足1024查看服务器处理结果*/

/*1.

如果数据只发送一次服务器将得不到完整请求就不会进行业务处理客户端也就得不到响应最终超时关闭连接2.

服务器会将后边的请求当作前边请求的正文进行处理而后便处理的时候有可能就会因为处理错误而关闭连接

*/#include

cli_sock;cli_sock.CreateClient(7777,

req

100\r\n\r\nbitejiuyeke;while(1)

{assert(cli_sock.Send(req.c_str(),

req.size())

-1);assert(cli_sock.Send(req.c_str(),

req.size())

-1);assert(cli_sock.Send(req.c_str(),

req.size())

buf);sleep(3);}cli_sock.Close();return

}注意这里测试的时候出了一个bug就是在测试服务器请求出错时导致缓冲区频繁扩容

因为在出错的时候想要调用关闭连接的操作没有将状态码重新设置如果状态码没有重新设置

4、HTTP服务器业务处理超时测试

cli_sock;cli_sock.CreateClient(8085,

req

{assert(cli_sock.Send(req.c_str(),

req.size())

buf);}cli_sock.Close();exit(0);}}while(1)

sleep(1);return

当服务器达到了一个性能瓶颈在一次业务处理中花费了太长的时间超过了服务器设置的非活跃超时时间

在一次业务处理中耗费太长时间导致其他的连接也被连累超时其他的连接有可能会被拖累超时释放假设现在

在处理1的时候花费了30s处理完超时了导致2345描述符因为长时间没有刷新活跃度1.

如果接下来的2345描述符都是通信连接描述符如果都就绪了则并不影响因为接下来就会进行处理并刷新活跃度2.

如果接下来的2号描述符是定时器事件描述符定时器触发超时执行定时任务就会将345描述符给释放掉这时候一旦345描述符对应的连接被释放接下来在处理345事件的时候就会导致程序崩溃内存访问错误因此这时候在本次事件处理中并不能直接对连接进行释放而应该将释放操作压入到任务池中等到事件处理完了执行任务池中的任务的时候再去释放5、HTTP服务器同时多条请求测试

../server.hppint

cli_sock;cli_sock.CreateClient(7777,

req

{assert(cli_sock.Send(req.c_str(),

req.size())

buf);sleep(3);}cli_sock.Close();return

}6、HTTP服务器大文件传输测试

/*大文件传输测试给服务器上传一个大文件服务器将文件保存下来观察处理结果*/

#include

cli_sock;cli_sock.CreateClient(7777,

req

body;Util::ReadFile(./hello.txt,

body);req

\r\n\r\n;assert(cli_sock.Send(req.c_str(),

req.size())

-1);assert(cli_sock.Send(body.c_str(),

body.size())

buf);sleep(3);cli_sock.Close();return

}7、服务器压力测试说明

原理创建大量的进程在进程中创建客户端连接服务器发送请求收到响应后关闭连接开始下一个连接的建立

测试环境

服务器是2核2g3m的云服务器服务器程序采⽤1主3从reactor模式

使用webbench以5000的并发量向服务器发送请求发送60s

虚拟机环境

服务器环境4核4G虚拟机服务器程序采⽤1主3从reactor模式

webbench客⼾端环境同⼀个虚拟机…