RPC实践-Thrift

Posted 2021-04-24 技术和生活小站

RPC允许程序像调用本地方法一样调用远程服务，它封装了消息的序列化、组装和网络传输的细节。RPC 本身是 client-server模型,也是一种 request-response 协议。大部分采用接口描述语言（Interface Description Language，IDL），方便跨平台的远程过程调用。

RPC调用流程图

服务的调用过程为：

1. client调用client stub，这是一次本地过程调用。

2. client stub将参数打包成一个消息，然后发送这个消息。打包过程也叫做 marshalling。

3. client所在的系统将消息发送给server。

4. server的的系统将收到的包传给server stub。

5. server stub解包得到数据，解包也被称作 unmarshalling。

6. 最后server stub调用服务过程，将结果传给client。

目前的 RPC 框架大致有两种不同的侧重方向，一种偏重于服务治理，另一种偏重于跨语言调用。

服务治理型的 RPC 框架有Dubbo、Motan 等，这类 RPC 框架的特点是功能丰富，提供高性能的远程调用以及服务发现和治理功能，适用于大型服务的微服务化拆分以及管理，对于特定语言（Java）的项目可以十分友好的透明化接入。但缺点是语言耦合度较高，跨语言支持难度较大。

跨语言调用型的 RPC 框架有 Thrift、gRPC、Hessian、Finagle 等，这一类的 RPC 框架重点关注于服务的跨语言调用，能够支持大部分的语言进行语言无关的调用，非常适合于为不同语言提供通用远程服务的场景。但这类框架没有服务发现相关机制，实际使用时一般需要代理层进行请求转发和负载均衡策略控制。

本文将结合一个例子来简单描述thrift的使用方法，并总结一些容易误入的坑。同时引入twitter的thrift连接池，提升高并发调用时单一长连接的性能问题，同时支持服务端多实例部署，避免单点服务故障。

一、定义IDL：IAirLineData.thrift

namespace java com.travelsky.data

namespace py com.travelsky.data

struct Airport {

    1: optional string code

    2: optional string cityName

    3: optional string airportName

    4: optional string airportReferred

    5: optional string javaTimeZone

    6: optional string phone

    7: optional double geolat

    8: optional double geolong

    9: optional string city

    10: optional string cityCode

    11: optional string countryCode

}

service IAirLineData {

    Airport getAirportByCode(1: string airportCode)

}

其中，namespace定义了生成代码的package结构。struct定义了一个机场对象。service定义了RPC的具体方法。注意，机场对象的每个属性都要有唯一的id，并且一旦定义好，不能再更改。如果后续再增加属性，只能排在后面，id累加。原因是thrift在序列化和反序列化对象时，是根据属性id和type来操作的，跟属性名称没有关系。一旦id变化，解析会出错，将导致线上旧版本的client调用出错。这一点，跟ProtoBuffer协议类似。

二、生成具体代码

thrift -r -gen java IAirLineData.thrift

thrift -r -gen py IAirLineData.thrift

三、Server端实现IAirLineData接口的getAirportByCode方法

public class AirLineDataImpl implements IAirLineData.Iface {

    public Airport getAirportByCode(String airportCode){

        // ...业务代码

    }

}

四、Server端启动服务

TNonblockingServerSocket serverTransport = new TNonblockingServerSocket(8798);

IAirLineData.Processor<AirLineDataImpl> processor = new IAirLineData.Processor<AirLineDataImpl>(new AirLineDataImpl());

TThreadedSelectorServer.Args tArgs = new TThreadedSelectorServer.Args(serverTransport);

tArgs.processor(processor);

tArgs.transportFactory(new TFramedTransport.Factory());

tArgs.protocolFactory(new TBinaryProtocol.Factory());

TServer server = new TThreadedSelectorServer(tArgs);

server.serve();

五、Client端远程调用

TFramedTransport transport = new TFramedTransport(new TSocket("Server IP", 8798, 5000));

transport.open();

TProtocol protocol = new TBinaryProtocol(transport);

AirLineDataService.Client client = new AirLineDataService.Client(protocol);

System.out.println(client.getAirportFourCodeByCode("PEK"));

注意，server和client的传输方式要保持一致，这里都为帧传输TFramedTransport，还支持TSocket, TMemoryTransport等其它方式。序列化格式除了默认的TBinaryProtocol，还支持TCompactProtocol, TJsonProtocol, TSimpleJsonProtocol。

以上是提供一个thrift服务大致需要的步骤。在实际运用过程中，发现server端的方法返回值不能为Null, 否则client直接抛出异常TTransportException，导致client调用失败，。这一点在thrift的官网上已经提及，需要特别注意。最好是返回一个特殊字符串，空串也可以。

另外，thrift自身的序列化Tserializer和反序列化TDeserializer，不建议在业务中使用(比如将对象序列化后缓存至redis中)，垃圾回收不好管理，很容易造成内存漏洞，最好换成其它方案（Kryo, FST, hessian2，ProtoBuffer等）。

在客户端大量调用的时，为了减少TCP短连接频繁建立和释放造成的性能损耗，一般采用TCP长连接方式，即多个远程方法复用一个长连接。由于socket是全双工通讯方式，这意味着客户端从发起一个请求，必须等待服务端响应完成后，才能进行下一次的请求。在高并发调用情况下，必须要在通讯传输时加锁，否则socket通讯就乱掉了，thrift直接报TSocketException。但一旦加锁，又会造成服务性能下降。

自然能想到，建立多路TCP长连接，同时服务多个客户端。那么怎么实现一个连接池来管理thrift多路长连接呢？最简单的思路，定义一个数组，初始化批量建立多路连接，然后从中选择一个空闲可用的。虽然对每个具体的长连接还是要加锁，但多个长连接仍然提升了并发性能。

可以利用apache的commons pool包来管理thrift连接池。另外github上也有其它的类似Object连接池，如fast object pool 和 stormpot，见附录参见文档。

这里重点提一下twitter的commons组件库里专门提供的thrift连接池。Maven引用为：

<dependency>

    <groupId>com.twitter.common</groupId>

    <artifactId>thrift</artifactId>

    <version>0.0.79</version>

</dependency>

创建连接池的关键代码是：

Thrift<AirLineDataService.Iface> pool = ThriftFactory.create(AirLineDataService.Iface.class)

        .withMaxConnectionsPerEndpoint(50)

        .withSocketTimeout(Amount.of(30L, Time.SECONDS))

        .useFramedTransport(true)

        .build(ImmutableSet.copyOf(Lists.asList(addr1, addr2, ...)));

它可以方便的指定连接池的最大连接数量，也支持多个服务器endpoint设定，避免单点服务器故障导致的服务中断。

下一篇准备实践dubbo构建机场公共数据查询服务。后续还会从源码角度解读twitter的commons thrift连接池。