一个高性能RPC框架的连接管理

朴灿烈づ我的快乐病毒、 2022-08-11 03:24 147阅读 0赞

既然说连接，先对EpollServer的连接管理做个介绍吧。客户端与服务器一次conn，被封装成为Connection类在服务器进行管理。

服务器连接有三种类型，分别为：

**enum**  EnumConnectionType

*EM\_TCP*  = 0,

*EM\_UDP*  = 1,

*EM\_BUS*  = 2

\};

**Connection成员属性**

time\_t \_iLastRefreshTime ;// 最后刷新时间，初始化为初始化的当前时间

BindAdapter  \* \_pBindAdapter ;// 适配器

TC\_Socket  \_sock ; // TC\_Socket

uint32\_t  \_uid ; // 连接的唯一编号， 默认为0

**int**  \_lfd ; // 监听的socket， 默认为-1

**int**  \_timeout ;// 超时时间

string \_ip ; // ip

uint16\_t  \_port ; // 端口， 默认为0

string \_recvbuffer ; // 接收数据buffer

string \_sendbuffer ; // 发送数据buffer

size\_t \_sendPos ;// 默认为0

**int**  \_iHeaderLen ; // 需要过滤的头部字节数， 默认为0

**bool**  \_bClose ; // 发送完当前数据就关闭连接， 默认为false

**int**  \_iMaxTemQueueSize ;// 临时队列的最大长度， 默认为100

EnumConnectionType  \_enType ; // 连接类型， 默认为TCP

**bool**  \_bEmptyConn ; // 是否为空连接，默认为false，UDP一直为false

**char**  \* \_pRecvBuffer ; // 接收数据的临时buffer,加这个目的是对  udp 接收数据包大小进行设置,默认为null

size\_t \_nRecvBufferSize ;// 默认为64K，每次默认读取数据块大小

Connection提供了一系列的getter/setter方法修改上述属性，并且将那个EpollServer列为friend class,提供接收数据的功能操作接口。

**注意!一大波数据正在袭来**

当连接需要接收数据的时候（边缘触发，所以这里关键是一次性要将所有数据接收完）,我们需要一次性把数据接收完以后，把数据解析成一个一个完整的数据包，放入到指定的队列 o 中，首先Connection的buf是string的，本身不存在长度限制（可优化点），每次read的buf是8192个字节，不断的把客户端管道中的数据读取到Connection的Buf中，直到客户端EAGINE（没有数据了），或者接收到的数据小于每次读取的8192个字节。则表示已经没有数据了。所以这里如果长时间不处理，将会使得Connection的buf string堆积很多数据，如果没有数据可读了，则调用解析协议。

**int** **TC\_EpollServer::Connection::recv**( recv\_queue ::queue\_type &o)

o.clear ();

**char**  buffer\[8192\] =  "\\0" ;

**while** ( **true** )

**int**  iBytesReceived = 0;

**if** ( \_lfd  == -1)

**if** ( \_pRecvBuffer ) //如果该connection分配了自己的数据接受buffer，则用自己的，否则每次接收每次new出来一块buffer。

iBytesReceived =  \_sock . recvfrom(( **void** \*) \_pRecvBuffer , \_nRecvBufferSize ,  \_ip ,  \_port , 0);

**else**

iBytesReceived =  \_sock . recvfrom(( **void** \*)buffer, **sizeof** (buffer),  \_ip ,  \_port , 0);

**else**

iBytesReceived = :: read( \_sock .getfd(), ( **void** \*)buffer,  **sizeof** (buffer));

**if**  (iBytesReceived < 0)

**if** ( errno == EAGAIN )

//没有数据了

**break** ;

**else**

//客户端主动关闭

\_pBindAdapter \->getEpollServer()-> debug( "recv \[" \+  \_ip  \+  ":"  \+  TC\_Common :: tostr( \_port ) +  "\] close connection" );

**return**  \-1;

**else**  **if** ( iBytesReceived == 0)

//客户端主动关闭

\_pBindAdapter \->getEpollServer()-> debug( "recv \[" \+  \_ip  \+  ":"  \+  TC\_Common :: tostr( \_port ) +  "\] close connection" );

**return**  \-1;

//保存接收到数据

**if** ( \_lfd  == -1)

**if** ( \_pRecvBuffer )

\_recvbuffer . append( \_pRecvBuffer , iBytesReceived);

**else**

\_recvbuffer . append(buffer, iBytesReceived);

**else**

\_recvbuffer . append(buffer, iBytesReceived);

//UDP协议

**if** ( \_lfd  == -1)

**if** ( \_pBindAdapter \-> isIpAllow( \_ip ) ==  **true** )

parseProtocol(o);

**else**

//udpip无权限

\_pBindAdapter \->getEpollServer()-> debug( "accept \["  \+  \_ip \+  ":" \+  TC\_Common ::tostr( \_port ) +  "\] \["  \+  TC\_Common :: tostr( \_lfd ) +  "\] not allowed" );

\_recvbuffer  =  "" ;

**else**

//接收到数据不超过buffer,没有数据了(如果有数据,内核会再通知你)

**if** (( size\_t)iBytesReceived <  **sizeof** (buffer))

**break** ;

//字符串太长时substr性能会急剧下降

**if** ( \_recvbuffer . length() > 8192)

parseProtocol(o);

**if** ( \_lfd  != -1)

**return** parseProtocol(o);

**return**  o.size();

解析方法将把Connection从客户端管道中获取的所有的数据（ \_recvbuffer ）调用协议解析器进行协议解析，直到剩下的数据不构成一个完整的接受包为止。把这些请求包放入方法参数的队列中。

**int** **TC\_EpollServer::Connection::****parseProtocol**( recv\_queue ::queue\_type &o)

**try**

**while** ( **true** )

//需要过滤首包包头

**if** ( \_iHeaderLen > 0)

**if** ( \_recvbuffer . length() >= ( **unsigned** ) \_iHeaderLen )

string header = \_recvbuffer .substr(0, \_iHeaderLen );

\_pBindAdapter \->getHeaderFilterFunctor()(( **int** )( TC\_EpollServer :: *PACKET\_FULL* ), header);

\_recvbuffer = \_recvbuffer . substr( \_iHeaderLen );

\_iHeaderLen = 0;

**else**

\_pBindAdapter \->getHeaderFilterFunctor()(( **int** )( TC\_EpollServer :: *PACKET\_LESS* ), \_recvbuffer );

\_iHeaderLen \-= \_recvbuffer . length();

\_recvbuffer = "" ;

**break** ;

string ro;

**int** b = \_pBindAdapter \->getProtocol()( \_recvbuffer , ro); //此处调用了Adapter关联的编码解码器，taf默认是AppProtocol::parse()

**if** (b == TC\_EpollServer :: *PACKET\_LESS* )

//包不完全

**break** ;

**else** **if** (b == TC\_EpollServer :: *PACKET\_FULL* )

tagRecvData \* recv = **new** tagRecvData();

recv-> buffer  = ro;

recv-> ip  = \_ip ;

recv-> port  = \_port ;

recv-> recvTimeStamp  = TC\_TimeProvider ::*getInstance*()->getNow();

recv-> uid  = getId();

recv-> isOverload  = **false** ;

recv-> isClosed  = **false** ;

//收到完整的包才算

**this** \-> \_bEmptyConn = **false** ;

//收到完整包

o. push\_back(recv);

**if** (( **int** ) o. size() > \_iMaxTemQueueSize )

insertRecvQueue(o);

o. clear();

**if** ( \_recvbuffer . empty())

**break** ;

**else**

\_pBindAdapter \->getEpollServer()-> error( "recv \[" \+ \_ip \+ ":" \+ TC\_Common :: tostr( \_port ) + "\],packet error." );

**return** \-1;  //协议解析错误

**catch** (exception &ex)

\_pBindAdapter \->getEpollServer()-> error( "recv protocol error:" \+ string (ex.what()));

**return** \-1;

**catch** (...)

\_pBindAdapter \->getEpollServer()-> error( "recv protocol error" );

**return** \-1;

**return** o.size();

AppProtocol解析如下：

/\*\*

\* 解析协议

\* @param in, 目前的buffer

\* @param out, 一个完整的包

\* @return int, 0表示没有接收完全, 1表示收到一个完整包

\*/

**static** **int** **parse**(string &in, string &out)

**return** parseLen<10000000>(in,out);

**template** <taf:: Int32 **iMaxLength** >

**static** **int** **parseLen**(string &in, string &out)

**if** (in.length() < **sizeof** (taf:: Int32 )) //首先需要一个int32来标志包的长度，所以我们的接受buff的长度起码得>int32,否则标识为收包不完整

**return** TC\_EpollServer :: *PACKET\_LESS* ;

taf:: Int32 iHeaderLen; //包的长度头

memcpy(&iHeaderLen, in.c\_str(), **sizeof** (taf:: Int32 )); //获取包的长度

iHeaderLen = ntohl(iHeaderLen); //将网络顺序变为主机顺序，非常重要

**if** (iHeaderLen < taf:: Int32 ( **sizeof** (taf:: Int32 ))|| iHeaderLen > **iMaxLength** ) //如果当前的包的长度超过了最大长度或者小于一个int32的长度（包头本身）则认为协议错误

**return** TC\_EpollServer :: *PACKET\_ERR* ;

**if** (( **int** )in.length() < iHeaderLen) //缓存区里面字段长度不够包头长度

**return** TC\_EpollServer :: *PACKET\_LESS* ;

out = in.substr( **sizeof** (taf:: Int32 ), iHeaderLen - **sizeof** (taf:: Int32 ));

in = in. substr(iHeaderLen);//将刚才已经解析的完整的包部分从输入中去掉

**return** TC\_EpollServer :: *PACKET\_FULL* ;

利用adapter中注册的protocol进行编解码，如果包不完整，则跳出，返回状态码，如果收包完整，则构建新的 tagRecvData，设置完整的包buff，并且赋予Connection的ID（这个Connection锁接收到的数据包都是这个Connection的ID）以及将接收数据 tagRecvData的状态设置为非空，将这个包放入临时接收数据队列。如果当前这个接受数据队列超出了临时队列的最大长度（否则则在返回后再插入，避免大量的请求阻塞在这里而不能得到及时处理导致的各种超时，所以每到一定数目，则先放入 adapter去进行处理 ），则将该接收队列中的数据包放到adapter的rbuffer中(这个rbuffer是所有Connection一起共享的，所以必须是线程安全的)，如果当前adapter已经过载，将数据包依次设置为过载状态（很重要）放前面，这样有利于该Adapter接口紧急处理，避免雪崩和滚雪球效应，否则放后面。

**void** **TC\_EpollServer::Connection::****insertRecvQueue**( recv\_queue ::queue\_type &vRecvData)

**if** (!vRecvData.empty())

//服务队列已超载 数据放在队列头部

**if** ( \_pBindAdapter \->isOverload())

recv\_queue ::queue\_type::iterator it = vRecvData.begin();

recv\_queue ::queue\_type::iterator itEnd = vRecvData.end();

**while** (it != itEnd)

(\*it)-> isOverload =  **true** ;

\++it;

\_pBindAdapter \-> insertRecvQueue(vRecvData, **false** );

**else**

\_pBindAdapter \-> insertRecvQueue(vRecvData);

Connection需要集中管理起来

所有Connection都会被ConnectionList一个连接队列管理起来。统一进行超时检测等，空连接超时检测等， 连接队列 存储的数据为

pair< Connection \*, multimap<time\_t,  uint32\_t >::iterator >  list\_data;

ConnectionList成员属性如下：

TC\_EpollServer  \* \_pEpollServer ; // 服务

uint32\_t  \_total ; // 总计连接数

list< uint32\_t >  \_free ; // 空闲链表

size\_t \_free\_size ; // 空闲链元素个数

list\_data  \* \_vConn ; // 链接

multimap<time\_t,  uint32\_t >  \_tl ; // 超时链表

time\_t \_lastTimeoutTime ; // 上次检查超时时间

uint32\_t  \_iConnectionMagic ; // 连接ID的魔数

ConnectionList初始化，指定ConnectionList可以管理的最大连接数目，建立最大连接数目需要的空间，并标识为可用。

**void** **TC\_EpollServer::ConnectionList::init**( uint32\_t  size)

\_lastTimeoutTime  =  TC\_TimeProvider ::*getInstance*()->getNow(); //初始化检查超时时间

\_total  = size;

\_free\_size  = 0; //空闲连接数目初始化为0

//初始化链接链表

**if** ( \_vConn )  **delete** \[\]  \_vConn ; //如果空间存在，则删除先。

//分配total+1个空间(多分配一个空间, 第一个空间其实无效)

\_vConn  =  **new**  list\_data \[ \_total \+1\]; //分配最大连接数目需要的空间

\_iConnectionMagic  = (( uint32\_t ) \_lastTimeoutTime ) << 20; //连接ID的魔数

//free从1开始分配, 这个值为uid, 0保留为管道用, epollwait根据0判断是否是管道消息 ，表示空间的连接数目

**for** ( uint32\_t  i = 1; i <=  \_total ; i++)

\_vConn \[i\].first = NULL;

\_free .push\_back(i);

\++ \_free\_size ;

获取唯一ID，从当前free的空闲连接资源中取得一个可用连接的ID（也就是链表下标），并将之从free中去掉，并且与魔数或运算生成唯一ID。分配ID，相当于分配资源了。这里的魔数是ConnectionList建立的时候的时间<<20的值(剩下的12位留出来给该ConnectionList的下标，表明这个Connection存储在data\_list的哪个位置)，主要是用于校验当前的链接是否是该ConnectionList管理的资源，Connection的ID是否是这个ConnectionList分配出去的。

uint32\_t **TC\_EpollServer::ConnectionList::****getUniqId** ()

TC\_ThreadLock :: Lock  lock(\* **this** );

uint32\_t  uid =  \_free .front();

assert (uid > 0 && uid <=  \_total );

\_free .pop\_front();

\-- \_free\_size ;

**return**  \_iConnectionMagic  | uid;

新增连接，首先锁住整个ConnectionList，获取连接的ID(该ID之前由该LIST生成)，获得当时分配资源的下标，并将至加载到指定链表数据的指定位置。并且将该connection放入到指定链表位置，同时将该connection的下标放入到超时管理的map中。

**void** **TC\_EpollServer::****ConnectionList****::add**( Connection  \*cPtr, time\_t iTimeOutStamp)

TC\_ThreadLock :: Lock  lock(\* **this** );

uint32\_t  muid = cPtr->getId();

uint32\_t  magi = muid & (0xFFFFFFFF << 20);

uint32\_t  uid = muid & (0x7FFFFFFF >> 11);

assert (magi ==  \_iConnectionMagic && uid > 0 && uid <=  \_total  && ! \_vConn \[uid\].first);

\_vConn \[uid\] = make\_pair(cPtr,  \_tl . insert(make\_pair (iTimeOutStamp, uid)));

删除某个连接资源，根据指定的连接的ID,获得其在链表中的下标，从超时检测的map中移除他，从链表中找到该pair，删除connection资源，并且重新将他放到free可用中，同时更新freesize

**void** **TC\_EpollServer::ConnectionList::****del**( uint32\_t uid)

TC\_ThreadLock :: Lock  lock(\* **this** );

uint32\_t  magi = uid & (0xFFFFFFFF << 20);

uid = uid & (0x7FFFFFFF >> 11);

assert (magi ==  \_iConnectionMagic && uid > 0 && uid <=  \_total  &&  \_vConn \[uid\].first);

\_del(uid);

**void** **TC\_EpollServer::ConnectionList::\_del**( uint32\_t  uid)

assert (uid > 0 && uid <=  \_total &&  \_vConn \[uid\].first);

\_tl .erase( \_vConn \[uid\].second);

**delete**  \_vConn \[uid\].first;

\_vConn \[uid\].first = NULL;

\_free .push\_back(uid);

\++ \_free\_size ;

超时检测，没秒检查一次，每次把超时最后检测时间更新一下，超时检测的时候需要锁住整个链表，

**void** **TC\_EpollServer::ConnectionList::checkTimeout** (time\_t iCurTime)

//至少1s才能检查一次

**if** (iCurTime -  \_lastTimeoutTime  < 1)

**return** ;

\_lastTimeoutTime  = iCurTime;

TC\_ThreadLock :: Lock  lock(\* **this** );

multimap<time\_t,  uint32\_t >::iterator it =  \_tl .begin();

**while** (it !=  \_tl .end())

//已经检查到当前时间点了, 后续不用在检查了

**if** (it->first > iCurTime)

**break** ;

uint32\_t  uid = it->second;

\++it;

//udp的监听端口, 不做处理

**if** ( \_vConn \[uid\]. first\->getListenfd () == -1 ||  \_vConn \[uid\].first\-> getType() ==  Connection :: *EM\_BUS* )

**continue** ;

//超时关闭

\_pEpollServer \->delConnection( \_vConn \[uid\].first,  **false** );

//从链表中删除

\_del(uid);

**if** ( \_pEpollServer \->IsEmptyConnCheck())

it =  \_tl . begin();

**while** (it !=  \_tl . end())

uint32\_t  uid = it-> second;

//遍历所有的空连接

**if** ( \_vConn \[uid\]. first\->IsEmptyConn ())

//获取空连接的超时时间点

time\_t iEmptyTimeout = (it->first \-  \_vConn \[uid\].first\-> getTimeout()) + ( \_pEpollServer \->getEmptyConnTimeout()/1000);

//已经检查到当前时间点了, 后续不用在检查了

**if** (iEmptyTimeout > iCurTime)

**break** ;

//udp的监听端口, 不做处理

**if** ( \_vConn \[uid\]. first\->getListenfd () == -1 ||  \_vConn \[uid\].first\-> getType() ==  Connection :: *EM\_BUS* )

\++it;

**continue** ;

//超时关闭

\_pEpollServer \-> delConnection( \_vConn \[uid\]. first,  **false** );

//从链表中删除

\_del(uid);

\++it;

一个高性能RPC框架的连接管理

发表评论取消回复

还没有评论，来说两句吧...

相关阅读

相关 Dubbo 开源的一款高性能Java RPC框架

相关 Netty高性能的RPC框架

相关 Dubbo一个高性能优秀的RPC框架

相关高性能微服务 Dubbo RPC Redis 分布式框架

相关【Dubbo】高性能 RPC 通信框架

相关 srpc 高性能通用 RPC 框架

相关一个高性能RPC框架的连接管理

相关一个高性能RPC框架原理剖析

相关自己实现一个RPC框架

相关设计一个分布式RPC框架