Thrift之TProcess类体系原理及源码详细解析

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

我的新浪微博：http://weibo.com/freshairbrucewoo。

欢迎大家相互交流，共同提高技术。

之前对Thrift自动生成代码的实现细节做了详细的分析，下面进行处理层的实现做详细分析了！会利用到自动代码生成的知识。

这部分是协议层和用户提供的服务实现之间的纽带，定义了调用服务实现的接口框架，真正实现某种服务接口是通过上一章介绍的代码生成工具生成的代码。本章将介绍这个框架的基本原理，然后通过生成的一个实例来具体介绍怎样完成一次完整的服务，这个可能涉及到下面章节的一些知识，对于这些知识不详细分析其功能，只是介绍它在其中起什么作用。选择的实例是Facebook内部用这个框架实现的一个分布式日志收集系统scribe。下面是这部分相关类的类关系图：

从上图中可以看出TProcessor是这个部分的顶层基类，其他之类基本上都是通过Thrift代码生成工具生成的，只有少数是为了扩展一些功能而直接写代码实现，如PeekProcessor类就增加了一些对原始数据处理的功能。scribeProcessor和FacebookServiceProcessor类就是用代码生成器根据IDL文件生成的，也是我们后面需要分析的一个实例。

第一节服务接口调用框架分析

这个基本的框架包括三个类，一个就是抽象类TProcessor，负责调用用户定义的服务接口，从一个接口读入数据，写入一个输出接口。一个最主要的函数定义如下：

1 virtual bool process(boost::shared_ptr<protocol::TProtocol> in,
2
3 boost::shared_ptr<protocol::TProtocol> out, void* connectionContext) = 0;

这个函数是一个纯虚函数，所以继承这个类的子类都必须实现这个函数，这个函数就是最主要的数据传输功能。

第二个类就是负责处理TProcessor类产生的事件的类TProcessorEventHandler，主要定义了一些当某事件发生时的处理函数，例如当读取参数之前可以做一些处理功能。下面是这个类定义的各个成员函数，每一个函数都处理一种事件发送时的情况：

函数名称	函数功能
getContext	调用其他回调函数之前调用，期望返回一些有序的上下文对象以便传递给其他回调函数使用
freeContext	期望释放一个上下文有关的资源
preRead	在读参数以前调用
postRead	在读参数和处理函数之间调用
preWrite	在处理和写响应之间调用
postWrite	在写响应之后调用
asyncComplete	当一个异步函数成功完成调用时调用
handlerError	如果处理函数抛出没有定义的异常就会调用此函数

最后一个类就是TProcessorContextFreer类，这个类是一个帮助类，帮助生成的代码来释放上下文资源。

第二节基于框架生成的服务实例分析

本节将对scribe服务器采用的服务实现进行详细分析。

1 接口定义语言文件（IDL）

（1）Facebook内部共用服务协议

主要有两个文件，一个是在Thrift中定义，是用于Facebook内部的一些接口服务定义，这个不仅仅用于scribe服务器，可能还用于Facebook内部其他系统，这个文件内容如下：

1 namespace java com.facebook.fb303
2
3 namespace cpp facebook.fb303
4
5 namespace perl Facebook.FB303
6
7 enum fb_status {
8
9 DEAD = 0,
10
11 STARTING = 1,
12
13 ALIVE = 2,
14
15 STOPPING = 3,
16
17 STOPPED = 4,
18
19 WARNING = 5,
20
21 }
22
23 service FacebookService {
24
25 string getName(),
26
27 string getVersion(),
28
29 fb_status getStatus(),
30
31 string getStatusDetails(),
32
33 map<string, i64> getCounters(),
34
35 i64 getCounter(1: string key),
36
37 void setOption(1: string key, 2: string value),
38
39 string getOption(1: string key),
40
41 map<string, string> getOptions(),
42
43 string getCpuProfile(1: i32 profileDurationInSec),
44
45 i64 aliveSince(),
46
47 oneway void reinitialize(),
48
49 oneway void shutdown(),
50
51 }

上面这个IDL文件定义了一个枚举类型用于表示服务的状态，还定义了一个名位FacebookService的服务，里面定义了各种操作，如获取服务状态的操作、得到计数的操作等等。

下面我们来看看根据这个IDL文件生成的C++代码是什么样的一个架构。首先生成了一个基于上面服务定义的抽象类如下：

class FacebookServiceIf { public: virtual ~FacebookServiceIf() {} virtual void getName(std::string& _return) = 0; virtual void getVersion(std::string& _return) = 0; virtual fb_status getStatus() = 0; virtual void getStatusDetails(std::string& _return) = 0; virtual void getCounters(std::map<std::string, int64_t> & _return) = 0; virtual int64_t getCounter(const std::string& key) = 0; virtual void setOption(const std::string& key, const std::string& value) = 0; virtual void getOption(std::string& _return, const std::string& key) = 0; virtual void getOptions(std::map<std::string, std::string> & _return) = 0; virtual void getCpuProfile(std::string& _return, const int32_t profileDurationInSec) = 0; virtual int64_t aliveSince() = 0; virtual void reinitialize() = 0; virtual void shutdown() = 0;};

注意观察，除了这个类多了一个虚析构函数，其他函数就是IDL中定义的。接着定义了类FacebookServiceNull，这个是上面那个抽象类的空实现（就是所有方法都没有做具体的事情），这样做的好处就是我们需要重写一些函数的时候只需要关注我们需要写的函数，而不是重写所有函数。接着又定义了封装每一个函数参数的相应类，就是一个函数的参数都用一个类来封装定义，函数的返回值也是这样处理。这样做的目的是统一远程调用的实现接口，因为传递参数都只需要这个封装类的对象就可以了。所以你会看到每一个服务里面定义的函数都有下面一组类的定义：

1 （1）class FacebookService_getName_args {…}
2
3 （2）class FacebookService_getName_pargs {…}
4
5 （3）typedef struct _FacebookService_getName_result__isset {…} _FacebookService_getName_result__isset;
6
7 （4）class FacebookService_getName_result{…}
8
9 （5）typedef struct _FacebookService_getName_presult__isset {…} _FacebookService_getName_presult__isset;
10
11 （6）class FacebookService_getName_presult{…}

上面这六个类定义就是为服务中的getName函数服务的，相应的每一个函数都会有这种类似的定义和实现。接下来就会定义三个具体实现IDL定义的功能的类，一个客户端的类，它继承定义的服务抽象类，每一个具体的函数实现都是同样的方式和思路，同样我结合getName函数的实现来看看这个过程，其他函数都是这样实现的，代码如下：

1 send_getName();
2
3 recv_getName(_return);

由上面代码可以看出首先调用函数发送函数名称及相关信息到远程，然后接受函数调用的返回值，发送函数send_getName()的代码如下：

1 int32_t cseqid = 0;
2
3 oprot_->writeMessageBegin("getName", ::apache::thrift::protocol::T_CALL, cseqid);//写一个函数调用消息RPC
4
5 FacebookService_getName_pargs args;
6
7 args.write(oprot_);//写入参数
8
9 oprot_->writeMessageEnd();
10
11 oprot_->getTransport()->writeEnd();
12
13 oprot_->getTransport()->flush();//保证这次写入过程立即生效

上面代码就完成了函数名称以及参数的传输，调用的是TProtocol相关的类的函数实现，具体的实现内容和方式会在TProtocol部分介绍。下面接着看一下接收返回值的函数recv_getName的代码：

1 int32_t rseqid = 0;//接收的消息序列号
2
3 std::string fname;//函数名称
4
5 ::apache::thrift::protocol::TMessageType mtype;//消息的类型（调用（T_CALL）、异常（T_EXCEPTION）等）
6
7 iprot_->readMessageBegin(fname, mtype, rseqid);//从返回消息读取函数名称、消息类型
8
9 if (mtype == ::apache::thrift::protocol::T_EXCEPTION) {//处理异常消息
10
11 ::apache::thrift::TApplicationException x;
12
13 x.read(iprot_);
14
15 iprot_->readMessageEnd();
16
17 iprot_->getTransport()->readEnd();
18
19 throw x;
20
21 }
22
23 if (mtype != ::apache::thrift::protocol::T_REPLY) {//处理返回消息
24
25 iprot_->skip(::apache::thrift::protocol::T_STRUCT);
26
27 iprot_->readMessageEnd();
28
29 iprot_->getTransport()->readEnd();
30
31 }
32
33 if (fname.compare("getName") != 0) {//看是否是我们需要的函数名，不是就跳过消息读取
34
35 iprot_->skip(::apache::thrift::protocol::T_STRUCT);
36
37 iprot_->readMessageEnd();
38
39 iprot_->getTransport()->readEnd();
40
41 }
42
43 FacebookService_getName_presult result;
44
45 result.success = &_return;
46
47 result.read(iprot_);//读取函数返回值
48
49 iprot_->readMessageEnd();
50
51 iprot_->getTransport()->readEnd();
52
53 if (result.__isset.success) {//成功就返回结果（已经在_return里面），否则抛出异常
54
55 return;
56
57 }
58
59 throw ::apache::thrift::TApplicationException(::apache::thrift::TApplicationException::MISSING_RESULT, "getName failed: unknown result");

上面代码就是处理远程调用的返回结果，代码里面有注释。一个服务函数的实现大概流程已经展现在我们面前了，处理的过程也已经清晰。这个只是用于客户端的处理流程，必须通过有效的机制来通知服务器端调用相应的函数（这就是RPC）在服务器端完成相应功能并将结果返回。这种机制就是通过我们这部分介绍的TProcessor类实现，这就是上面提到三个类中的第二个类，在这个实例中是FacebookServiceProcessor类，它从TProcessor类继承，重点实现两个函数process和process_fn，其中process会调用process_fn函数来处理客户端具体调用的那个服务函数，process函数定义如下：

1 bool FacebookServiceProcessor::process(boost::shared_ptr< ::apache::thrift::protocol::TProtocol> piprot,
2
3 boost::shared_ptr< ::apache::thrift::protocol::TProtocol> poprot, void* callContext) {
4
5 ::apache::thrift::protocol::TProtocol* iprot = piprot.get();
6
7 ::apache::thrift::protocol::TProtocol* oprot = poprot.get();
8
9 std::string fname;
10
11 ::apache::thrift::protocol::TMessageType mtype;
12
13 int32_t seqid;
14
15 iprot->readMessageBegin(fname, mtype, seqid);//读取得到函数名称、消息类型和函数序列号
16
17 //处理不是函数调用消息的情况
18
19 if (mtype != ::apache::thrift::protocol::T_CALL && mtype != ::apache::thrift::protocol::T_ONEWAY) {
20
21 iprot->skip(::apache::thrift::protocol::T_STRUCT);
22
23 iprot->readMessageEnd();
24
25 iprot->getTransport()->readEnd();
26
27 ::apache::thrift::TApplicationException x(::apache::thrift::TApplicationException::INVALID_MESSAGE_TYPE);
28
29 //写入（返回）一个异常信息给调用客户端，客户端会根据返回结果处理异常
30
31 oprot->writeMessageBegin(fname, ::apache::thrift::protocol::T_EXCEPTION, seqid);
32
33 x.write(oprot);
34
35 oprot->writeMessageEnd();
36
37 oprot->getTransport()->writeEnd();
38
39 oprot->getTransport()->flush();
40
41 return true;
42
43 }
44
45 return process_fn(iprot, oprot, fname, seqid, callContext);//调用实际的函数处理
46
47 }

上面代码有比较详细的注释，还需要说明一点的就是如果传递的不是函数调用的消息类型就会返回给客户端一个异常的消息，客户端的接收返回值的函数就会根据收到的异常消息做相应处理，上面getName函数的接收返回值函数就是抛出一个服务器端给的异常信息。下面继续看最终服务器端调用相应映射函数的处理，这个是通过process_fn函数实现：具体定义如下：

1 bool FacebookServiceProcessor::process_fn(::apache::thrift::protocol::TProtocol* iprot,
2
3 ::apache::thrift::protocol::TProtocol* oprot, std::string& fname, int32_t seqid, void* callContext) {
4
5 //定义个map的迭代器，用于接收在函数映射查找到的映射函数
6
7 std::map<std::string, void (FacebookServiceProcessor::*)(int32_t, ::apache::thrift::protocol::TProtocol*,
8
9 ::apache::thrift::protocol::TProtocol*, void*)>::iterator pfn;
10
11 pfn = processMap_.find(fname);//根据函数名称查找对应的映射处理函数
12
13 if (pfn == processMap_.end()) {//如果没有找到，做下面的处理
14
15 iprot->skip(::apache::thrift::protocol::T_STRUCT);
16
17 iprot->readMessageEnd();
18
19 iprot->getTransport()->readEnd();
20
21 //抛出一个不知道的方法的异常
22
23 ::apache::thrift::TApplicationException x(::apache::thrift::TApplicationException::UNKNOWN_METHOD,
24
25 "Invalid method name: '"+fname+"'");
26
27 //写入到调用客户端
28
29 oprot->writeMessageBegin(fname, ::apache::thrift::protocol::T_EXCEPTION, seqid);
30
31 x.write(oprot);
32
33 oprot->writeMessageEnd();
34
35 oprot->getTransport()->writeEnd();
36
37 oprot->getTransport()->flush();
38
39 return true;
40
41 }
42
43 (this->*(pfn->second))(seqid, iprot, oprot, callContext);//调用具体的函数（RPC过程完成）
44
45 return true;
46
47 }

上面这个函数最终完成了RPC的过程，那个函数与映射函数的对应关系的map结构是在构造函数中初始化的，所以可以找到，例如我们举例的getName函数是下面这样初始化的：

1 processMap_["getName"] = &FacebookServiceProcessor::process_getName;

和getName函数一样，对于IDL定义的每一个函数在FacebookServiceProcessor类中都有一个映射的处理函数，为了展示一个完整的处理过程我们在看看getName函数的映射处理函数process_getName，它的定义如下：

1 void FacebookServiceProcessor::process_getName(int32_t seqid,
2
3 ::apache::thrift::protocol::TProtocol* iprot, ::apache::thrift::protocol::TProtocol* oprot, void* callContext)
4
5 {
6
7 void* ctx = NULL;
8
9 if (eventHandler_.get() != NULL) {
10
11 //得到上下文调用环境
12
13 ctx = eventHandler_->getContext("FacebookService.getName", callContext);
14
15 }
16
17 //定义并初始化一个用于释放资源的帮助类对象
18
19 ::apache::thrift::TProcessorContextFreer freer(eventHandler_.get(), ctx, "FacebookService.getName");
20
21 if (eventHandler_.get() != NULL) {
22
23 eventHandler_->preRead(ctx, "FacebookService.getName");//读之前事件处理
24
25 }
26
27 FacebookService_getName_args args;
28
29 args.read(iprot);
30
31 iprot->readMessageEnd();
32
33 uint32_t bytes = iprot->getTransport()->readEnd();
34
35 if (eventHandler_.get() != NULL) {
36
37 eventHandler_->postRead(ctx, "FacebookService.getName", bytes);//读取和读完之间的事件处理
38
39 }
40
41 FacebookService_getName_result result;
42
43 try {
44
45 iface_->getName(result.success);//这是重点：调用服务器端的getName函数
46
47 result.__isset.success = true;
48
49 } catch (const std::exception& e) {
50
51 if (eventHandler_.get() != NULL) {
52
53 eventHandler_->handlerError(ctx, "FacebookService.getName");//错误处理
54
55 }
56
57 //写入具体的异常到客户端
58
59 ::apache::thrift::TApplicationException x(e.what());
60
61 oprot->writeMessageBegin("getName", ::apache::thrift::protocol::T_EXCEPTION, seqid);
62
63 x.write(oprot);
64
65 oprot->writeMessageEnd();
66
67 oprot->getTransport()->writeEnd();
68
69 oprot->getTransport()->flush();
70
71 return;
72
73 }
74
75 if (eventHandler_.get() != NULL) {
76
77 eventHandler_->preWrite(ctx, "FacebookService.getName");//写入之前事件处理
78
79 }
80
81 //写入调用返回值（T_REPLY）消息到调用客户端
82
83 oprot->writeMessageBegin("getName", ::apache::thrift::protocol::T_REPLY, seqid);
84
85 result.write(oprot);
86
87 oprot->writeMessageEnd();
88
89 bytes = oprot->getTransport()->writeEnd();
90
91 oprot->getTransport()->flush();
92
93 if (eventHandler_.get() != NULL) {
94
95 eventHandler_->postWrite(ctx, "FacebookService.getName", bytes);//写相应之后处理
96
97 }
98
99 }

上面这个函数就是真正完成服务器端调用客户端传递过来的函数的处理过程，有事件处理类处理相应的事件（不过，目前都还是空实现，以后可以继承这个处理类重写需要处理事件的函数，例如：在调用服务器真正的处理函数之前可以先处理一下参数，验证参数是否正确之类的），也有帮助释放资源的帮助类。

（2）scribe服务IDL文件

1 include "/home/brucewoo/thrift-0.6.1/contrib/fb303/if/fb303.thrift"
2
3 namespace cpp scribe.thrift
4
5 namespace java scribe.thrift
6
7 namespace perl Scribe.Thrift
8
9 enum ResultCode
10
11 {
12
13 OK,
14
15 TRY_LATER
16
17 }
18
19 struct LogEntry
20
21 {
22
23 1: string category,
24
25 2: string message
26
27 }
28
29 service scribe extends fb303.FacebookService
30
31 {
32
33 ResultCode Log(1: list<LogEntry> messages);
34
35 }

这个IDL文件只定义了一个服务接口，就是用完成日志文件传输的几个Log，不过这个服务继承FacebookService服务，所以上面介绍FacebookService服务的功能它也具备，传输日志的结构就是分类和具体的消息。这个服务的具体实现和上面介绍的FacebookService流程都是一样的，不在详细介绍，只要知道一点就是：客户端在调用Log写日志到scribe服务器的时候就会传递到服务器端来调用同名的函数处理日志。