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

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

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

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

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

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

virtual bool process(boost::shared_ptr<protocol::TProtocol> in,

                       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内部其他系统,这个文件内容如下:

namespace Java com.facebook.fb303

namespace cpp facebook.fb303

namespace perl Facebook.FB303

enum fb_status {

  DEAD = 0,

  STARTING = 1,

  ALIVE = 2,

  STOPPING = 3,

  STOPPED = 4,

  WARNING = 5,

}

service FacebookService {

  string getName(),

  string getVersion(),

  fb_status getStatus(),

  string getStatusDetails(),

  map<string, i64> getCounters(),

  i64 getCounter(1: string key),

  void setOption(1: string key, 2: string value),

  string getOption(1: string key),

  map<string, string> getOptions(),

  string getCpuProfile(1: i32 profileDurationInSec),

  i64 aliveSince(),

  oneway void reinitialize(),

  oneway void shutdown(),

}

上面这个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)class FacebookService_getName_args {…}

(2)class FacebookService_getName_pargs {…}

(3)typedef struct _FacebookService_getName_result__isset {…} _FacebookService_getName_result__isset;

(4)class FacebookService_getName_result{…}

(5)typedef struct _FacebookService_getName_presult__isset {…} _FacebookService_getName_presult__isset;

(6)class FacebookService_getName_presult{…}

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

send_getName();

recv_getName(_return);

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

int32_t cseqid = 0;

oprot_->writeMessageBegin(“getName”, ::apache::thrift::protocol::T_CALL, cseqid);//写一个函数调用消息RPC

FacebookService_getName_pargs args;

args.write(oprot_);//写入参数

oprot_->writeMessageEnd();

oprot_->getTransport()->writeEnd();

oprot_->getTransport()->flush();//保证这次写入过程立即生效

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

int32_t rseqid = 0;//接收的消息序列号

  std::string fname;//函数名称

  ::apache::thrift::protocol::TMessageType mtype;//消息的类型(调用(T_CALL)、异常(T_EXCEPTION)等)

  iprot_->readMessageBegin(fname, mtype, rseqid);//从返回消息读取函数名称、消息类型

  if (mtype == ::apache::thrift::protocol::T_EXCEPTION) {//处理异常消息

    ::apache::thrift::TApplicationException x;

    x.read(iprot_);

    iprot_->readMessageEnd();

    iprot_->getTransport()->readEnd();

    throw x;

  }

  if (mtype != ::apache::thrift::protocol::T_REPLY) {//处理返回消息

    iprot_->skip(::apache::thrift::protocol::T_STRUCT);

    iprot_->readMessageEnd();

    iprot_->getTransport()->readEnd();

  }

  if (fname.compare(“getName”) != 0) {//看是否是我们需要的函数名,不是就跳过消息读取

    iprot_->skip(::apache::thrift::protocol::T_STRUCT);

    iprot_->readMessageEnd();

    iprot_->getTransport()->readEnd();

  }

  FacebookService_getName_presult result;

  result.success = &_return;

  result.read(iprot_);//读取函数返回值

  iprot_->readMessageEnd();

  iprot_->getTransport()->readEnd();

  if (result.__isset.success) {//成功就返回结果(已经在_return里面),否则抛出异常

    return;

  }

  throw ::apache::thrift::TApplicationException(::apache::thrift::TApplicationException::MISSING_RESULT, “getName failed: unknown result”);

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

bool FacebookServiceProcessor::process(boost::shared_ptr< ::apache::thrift::protocol::TProtocol> piprot, 

boost::shared_ptr< ::apache::thrift::protocol::TProtocol> poprot, void* callContext) {

::apache::thrift::protocol::TProtocol* iprot = piprot.get();

::apache::thrift::protocol::TProtocol* oprot = poprot.get();

std::string fname;

::apache::thrift::protocol::TMessageType mtype;

int32_t seqid;

iprot->readMessageBegin(fname, mtype, seqid);//读取得到函数名称、消息类型和函数序列号

//处理不是函数调用消息的情况

if (mtype != ::apache::thrift::protocol::T_CALL && mtype != ::apache::thrift::protocol::T_ONEWAY) {

iprot->skip(::apache::thrift::protocol::T_STRUCT);

iprot->readMessageEnd();

iprot->getTransport()->readEnd();

    ::apache::thrift::TApplicationException x(::apache::thrift::TApplicationException::INVALID_MESSAGE_TYPE);

//写入(返回)一个异常信息给调用客户端,客户端会根据返回结果处理异常

oprot->writeMessageBegin(fname, ::apache::thrift::protocol::T_EXCEPTION, seqid);

x.write(oprot);

oprot->writeMessageEnd();

oprot->getTransport()->writeEnd();

oprot->getTransport()->flush();

return true;

}

return process_fn(iprot, oprot, fname, seqid, callContext);//调用实际的函数处理

}

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

bool FacebookServiceProcessor::process_fn(::apache::thrift::protocol::TProtocol* iprot,

::apache::thrift::protocol::TProtocol* oprot, std::string& fname, int32_t seqid, void* callContext) {

//定义个map的迭代器,用于接收在函数映射查找到的映射函数

std::map<std::string, void (FacebookServiceProcessor::*)(int32_t, ::apache::thrift::protocol::TProtocol*, 

::apache::thrift::protocol::TProtocol*, void*)>::iterator pfn;

pfn = processMap_.find(fname);//根据函数名称查找对应的映射处理函数

if (pfn == processMap_.end()) {//如果没有找到,做下面的处理

iprot->skip(::apache::thrift::protocol::T_STRUCT);

iprot->readMessageEnd();

iprot->getTransport()->readEnd();

//抛出一个不知道的方法的异常

::apache::thrift::TApplicationException x(::apache::thrift::TApplicationException::UNKNOWN_METHOD, 

“Invalid method name: ‘”+fname+”‘”);

//写入到调用客户端

oprot->writeMessageBegin(fname, ::apache::thrift::protocol::T_EXCEPTION, seqid);

x.write(oprot);

oprot->writeMessageEnd();

oprot->getTransport()->writeEnd();

oprot->getTransport()->flush();

return true;

}

(this->*(pfn->second))(seqid, iprot, oprot, callContext);//调用具体的函数(RPC过程完成)

return true;

}

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

processMap_[“getName”] = &FacebookServiceProcessor::process_getName;

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

void FacebookServiceProcessor::process_getName(int32_t seqid,

::apache::thrift::protocol::TProtocol* iprot, ::apache::thrift::protocol::TProtocol* oprot, void* callContext)

{

void* ctx = NULL;

if (eventHandler_.get() != NULL) {

//得到上下文调用环境

ctx = eventHandler_->getContext(“FacebookService.getName”, callContext);

}

//定义并初始化一个用于释放资源的帮助类对象

::apache::thrift::TProcessorContextFreer freer(eventHandler_.get(), ctx, “FacebookService.getName”);

if (eventHandler_.get() != NULL) {

eventHandler_->preRead(ctx, “FacebookService.getName”);//读之前事件处理

}

FacebookService_getName_args args;

args.read(iprot);

iprot->readMessageEnd();

uint32_t bytes = iprot->getTransport()->readEnd();

if (eventHandler_.get() != NULL) {

eventHandler_->postRead(ctx, “FacebookService.getName”, bytes);//读取和读完之间的事件处理

}

FacebookService_getName_result result;

try {

iface_->getName(result.success);//这是重点:调用服务器端的getName函数

result.__isset.success = true;

} catch (const std::exception& e) {

if (eventHandler_.get() != NULL) {

eventHandler_->handlerError(ctx, “FacebookService.getName”);//错误处理

}

//写入具体的异常到客户端

::apache::thrift::TApplicationException x(e.what());

oprot->writeMessageBegin(“getName”, ::apache::thrift::protocol::T_EXCEPTION, seqid);

x.write(oprot);

oprot->writeMessageEnd();

oprot->getTransport()->writeEnd();

oprot->getTransport()->flush();

return;

}

if (eventHandler_.get() != NULL) {

eventHandler_->preWrite(ctx, “FacebookService.getName”);//写入之前事件处理

}

//写入调用返回值(T_REPLY)消息到调用客户端

oprot->writeMessageBegin(“getName”, ::apache::thrift::protocol::T_REPLY, seqid);

result.write(oprot);

oprot->writeMessageEnd();

bytes = oprot->getTransport()->writeEnd();

oprot->getTransport()->flush();

if (eventHandler_.get() != NULL) {

eventHandler_->postWrite(ctx, “FacebookService.getName”, bytes);//写相应之后处理

}

}

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

(2)scribe服务IDL文件

include “/home/brucewoo/thrift-0.6.1/contrib/fb303/if/fb303.thrift”

namespace cpp scribe.thrift

namespace java scribe.thrift

namespace perl Scribe.Thrift

enum ResultCode

{

  OK,

  TRY_LATER

}

struct LogEntry

{

  1:  string category,

  2:  string message

}

service scribe extends fb303.FacebookService

{

  ResultCode Log(1: list<LogEntry> messages);

}

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

第三节 总结

TProcessor类体系主要定义一个服务生产的框架,通过这个框架生产的各种语言的代码可以实现RPC调用,具体的传输细节、协议和方式是通过后面讲解的内容实现的。

第二节对一个具体服务的实现内容做详细分析,不过都是基于文字描述和代码分析,下面根据scribe服务提供的Log函数怎样完成一次具体的处理过程用下面的图形展示:

这个图形并没有展示内部数据通信的细节,只是简单的说明了一个客户端的调用是怎样完成的,服务器处理还涉及到很多相关细节,将在后面章节中详细分析。

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏枕边书

搭建自己的PHP框架心得(二)

续言 对于本次更新,我想说: 本框架由本人挑时间完善,而我还不是PHP大神级的人物,所以框架漏洞难免,求大神们指出。 本框架的知识点应用都会写在博客里,大家有什...

2438
来自专栏WindCoder

《Linux内核分析》之计算机是如何工作的 实验总结

马马虎虎学完了Python课程,一直想学下linux,看到里面有个linux的就选上了。当初没细看,如今听完第一节课有点傻眼,竟然糊里糊涂给自己找了一科汇编语言...

1231
来自专栏java系列博客

Java面试通关要点汇总集基础篇之参考答案

2224
来自专栏我是攻城师

关于Java里面多线程同步的一些知识

对于任何Java开发者来说多线程和同步是一个非常重要的话题。比较好的掌握同步和线程安全相关的知识将使得我们则更加有优势,同时这些知识并不是非常容易就能熟练掌握的...

1073
来自专栏吴伟祥

Jmockdata随机模拟 Java 数据插件

     Jmockdta是一款实现模拟JAVA类型或对象的实例化并随机初始化对象的数据的工具框架。

1032
来自专栏玩转JavaEE

MongoDB数据类型

上篇文章我们介绍了MongoDB的最基本的增删改查操作,也介绍了一些基础的概念,MongoDB中每条记录称作一个文档,这个文档和我们平时用的JSON有点像,但也...

3375
来自专栏技术博文

PHP编程风格规范

本规范由 EasyChen 借鉴 SINA网络应用开发部《C++开发规范》和互动技术部《PHP4开发规范》,以及phpDocument规范 整理出的开发规范。我...

3667
来自专栏眯眯眼猫头鹰的小树杈

猫头鹰的深夜翻译:理解java的classloader

Java ClassLoader是java运行系统中一个至关重要但是经常被忽略的组件。它负责在运行时寻找并加载类文件。创建自定义的ClassLoader可以彻底...

1394
来自专栏史上最简单的Spring Cloud教程

Openresty最佳案例 | 第2篇:Lua入门

什么是lua Lua 是一种轻量小巧的脚本语言,用标准C语言编写并以源代码形式开放, 其设计目的是为了嵌入应用程序中,从而为应用程序提供灵活的扩展和定制功能...

7479
来自专栏有趣的django

7.python常用模块

time模块 常用表示时间方式: 时间戳,格式化的时间字符串,元组(struct_time) UTC(Coordinated Universal Time,世界...

47111

扫码关注云+社区

领取腾讯云代金券