TarsRPC源码解读篇：使用C++重写Tars-RPC主逻辑框架

前言

Tars(https://github.com/TarsCloud/Tars) 是腾讯开源的一套微服务框架。其基础是Tars RPC。对于有一些基础的同学来说，直接看RPC源码无疑是了解Tars的最佳途径。

代码注释和说明无疑是解读源码的一种好方式，然而在代码量较大的时候，部分小伙伴短时间可能较难get到关键点。所以在这里我尝试用重写RPC主框架的方式来解读Tars RPC部分的源码。初始的工程只有寥寥几个文件，很容易理解，并且自己动手调试超级方便。

我们从最基本的epoll模型开始，一步步向官方Tars RPC官方Tars RPC靠拢，并且尽量记录下关键的结点。

使用的系统环境：

• 操作系统版本：Ubuntu 18.04.1 LTS (GNU/Linux 4.15.0-42-generic x86_64)
• GCC版本：gcc version 7.3.0 (Ubuntu 7.3.0-27ubuntu1~18.04)

怎样看这份笔记？

• 代码提交记录可以很好的追踪代码变化。并且下面笔记中1.1、 1.2的内容与代码提交记录中的1.1、1.2是完全匹配的，可以对照理解

这只是实验性质的代码，因为更多关注主体逻辑，很多细节可能忽略掉了 代码编译只需要运行./run.sh，大多数人都可以轻松编译调试

1 重写RPC server

1.1 建立简单的epoll模型，完成与client端的收发测试

获取源码

编译源码：直接执行./run.sh，会生成tar-demo与client

执行代码：（1）执行./tar-demo （2）另外一个窗口执行./client， 输入测试数据 （3）查看两个窗口的结果

运行tar-demo：

tar-demo1.jpg

运行client：

tar-client1.jpg

• tc_epoller.h 里的TC_Epoller是对epoll各个原生函数的封装
• tc_epoll_server.h里的NetThread负责TC_Epoller的调度。

epoll.jpg

与平常使用习惯不同的地方，Tars里用epoll_event ev中的ev.data.u64作为ET_LISTEN、ET_CLOSE、ET_NOTIFY、ET_NET类型标识

1.2 重构代码，封装原生socket

获取源码

做完1.1，就已经嗅到了代码中的“坏味道”，NetThread中socket部分的代码是可以复用的，所以可以单独拿出来。

tc_socket.h中的TC_Socket是对原生socket的封装，注意同步修改tc_epoll_server中的相应代码

1.3 增加锁，条件变量和队列的封装，为多线程做准备

获取源码

tc_thread_mutex.h 是对pthread_mutex_t的封装

tc_thread_cond.h 是对pthread_cond_t的封装

tc_monitor.h 与 tc_lock.h 提供了对tc_thread_cond.h与tc_thread_mutex.h的封装

简单梳理下使用逻辑：

一般实际使用的是TC_ThreadLock::Lock，即：

typedef TC_LockT<TC_Monitor<T, P> > Lock;

这里面有两个元素，一是TC_LockT模板类（在tc_lock.h中定义），一是TC_Monitor模板类

TC_Monitor的定义如下：

typedef TC_Monitor<TC_ThreadMutex, TC_ThreadCond> TC_ThreadLock;

举例说明使用逻辑：

TC_LockT的构造函数调用了TC_Monitor的lock()方法：

TC_LockT(const T& mutex) : _mutex(mutex)  

{  
    //_mutex为TC_Monitor  
    _mutex.lock();
    _acquired = true;  
}

TC_Monitor的lock()方法调用了TC_ThreadMutex的lock()方法：

TC_LockT(const T& mutex) : _mutex(mutex)
{
    //_mutex为TC_ThreadMutex
    _mutex.lock();
    _acquired = true;
}

TC_ThreadMutex的lock()方法调用了原生的pthread_mutex_lock

void TC_ThreadMutex::lock() const
{
    int rc = pthread_mutex_lock(&_mutex);
    if(rc != 0)
    {
        if(rc == EDEADLK)
        {
            cout<<"[TC_ThreadMutex::lock] pthread_mutex_lock dead lock error"<<rc<<endl;
        }
        else
        {
            cout<<"[TC_ThreadMutex::lock] pthread_mutex_lock error"<<rc<<endl;
        }
    }
}

1.4 建立多线程模型，主线程NetThread负责请求接受和结果发送，Handle线程负责业务逻辑处理

获取源码

1.3的模型只有一个线程，请求的接收、加工处理、结果的发送都在同一个线程中完成，并不能充分利用io和cpu。所以我们做第一步改进，将请求的接收与结果的发送放在NetThread线程中，将请求的逻辑处理放在Handle线程中，两个线程沟通的桥梁是两个队列。如下图所示：

1.4 多编程模型线程处理逻辑.png

• 主线程NetThread将客户端请求放入接收队列_rbuffer，Handle线程从_rbuffer中读取请求进行处理
• Handle线程将处理完的结果放入发送队列_sbuffer，主线程NetThread从_sbuffer中读取结果返回给客户端

在这一节中，我们增加了TC_Thread类作为线程的封装，以便于扩展多线程；将NetThread和Handle两个类放到TC_EpollServer类中，以便于两个线程间的交互。

1.5 继续重构，增加BindAdapter保管ip和port，同时引入标准库智能指针

获取源码

NetThread类在可预见的未来方法会越来越多，所以我们慢慢把一些职责从NetThread中分离出来，作为NetThread的依赖存在。

新增的BindAdapter类接管了原NetThread中的int bind(string& ip, int& port);方法，把对IP和PORT的处理放到了TC_Endpoint中。

insertRecvQueue和waitForRecvQueue出现在了BindAdapter中，但这节我们还没有使用到，后面会用他们替换NetThread中的insertRecvQueue和waitForRecvQueue

1.6 增加Handle数目，完善1接收，n处理模型

获取源码

• 执行：./tar-demo
• 打开另一个窗口执行：./client 连续快速输入多个test(回车)
• 可以在tar-demo窗口观察到类似Handle thread id is 139941899609856这样的信息，并且Handle thread id会有变化

1.4节中我们解耦了线程的收发与业务逻辑处理，一个NetThread线程负责收发，一个Handle线程负责业务逻辑处理。现在又有新需求了，我们认为业务处理逻辑比较复杂，希望有多个Handle线程处理来自NetThread的请求。因为有之前的基础，仅仅需要修改main.cpp中的几行代码就能满足上面的需求。模型示意图如下：

1.6-多编程模型线程处理逻辑.png

1.7 增加Connection，第一步改造，接管accept文件操作符

获取源码

在NetThread类中增加Connection类，接管NetThread类中对accept文件操作符的管理。NetThread中所有跟accept文件操作符有关的函数（例如TC_EpollServer::NetThread::accept(int fd)）都需要进行修改

1.8 利用Connection，第二步改造，接管NetThread的发送与接收职责

获取源码

我们用Connection和BindAdapter中的insertRecvQueue与waitForRecvQueue替换掉了NetThread中的对应方法

Connection的recv与send方法接管了NetThread中的原生read和send方法

梳理下主要逻辑：

1. TC_EpollServer::NetThread::accept(int fd)方法中，Connection保存了accept文件操作符
2. TC_EpollServer::NetThread::addTcpConnection(TC_EpollServer::NetThread::Connection *cPtr)方法中，建立起唯一编码（uid）与Connection之间的对应关系（_uid_connectionuid = cPtr）
3. uid通过_epoller.add传递到TC_EpollServer::NetThread::processNet(const epoll_event &ev)
4. 在TC_EpollServer::NetThread::processNet(const epoll_event &ev)中uid住在Connection中，存入vRecvData（最后放入了_rbuffer队列）
5. TC_EpollServer::Handle::handleImp()从_rbuffer拿到recv的uid信息，通过sendResponse放回到NetThread中的_sbuffer中
6. TC_EpollServer::NetThread::processPipe()从_sbuffer拿取uid信息，通过uid获取Connection信息，通过Connection返回结果给客户端

1.9 NetThread转为线程，N（NetThread）收发，N（Handle）处理模型建立

获取源码

前面小节里，建立了1收发（NetThread），多（N）处理（Handle）的模型，这里再进一步，收发也改为多线程，Connection按照一定规则（getNetThreadOfFd）分到不同NetThread上完成接收，同时send也按照一定规则（getNetThreadOfFd）分配到不同NetThread上完成发送。这样就由原来的1：N变成了N：N模型，如下图所示：

19-N（NetThread）收发，N（Handle）处理模型建立.png

上图箭头编号表示处理顺序：

1. NetThread线程启动了4个，NetThread0作为唯一的监听线程存在， 监听request请求
2. 当有请求进来时候，NetThread0线程将accept操作符封装为Connection
3. NetThread0线程按照（getNetThreadOfFd）方法将Connection分配给NetThread0、NetThread1、NetThread2、NetThread3
4. 不同线程读取完请求后放入到同一个接收队列_rbuffer
5. 多个Handle线程在监听_rbuffer，有请求进来后会有一个Handle拿到请求并处理
6. Handle处理完后同样按照（getNetThreadOfFd）方法将结果分配给NetThread0、NetThread1、NetThread2、NetThread3

1.10 将Handle中的业务处理逻辑具体化为HelloImp

获取源码

当前模型的业务处理在Handle类中，处理逻辑也是极为简单，将请求字符串复制到结果字符串，实现的是类似"echo"的功能。

本着“对修改关闭，对扩展开放”的原则，我们应该考虑将具体的处理逻辑抽象出来，即Handle中处理代码应该是不变的，具体逻辑代码单独实现，并可以添加到Handle中进行处理。

HelloImp插件机制.png

• ServantHelperManager作为单例存在，可全局使用
• 外界编写新的业务逻辑，只需要像HelloImp一样继承Servant，实现doRequest
• 注册新的业务逻辑，只需要ServantHelperManager的setAdapterServant和addServant两步
• 实例化业务逻辑是在ServantHelperManager的create中完成的，是个通用的模板，故而放在了ServantHandle的initialize函数中

2 重写RPC client

从这一节开始，跟随我们多年的小兄弟client要退休了，取而代之的是同步调用客户端：tar-client和异步调用客户端：tar-client-async，rpc的client端性能至关重要，直接影响到外界系统的调用表现。

2.1 client模型建立，与tar-demo建立连接

获取源码

执行./run.sh 会生成tar-client

tar-client运行流程.png

• 程序入口在tar_client.cpp
• 建立ReqMessage，放入ReqInfoQueue中，并通知CommunicatorEpoll
• CommunicatorEpoll的run()得到通知后，根据pFDInfo->iType的类型进行处理
• 本节功能只更新到建立连接，还不能发送接收消息

关于执行./tar-client出现“doConnect errno is 115”提示的解释： 115：当链接设置为非阻塞时，目标没有及时应答，返回此错误，socket可以继续使用

2.2 完成client的同步调用

获取源码

今天终于完成了tar-client的同步调用，过程是比较痛苦的，结果是快乐的~~~

函数的调用流程梳理如下：

tar-client同步调用流程.png

黑色箭头为client发送请求的流程，红色箭头是client接收来自tar-demo结果的流程

需要注意的地方：

• ObjectProxy::invoke(ReqMessage * msg)函数中对msg->sReqData进行了序列化操作，把msg->iRequestId也作为请求的一部分发送给了服务端，服务端结果的返回也会带上这个iRequestId
• 在ObjectProxy::finishInvoke(const string& rsp)方法中，_timeoutQueue承担了一个请求和返回的上下文保存功能，通过返回结果中的iRequestId找到请求的msg，把返回结果放入msg->response中

2.3 完成client的异步调用，client为tar-client-async

获取代码

执行./run.sh 会生成tar-client-async

异步调用的实质就是返回结果在另外线程处理，与同步调用相比，异步调用就是将返回结果放到了一个单独队列，AsyncProcThread线程从这个队列中拿取结果调用回调函数，如下图所示：

tar-client异步调用流程.png

• 虚线框是相比同步调用增加的异步流程
• 蓝色的虚线框是为了异步处理结果新启动的线程，等待在队列上，一旦有新的结果进入队列，就取出调用回调函数进行处理

3 RPC Server加入协程

获取代码

前面两章已经初步完成了RPC Server与client的异步模型，本章开始着手将协程Coroutine加入RPC Server

3.1 为什么要引入协程

先复习一下Handle线程的处理逻辑，如下图：

ServantHandle异步模式.png

TC_EpollServer::BindAdapter::waitForRecvQueue与void ServantHandle::handle这两个方法是顺序处理的，意味着在while循环里handle这个方法执行完了才能继续接受下一个请求，即waitForRecvQueue

引入协程，可以让handle（协程里handle方法由handleRecvData取代，但为了跟上面解说一致，还是借用handle）和waitForRecvQueue看起来是并行的，waitForRecvQueue不必等待handle处理完成就能从队列中拿请求。理论上单位时间内Handle线程就可以从队列中拿取更多的需要处理的请求，如下图：

ServantHandle协程示意图.png

看到这里，很多人应该还有疑惑，协程本质还是一个线程在在工作，那到底什么时候处理handle方法，又什么时候处理waitForRecvQueue呢，这就需要一个协程的调度器，Tars代码里叫做CoroutineScheduler。

3.2 协程调度器CoroutineScheduler

ServantHandle协程调度器.png

• 了解协程流程的关键的是理解主协程和从协程的切换机制
• 我们把ServantHandle::run中的流程称为主协程，由主协程调用createCoroutine方法得到的为从协程。ServantHandle中有两个从协程，1号从协程为ServantHandle::handleRequest函数处理流程；2号从协程为ServantHandle::handleRecvData函数处理流程
• 整个协程调度的核心是主协程中的_coroSched->tars_run();
• 紫色箭头是主协程切换到1号、2号从协程，红色箭头是1号、2号从协程切换回主协程
• 主协程通过createCoroutine方法把一号从协程放入_avail链表中，接着进入while循环，tars_run里的_avail里这时只有1号从协程，因此从主协程切换到1号从协程
• 1号从协程里又通过createCoroutine把2号从协程放入_avail链表中，最后通过_coroSched->yield方法切换回主协程
• 主协程里_avail里有了2号从协程，所以从主协程切换到2号从协程运行，2号从协程运行完成后，回归到_free链表中，然后切换回主协程
• 主协程的_avail这时依然只有1号从协程，从主协程切换到1号从协程，然后回到上面第2步

4 包装同步RPC Client-向官方例子靠拢

获取源码

第3章和本章在代码提交记录上没有分类似3.1这样的分节，是一整章的提交记录，请知悉

前面章节的RPC Client外表看起来还不太像example/QuickStartDemo/HelloServer/Client里的使用方式，这一节的目标是完成向example中client使用方式的靠拢。

4.1 主要改动点说明

• ObjectProxy类之前已经完成了对ip和port的封装
• 增加ServantProxy类对ObjectProxy的包装，即外界一般都通过ServantProxy来使用ObjectProxy
• 增加Communicator类对ServantProxy和CommunicatorEpoll的封装
• 增加HelloProxy类（继承于ServantProxy），完成对具体调用函数testHello的封装（包括了对ReqMessage初始化的封装）
• tar_client_async_improve.cpp是本章的客户端

核心理解点是Communicator类通过stringToProxy方法，将实例化好的ServantProxy指针（父类指针）赋值给了HelloProxy指针（子类指针），HelloProxy指针负责请求的格式化和调度。这句话有点不好理解，事实上这部分代码对于初次阅读者可能都不太友好，不过没关系，我们可以从下面的模型示意图中理解上面这样玩法的用意。

4.2 逻辑流程说明

进一步完善RPC-Client(同步)逻辑图.png

1. 第4步创建了多个CommunicatorEpoll线程
2. 第6步创建了名为ppObjectProxy的ObjectProxy数组，注意ObjectProxy数组里的每一个ObjectProxy都是由具体的CommunicatorEpoll线程生成的
3. 第7步中ServantProxy将ppObjectProxy数组指针赋值给了自己的私有成员变量_objectProxy，完成了对ObjectProxy的包装
4. 第8步是转折点，HelloProxy是ServantProxy的子类，这一步中将第7步中的ServantProxy sp赋值给了HelloProxy prx，即父类指针赋值给了子类指针，这时候HelloProxy prx里就继承了ServantProxy sp中的所有内容，包括ppObjectProxy数组（ppObjectProxy数组在ServantProxy类中名字变为了_objectProxy）
5. 第12步ServantProxy::selectNetThreadInfo函数中采取类似轮询方式（pSptd->_netSeq++）获取_objectProxy数组中的具体ObjectProxy。每个ObjectProxy都会指向一个具体的CommunicatorEpoll线程，所以ObjectProxy就承担了通知其对应的CommunicatorEpoll线程接收请求的职责

5 包装异步RPC Client-向官方例子靠拢

获取源码

使用效果与examples/QuickStartDemo/HelloServer/AsyncClient/main.cpp看齐

5.1 修改点说明

• tar_client_async_improve.cpp是新增加的异步client类
• HelloPrxCallback函数放到了Hello.h中实现
• ServantProxy中增加tars_invoke_async方法
• 修改tar-demo部分代码，对应client的函数调用。主要涉及HelloImp.h Hello.h Servant.h及其对应的cpp

5.2 调用逻辑说明

tar_client_async_improve.cpp中异步调用的逻辑很简单：（1）编写回调函数HelloCallBack （2）调用异步接口async_testHello，将写好的回调函数出入即可。

这里想着重说下客户端调用了testHello函数后，是怎么调用到tar-demo中的testHello处理逻辑中的

• 以tar-client-async-improve为例，调用async_testHello时，实际发往tar-demo的请求是1:testHello:hello world。1来自于ObjectProxy::invoke中的msg->iRequestId，是一个请求序列号；testHello是调用方法名称；hello world为要请求的实际内容。
• 服务端接收到这个请求后，会在Hello的onDispatch中解析这个请求，得到方法名称是testHello，因此把这个请求转到HelloImp::testHello方法中进行处理

6 包装RPC Server-向官方例子靠拢

获取源码

本章的目的是让main.cpp向examples/QuickStartDemo/HelloServer/HelloServer.cpp看齐，通过增加Application类和对相应细节调整，完成对之前代码的封装

6.1 改动点说明

1. 增加Application类，封装对TC_EpollServer的操作
2. 增加HelloServer类，继承Application类，是对外使用的类
3. 在TC_EpollServer里增加HandleGroup，封装了BindAdapter和Handle。可绑定多个BindAdapter进行监听
4. 在run.sh里将tar-demo升级为tar-demo-improve, 原有的main.cpp不再使用

6.2 框架的变与不变

其实到本章为止，我们自己代码长的样子与官方example里已经有八九分相像了。服务端有了：

• HelloServer.h HelloServer.cpp
• HelloImp.cpp HelloImp.h
• Hello.h

可以对照QuickStartDemo/HelloServer下的对应文件。

客户端有了：

• tar_client_improve.cpp

对照QuickStartDemo/HelloServer/Client的main.cpp

• tar_client_async_improve.cpp

对照QuickStartDemo/HelloServer/AsyncClient的main.cpp。

回头重新审视下，我们发现如果仅仅是实现业务逻辑，需要修改的就是上面列出的文件。其余文件都是框架本身需要的。那么框架是怎么做到变与不变的分离呢？在想这个问题时候，可以重温下1.10节的内容。我们这里再重新梳理一下。

6-变与不变.png

• 在可变的部分，继承于Application的HelloServer通过initialize方法提供了一个可供任意业务逻辑（例如HelloImp）插入的接口，这个接口外表是addServant<HelloImp>，实际上是将HelloImp放到了单例类ServantHelperManager的成员变量中
• 在不变的框架逻辑中，ServantHandle线程通过initialize方法从ServantHelperManager获得了HelloImp并进行了实例化（这里用到了多态，使用的是Servant的指针），然后通过handle方法调用了Servant::onDispatch
• Servant::onDispatch的具体实现是在Hello中
• 总结下，Tars实现业务逻辑与框架本身的可插拔设计，是借用了ServantHelperManager这个全局的单例，另外就是利用多态实现运行时调用具体方法

7 Framework篇：NodeServer

获取源码

执行./run.sh，编译出两个server。HelloServer（原tar-demo-improve）和NodeServer

上面代码实现了最基本的heartbeat功能，即HelloServer定时向NodeServer发送心跳消息

实现发送心跳消息原理很简单：HelloServer中有一个client定时发异步请求到NodeServer，因此代码修改步骤如下：

• 定义NodeF.h，实现ServerFProxy和ServerF，前者是HelloServer的client需要用到，后者是NodeServer的ServerImp需要用到
• 实现Application::initializeClient()，完成_communicator初始化，增加TarsNodeFHelper，完成具体调用功能
• 增加AdminServant，其Adapter名称为AdminAdapter，这是因为TarsNodeFHelper中adapter名字必须为“AdminAdapter”才会去发起远程过程调用
• 增加ServerImp，完成NodeServer建造

先启动NodeServer(./NodeServer)，再另外开一个窗口启动HelloServer(./HelloServer)，过一会儿就会在NodeServer看到来自HelloServer的心跳请求。

7.1 通过NodeServer启动HelloServer

获取源码

改动点：

• 定义新接口Node.h， NodeImp通过继承其中的Node类，实现了startServer方法
• 新增NodeServer-start.sh和NodeServer-execute.sh，startServer方法通过执行NodeServer-start.sh脚本启动HelloServer
• 改动Application类，使其可以包含两个Imp（ServerImp和NodeImp），这块比较挫，后面再改进
• 修改ServantHandle::handleRequest方法，修复小bug，协程模式在客户端主动关闭时候，不要再继续调用handleRecvData方法
• 增加tar-client-NodeServer客户端，用来发送命令给NodeServer

请确保下载下来的NodeServer-start.sh和NodeServer-execute.sh有755的权限（chmod 755 * .sh）

如何利用NodeServer启动HelloServer：

• 执行./run.sh，生成HelloServer、NodeServer、tar-client-NodeServer
• 执行./NodeServer
• 另外开一个窗口，执行./tar-client-NodeServer
• 执行ps -ef | grep HelloServer, 能看到HelloServer的进程已经存在

总结下现在HelloServer与NodeServer的关系：

NodeServer-HelloServer.png

• 红色箭头是第一个功能，HelloServer定时向NodeServer发送心跳
• 紫色箭头是第二个功能，可以通过向NodeServer发送startServer命令来启动HelloServer,实质是通过执行脚本来启动HelloServer

7.2 使用Makefile重新组织源代码

获取源码

为了便于理解，Makefile里没有使用多少“高级”语法，同时为了跟官方代码组织结构看齐，这次索性把目录结构一并调整了，使其看起来比较接近https://github.com/TarsCloud/TarsCpp 中的结构了。

关于Makefile的简单说明：

• 下载源码后，直接make，默认编译出所有的目标文件。如果只想编译一个目标文件，使用make "目标文件"（make HelloServer）即可。如果操作系统还没有安装make，要先安装
• 如果要新增自己的目标文件，需要几个步骤：（1）设置源文件路径_SRC，头文件路径_INC，object文件_OBJ （2）在“Build Target”中设置需要编译出的目标文件名称和依赖的object文件 （3）在“Build Object”中设置_OBJ的编译条件。基本上按照已有的模板修改即可。

8 Framework篇：AdminRegistryServer

8.1 增加AdminRegistryServer

获取源码

• 直接make，可以看到多出两个可执行文件：AdminRegServer和AdminRegServer-client
• 第一个窗口执行:./NodeServer 第二个窗口执行：./AdminRegServer 第三个窗口执行：AdminRegServer-client
• 执行后，打开第四个窗口执行ps -ef | grep HelloServer, 可以看到HelloServer进程已经启动了。

流程图更新：

framework示意图.png

web管理平台发送进程启动指令给AdminRegistryServer，AdminRegistryServer发给NodeServer，最后由NodeServer通过sh脚本启动HelloServer

8.2 小结

framework篇暂时写到这里，这里只想通过HelloServer、NodeServer、AdminRegistryServer这三者的交互，说明framework是怎么一步步建立起来的，以及web是如何控制具体server的。

上面的源代码都是简化版的，旨在说明意图，理解后可以按照个人想法，继续丰富framework