基于Protobuf的分布式高性能RPC框架——Navi-Pbrpc

基于Protobuf的分布式高性能RPC框架——Navi-Pbrpc

• 二月 8, 2016

1 简介

Navi-pbrpc框架是一个高性能的远程调用RPC框架，使用netty4技术提供非阻塞、异步、全双工的信道，使用protobuf做为序列化协议，同时提供长、短链接模式，支持non-blocking和传统的blocking io，以及负载均衡，容错处理策略等，对于基于socket的分布式调用提供通讯基础。

若是你的项目中须要高性能的RPC解决方案，那么navi-pbrpc能够帮助到你构建一个强大的远程调用系统。

Navi-pbrpc使用netty nio开发，全双工、异步、非阻塞的通讯模型，保证了高性能和理想的QPS，了解详细性能测试报告见附录性能测试。

单测覆盖率见附录。

设计关于UML类图见附录。 

github已开源，连接请点此https://github.com/neoremind/navi-pbrpc。

 

2 协议介绍

Navi-pbrpc通讯模型以下，服务端与客户端通讯采用4层TCP Socket通讯，支持长、短链接链路，应用层采用header+body方式做为一个package或者叫作frame，header内含的body length属性来代表二进制数据长度，body采用通过protobuf压缩后的二进制数据。 

 

------------- ------------- | | | | | 客户端 | | 服务端 | | | | | | | | | | | | | | 应用层 | ----NsHead + protobuf序列化body(byte[])-----| 应用层 | |-------------| |-------------| | | ----------- 全双工短链接tcp socket --------| | | | ------------[全双工长链接tcp socket]---------| | | | . | | | | . | | | 传输层 | (1-n条channel) | 传输层 | | | . | | | | . | | | | ------------[全双工长链接tcp socket]---------| | |-------------| |-------------| | 网络层 | | 网络层 | |-------------| |-------------| | 链路层 | | 链路层 | |-------------| |-------------| | 物理层 | ================== <<->> ================= | 物理层 | ------------- -------------

Header在框架内部叫作NsHead，NsHead + protobuf序列化body包结构示意以下，关于NsHead头结构更多信息见附录。

Byte/ 0 | 1 | 2 | 3 | / | | | | |0 1 2 3 4 5 6 7|0 1 2 3 4 5 6 7|0 1 2 3 4 5 6 7|0 1 2 3 4 5 6 7| +---------------+---------------+---------------+---------------+ 0/ NsHead / / / / / / / +---------------+---------------+---------------+---------------+ 36/ protobuf序列化后的数据 / +/ (body长度在NsHead中定义) / +---------------+---------------+---------------+---------------+

 

3 使用方法

3.1 准备工做

使用Maven管理的工程POM依赖请添加：

<dependency> <groupId>com.baidu.beidou</groupId> <artifactId>navi-pbrpc</artifactId> <version>1.1.1</version> </dependency>

最新依赖请查找：Sonatype（https://oss.sonatype.org/#nexus-search;quick%7Enavi-pbrpc） 

Maven依赖树以下：

+- commons-pool:commons-pool:jar:1.5.7:compile +- com.google.protobuf:protobuf-java:jar:2.5.0:compile +- io.netty:netty-all:jar:4.0.28.Final:compile +- org.javassist:javassist:jar:3.18.1-GA:compile +- org.slf4j:slf4j-api:jar:1.7.7:compile +- org.slf4j:slf4j-log4j12:jar:1.7.7:compile | \- log4j:log4j:jar:1.2.17:compile

3.2 服务端开发

3.2.1 protoc生成代码

首先定义服务的proto，例如新建一个demo.proto文件，内容以下：

package com.baidu.beidou.navi.pbrpc.demo.proto;   option cc_generic_services = true;   message DemoRequest { optional int32 user_id = 1; }   message DemoResponse { optional int32 user_id = 1; optional string user_name = 2; enum GenderType { MALE = 1; FEMALE = 2; } optional GenderType gender_type = 3; }

使用protoc命令编译，生成Demo.java，方法见附录。

3.2.2 开发服务实现

开发一个服务端的实现，例如DemoServiceImpl，代码以下：

public class DemoServiceImpl implements DemoService {   @Override public DemoResponse doSmth(DemoRequest req) { DemoResponse.Builder builder = DemoResponse.newBuilder(); builder.setUserId(1); builder.setUserName("name-1"); builder.setGenderType(DemoResponse.GenderType.MALE); return builder.build(); } }

特别注意，一个方法若想暴露为服务必须知足以下限制：

• 参数必须只有1个。
• 参数和返回值类型必须为继承自com.google.protobuf.GeneratedMessage。由protoc生成的java bean都会继承这个类。

3.2.3 暴露而且启动服务

启动服务端，代码以下： 

PbrpcServer server = new PbrpcServer(8088); server.register(100, new DemoServiceImpl()); server.start();

表示开放端口为8088，将DemoServiceImpl这个对象中的方法注入server，做为服务。register(int, Object)中的第一个参数做为服务标示的起始值，默认会遍历Object中的全部方法，把符合上述限制条件的方法暴露为服务，其标示从int起始值开始，依次递增1，这个例子中DemoServiceImpl.doSmth(..)方法的标示就是100，若是还有其余方法能够暴露，则从101开始递增。

这里注意，服务端默认若是全双工的channel链路在1个小时以内没有任何数据写入，那么会自动关闭该链路，避免浪费服务端资源。Navi-rpc短链接调用不受影响，对于池化的长链接再下次发起请求的时候会从新make connection，若是是非Navi-rpc客户端的其余长链接接入，请注意这个限制。

3.2.4 关闭服务

安全关闭链接的方法以下：

server.shutdown();

 

4 客户端开发

4.1 同步调用与异步调用

在下面的代码示例中，会看到client调用远程RPC，会有同步以及异步的方式，做为异步方式的调用示例以下：

// 异步调用 CallFuture<demoresponse> future = client.asyncTransport(DemoResponse.class, msg); // 阻塞线程，等待结果 DemoResponse res = future.get(); </demoresponse>

调用客户端能够发送完请求后，拿到future，选择作其余逻辑，或者在get()上阻塞等待。

做为同步方式的调用示例以下： 

// 同步调用 DemoResponse res = client.syncTransport(DemoResponse.class, msg);

调用客户端会一直阻塞等待。

4.2 nio短链接调用

// 构造客户端 PbrpcClient client = PbrpcClientFactory.buildShortLiveConnection("127.0.0.1", 8088, 60000);   // 构建请求数据 DemoRequest.Builder req = DemoRequest.newBuilder(); req.setUserId(1); byte[] data = req.build().toByteArray();   // 构造请求消息 PbrpcMsg msg = new PbrpcMsg(); msg.setServiceId(100); msg.setProvider("beidou"); msg.setData(data);   // 异步调用 CallFuture<demoresponse> future = client.asyncTransport(DemoResponse.class, msg); // 阻塞线程，等待结果 DemoResponse res = future.get(); // 打印结果 System.out.println(res); 这里注意，一旦PbrpcClient创建好是能够复用的，无需每次从新新建。 </demoresponse>

PbrpcClientFactory是一个client工厂，帮助构造短链接调用，其余参数以下：

public static SimplePbrpcClient buildShortLiveConnection(String ip, int port); public static SimplePbrpcClient buildShortLiveConnection(String ip, int port, int readTimeout); public static SimplePbrpcClient buildShortLiveConnection(String ip, int port, int connTimeout, int readTimeout);

其中connTimeout表示客户端链接时间，单位毫秒。

readTimeout表示客户端调用时间，单位毫秒，超时会抛出TimeoutException。例如以下：

Exception in thread "main" java.lang.RuntimeException: Error occurrs due to Client call timeout, request logId=1696636656 at com.baidu.beidou.navi.pbrpc.client.callback.CallFuture.get(CallFuture.java:97) at com.baidu.beidou.navi.pbrpc.client.PooledPbrpcClient.syncTransport(PooledPbrpcClient.java:109) at com.baidu.unbiz.soma.biz.siconf.rpc.pbrpc.product.protocol.TestBiz.main(TestBiz.java:31) Caused by: com.baidu.beidou.navi.pbrpc.exception.TimeoutException: Client call timeout, request logId=1696636656 at com.baidu.beidou.navi.pbrpc.client.TimeoutEvictor.detectTimetout(TimeoutEvictor.java:68) at com.baidu.beidou.navi.pbrpc.client.TimeoutEvictor.run(TimeoutEvictor.java:47) at java.util.TimerThread.mainLoop(Timer.java:512) at java.util.TimerThread.run(Timer.java:462)

4.3 nio长链接池调用

链接池默认开启8个keepAlive长链接，代码以下：

// 构造客户端 PbrpcClient client = PbrpcClientFactory.buildPooledConnection(new PooledConfiguration(), "127.0.0.1", 8088, 60000);   // 构建请求数据 DemoRequest.Builder req = DemoRequest.newBuilder(); req.setUserId(1); byte[] data = req.build().toByteArray();   // 构造请求消息 PbrpcMsg msg = new PbrpcMsg(); msg.setServiceId(100); msg.setProvider("beidou"); msg.setData(data);   // 异步调用 CallFuture<demoresponse> future = client.asyncTransport(DemoResponse.class, msg); // 阻塞线程，等待结果 DemoResponse res = future.get(); // 打印结果 System.out.println(res); </demoresponse>

其中PooledConfiguration能够设置链接池相关的参数，例如多少个长链接等策略。

PbrpcClientFactory是一个client工厂，帮助构造长链接池调用，其余参数以下： 

public static PooledPbrpcClient buildPooledConnection(String ip, int port); public static PooledPbrpcClient buildPooledConnection(String ip, int port, int readTimeout); public static PooledPbrpcClient buildPooledConnection(PooledConfiguration configuration, String ip, int port, int readTimeout); public static PooledPbrpcClient buildPooledConnection(PooledConfiguration configuration, String ip, int port, int connTimeout, int readTimeout);

其中connTimeout表示客户端链接时间，单位毫秒。

readTimeout表示客户端调用时间，单位毫秒，超时会抛出TimeoutException。

4.4 Blocking IO短链接调用 

// 构造客户端 PbrpcClient client = PbrpcClientFactory.buildShortLiveBlockingIOConnection("127.0.0.1", 8088, 60000);   // 构建请求数据 DemoRequest.Builder req = DemoRequest.newBuilder(); req.setUserId(1); byte[] data = req.build().toByteArray();   // 构造请求消息 PbrpcMsg msg = new PbrpcMsg(); msg.setServiceId(100); msg.setProvider("beidou"); msg.setData(data);   // 同步调用，blocking IO只支持同步调用 DemoResponse res = client.syncTransport(DemoResponse.class, msg);   // 打印结果 System.out.println(res);

默认只支持同步调用，其余构造方法以下：

public static BlockingIOPbrpcClient buildShortLiveBlockingIOConnection(String ip, int port); public static BlockingIOPbrpcClient buildShortLiveBlockingIOConnection(String ip, int port, int readTimeout); public static BlockingIOPbrpcClient buildShortLiveBlockingIOConnection(String ip, int port, int connTimeout, int readTimeout);

特别注意，调用一个不能定位logId的pbrpc服务，请必须使用blocking IO方式，半双工通讯方式，即一问一答，流程以下图所示：

1.request -------------------------> client --------single TCP connection-------- server <-------------------------2.response

对于netty nio来讲没法标示到全双工后服务端发送回来的一个包到底映射到本地哪一个调用请求上，对于经过Navi-pbrpc暴露的service服务，各类方式能够随意使用。

4.5 Blocking IO长链接池调用

// 构造客户端 PbrpcClient client = PbrpcClientFactory.buildPooledBlockingIOConnection("127.0.0.1", 8088, 60000);   // 构建请求数据 DemoRequest.Builder req = DemoRequest.newBuilder(); req.setUserId(1); byte[] data = req.build().toByteArray();   // 构造请求消息 PbrpcMsg msg = new PbrpcMsg(); msg.setServiceId(100); msg.setProvider("beidou"); msg.setData(data);   // 同步调用，blocking IO只支持同步调用 DemoResponse res = client.syncTransport(DemoResponse.class, msg);   // 打印结果 System.out.println(res);

默认只支持同步调用，其余构造方法以下：

public static BlockingIOPooledPbrpcClient buildPooledBlockingIOConnection(String ip, int port); public static BlockingIOPooledPbrpcClient buildPooledBlockingIOConnection(String ip, int port, int readTimeout); public static BlockingIOPooledPbrpcClient buildPooledBlockingIOConnection( PooledConfiguration configuration, String ip, int port, int readTimeout); public static BlockingIOPooledPbrpcClient buildPooledBlockingIOConnection( PooledConfiguration configuration, String ip, int port, int connTimeout, int readTimeout);

4.6 带有负载均衡以及容错策略的HA客户端调用

// 构造客户端 PbrpcClient client = HAPbrpcClientFactory.buildShortLiveConnection("127.0.0.1:8088,1.1.1.1:9999", new RRLoadBalanceStrategy(new FailOverStrategy(2)));   // 构建请求数据 DemoRequest.Builder req = DemoRequest.newBuilder(); req.setUserId(1); byte[] data = req.build().toByteArray();   // 构造请求消息 PbrpcMsg msg = new PbrpcMsg(); msg.setServiceId(100); msg.setProvider("beidou"); msg.setData(data);   // 异步调用 CallFuture<demoresponse> future = client.asyncTransport(DemoResponse.class, msg); // 阻塞线程，等待结果 DemoResponse res = future.get(); // 打印结果 System.out.println(res); </demoresponse>

其中HAPbrpcClientFactory是负责构造高可用客户端的工厂，第一个参数是一个IP:PORT串，按照逗号分隔。

其后面的参数是可扩展的负载均衡策略和容错处理策略，RRLoadBalanceStrategy表示使用轮训（Round Robin）策略，FailOverStrategy表示容错策略为失败重试，最多重试次数为2。

还支持的其余策略组合为RandomLoadBalanceStrategy标示随机策略，FailFastStrategy表示失败当即退出。能够随意组合。

其余构造方法以下：

public static HAPbrpcClient buildShortLiveConnection(String connectString, LoadBalanceStrategy lb); public static HAPbrpcClient buildShortLiveConnection(String connectString, int readTimeout, LoadBalanceStrategy lb); public static HAPbrpcClient buildShortLiveConnection(String connectString, int connTimeout, int readTimeout, LoadBalanceStrategy lb); public static HAPbrpcClient buildPooledConnection(String connectString, LoadBalanceStrategy lb); public static HAPbrpcClient buildPooledConnection(String connectString, int readTimeout, LoadBalanceStrategy lb); public static HAPbrpcClient buildPooledConnection(PooledConfiguration configuration, String connectString, int readTimeout, LoadBalanceStrategy lb); public static HAPbrpcClient buildPooledConnection(PooledConfiguration configuration, String connectString, int connTimeout, int readTimeout, LoadBalanceStrategy lb);   public static HAPbrpcClient buildShortLiveBlockingIOConnection(String connectString, LoadBalanceStrategy lb); public static HAPbrpcClient buildShortLiveBlockingIOConnection(String connectString, int readTimeout, LoadBalanceStrategy lb); public static HAPbrpcClient buildShortLiveBlockingIOConnection(String connectString, int connTimeout, int readTimeout, LoadBalanceStrategy lb); public static HAPbrpcClient buildShortLiveBlockingIOConnection( PbrpcClientConfiguration configuration, String connectString, int connTimeout, int readTimeout, LoadBalanceStrategy lb); public static HAPbrpcClient buildPooledBlockingIOConnection(String connectString, LoadBalanceStrategy lb); public static HAPbrpcClient buildPooledBlockingIOConnection(String connectString, int readTimeout, LoadBalanceStrategy lb); public static HAPbrpcClient buildPooledBlockingIOConnection(PooledConfiguration configuration, String connectString, int readTimeout, LoadBalanceStrategy lb); public static HAPbrpcClient buildPooledBlockingIOConnection(String connectString, int connTimeout, int readTimeout, LoadBalanceStrategy lb); public static HAPbrpcClient buildPooledBlockingIOConnection(PooledConfiguration configuration, String connectString, int connTimeout, int readTimeout, LoadBalanceStrategy lb); public static HAPbrpcClient buildPooledBlockingIOConnection(PooledConfiguration configuration, PbrpcClientConfiguration clientConfig, String connectString, int connTimeout, int readTimeout, LoadBalanceStrategy lb);

4.7 关闭链接

安全关闭链接和各类链接池的方法以下：

client.shutdown();

 

5 与Spring集成

5.1 准备工做

Maven POM依赖请添加：

<dependency> <groupId>com.baidu.beidou</groupId> <artifactId>navi-pbrpc-spring</artifactId> <version>1.1.1</version> </dependency>

5.2 开发服务接口

1）根据服务提供方的proto文件生成java代码。此处省略具体方法。详细见第一部分。

2）开发一个Java的Interface

接口名称随意，达意便可。

入参有且仅有一个请求类型，参数和返回值类型必须继承自com.google.protobuf.GeneratedMessage。由protoc生成的java bean都会继承这个类。

方法名随意，达意便可。

方法上加入一个PbrpcMethodId的注解，标明远程服务的method id，若是没有注解则默认为0。

一个实例以下，这里的DemoResponse和DemoRequest都是根据proto生成的java类定义，100标示远程服务的method id标识。

/** * ClassName: DemoService <br /> * Function: 远程服务接口demo * * @author Zhang Xu */ public interface DemoService {   /** * 干点什么 * * @param req 请求 * @return 响应 */ @PbrpcMethodId(100) DemoResponse doSmth(DemoRequest req);   }

5.3 配置XML

一般项目均会与Spring集成，利用Spring的IoC配置管理，能够作到功能的灵活插拔可扩展，一个最经常使用的典型配置是

使用properties文件中配置的IP:PORT列表标示远程服务

使用短链接blocking io访问远程服务

将下面的配置加入到你的XML文件便可，说明全在注释中。

<?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context" xmlns:aop="http://www.springframework.org/schema/aop" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop.xsd">   <aop:aspectj-autoproxy proxy-target-class="true"/>   <context:annotation-config/> <context:component-scan base-package="com.baidu.beidou"/>   <!-- properties配置文件，内含ip端口列表或者一些timeout设置 --> <bean id="propertyPlaceholderConfigurerConfig" class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer"> <property name="systemPropertiesModeName" value="SYSTEM_PROPERTIES_MODE_OVERRIDE"/> <property name="ignoreResourceNotFound" value="true"/> <property name="ignoreUnresolvablePlaceholders" value="true"/> <property name="locations"> <list> <value>classpath:autoevict/application.properties</value> </list> </property> </bean>   <!-- 自动剔除传输回调callback，单位时间内调用失败率大于某个百分比，则剔除掉该客户端 --> <!-- 下面的例子表示服务启动后2s（initDelay）开始第一次检查，检查周期是6s（checkPeriod）， --> <!-- 检查周期内错误率大于80%（maxFailPercentage）而且调用次数大于3次（minInvokeNumber）则剔除 --> <bean id="autoEvictTransportCallback" class="com.baidu.beidou.navi.pbrpc.client.AutoEvictTransportCallback"> <property name="checkPeriod" value="6000"/> <property name="minInvokeNumber" value="3"/> <property name="initDelay" value="2000"/> <property name="maxFailPercentage" value="80"/> </bean>   <!-- 高可用相关配置，FailOverStrategy表明失败重试，FailFastStrategy表明失败当即退出 --> <!-- 负载均衡配置中，RRLoadBalanceStrategy表明轮训调用服务器，RandomLoadBalanceStrategy表明随机选择服务器调用 --> <!-- 默认transportCallback不作任何事情，能够配置AutoEvictTransportCallback作自动剔除失效连接 --> <bean id="failoverStrategy" class="com.baidu.beidou.navi.pbrpc.client.ha.FailOverStrategy"/> <bean id="roundRobinLoadBalanceStrategy" class="com.baidu.beidou.navi.pbrpc.client.ha.RRLoadBalanceStrategy"> <property name="failStrategy" ref="failoverStrategy"/> <property name="transportCallback" ref="autoEvictTransportCallback"/> </bean>   <!-- Pbprc服务server定位locator工厂，这里使用BlockingIO短链接 --> <bean id="pbrpcServerLocator" class="com.baidu.beidou.navi.pbrpc.client.IpPortShortLiveBlockingIOPbrpcServerLocator"/>   <!-- 经过Pbprc服务server定位locator工厂构造高可用客户端 --> <bean id="haPbrpcClient" factory-bean="pbrpcServerLocator" factory-method="factory"> <constructor-arg value="${pbrpc.client.server}"/> <constructor-arg value="${pbrpc.client.connect.timeout}"/> <constructor-arg value="${pbrpc.client.read.timeout}"/> <constructor-arg ref="roundRobinLoadBalanceStrategy"/> </bean>   <!-- Pbprc代理proxy生成器，须要指定高可用pbrpc客户端和provider标示 --> <!-- 这里的proxy是利用jdk的动态代理技术构建的，proxy也可使用javassist动态字节码技术生成 --> <bean id="integrationProxy" class="com.baidu.beidou.navi.pbrpc.spring.JdkDynamicIntegrationProxy"> <property name="pbrpcClient" ref="haPbrpcClient"/> <property name="provider" value="beidou"/> </bean>   <!-- 服务bean定义，使用Spring的FactoryBean来作bean代理，可使用Resource注解注入这个bean --> <bean id="demoService" class="com.baidu.beidou.navi.pbrpc.spring.PbrpcProxyFactoryBean"> <property name="integrationProxy" ref="integrationProxy"/> <property name="serviceInterface"> <value>com.baidu.beidou.navi.pbrpc.demo.service.DemoService</value> </property> </bean>   </beans>

properties配置以下：

pbrpc.client.server=127.0.0.1:14419,127.0.0.1:14420 pbrpc.client.connect.timeout=2000 pbrpc.client.read.timeout=5000

了解更多可选配置见下面小节。

5.4 开始调用

因为上面配置了DemoService的代理，所以能够用@Resource很天然地来使用bean，一个testcase以下。 

@ContextConfiguration(locations = {"classpath*:applicationContext.xml"}) public class SpringIntegrationIpPortListTest extends AbstractJUnit4SpringContextTests {   @Autowired private DemoService demoService;   @Test public void testDoSmth() { Demo.DemoRequest.Builder req = Demo.DemoRequest.newBuilder(); req.setUserId(1); Demo.DemoResponse response = demoService.doSmth(req.build()); System.out.println(response); assertThat(response.getUserId(), is(1)); } }

5.5 其余配置

5.5.1 单点的配置IP:PORT而且不启用自动失效剔除

<?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context" xmlns:aop="http://www.springframework.org/schema/aop" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop.xsd">   <aop:aspectj-autoproxy proxy-target-class="true"/>   <context:annotation-config/> <context:component-scan base-package="com.baidu.beidou"/>   <!-- properties配置文件，内含ip端口列表或者一些timeout设置 --> <bean id="propertyPlaceholderConfigurerConfig" class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer"> <property name="systemPropertiesModeName" value="SYSTEM_PROPERTIES_MODE_OVERRIDE"/> <property name="ignoreResourceNotFound" value="true"/> <property name="ignoreUnresolvablePlaceholders" value="true"/> <property name="locations"> <list> <value>classpath:ipportlist/application.properties</value> </list> </property> </bean>   <!-- 高可用相关配置，FailOverStrategy表明失败重试，FailFastStrategy表明失败当即退出 --> <!-- 负载均衡配置中，RRLoadBalanceStrategy表明轮训调用服务器，RandomLoadBalanceStrategy表明随机选择服务器调用 --> <!-- 默认transportCallback不作任何事情，能够配置AutoEvictTransportCallback作自动剔除失效连接 --> <bean id="failoverStrategy" class="com.baidu.beidou.navi.pbrpc.client.ha.FailOverStrategy"/> <bean id="roundRobinLoadBalanceStrategy" class="com.baidu.beidou.navi.pbrpc.client.ha.RRLoadBalanceStrategy"> <property name="failStrategy" ref="failoverStrategy"/> </bean>   <!-- 手工配置单点pbrpc客户端，能够配置1到多个 --> <!-- 这里使用BlockingIO短链接 --> <bean id="pbrpcClient1" class="com.baidu.beidou.navi.pbrpc.client.BlockingIOPbrpcClient"> <property name="ip" value="${pbrpc.client1.ip}"/> <property name="port" value="${pbrpc.client1.port}"/> <property name="readTimeout" value="${pbrpc.client.read.timeout}"/> <property name="connTimeout" value="${pbrpc.client.connect.timeout}"/> </bean> <bean id="pbrpcClient2" class="com.baidu.beidou.navi.pbrpc.client.BlockingIOPbrpcClient"> <property name="ip" value="${pbrpc.client2.ip}"/> <property name="port" value="${pbrpc.client2.port}"/> <property name="readTimeout" value="${pbrpc.client.read.timeout}"/> <property name="connTimeout" value="${pbrpc.client.connect.timeout}"/> </bean>   <!-- 高可用pbrpc客户端，集成多个单点客户端以及负载均衡策略 --> <bean id="haPbrpcClient" class="com.baidu.beidou.navi.pbrpc.client.HAPbrpcClient"> <property name="loadBalanceStrategy" ref="roundRobinLoadBalanceStrategy"/> <property name="clientList"> <list> <ref bean="pbrpcClient1"/> <ref bean="pbrpcClient2"/> </list> </property> </bean>   <!-- Pbprc代理proxy生成器，须要指定高可用pbrpc客户端和provider标示 --> <bean id="integrationProxy" class="com.baidu.beidou.navi.pbrpc.spring.JdkDynamicIntegrationProxy"> <property name="pbrpcClient" ref="haPbrpcClient"/> <property name="provider" value="beidou"/> </bean>   <!-- 服务bean定义，使用Spring的FactoryBean来作代理 --> <bean id="demoService" class="com.baidu.beidou.navi.pbrpc.spring.PbrpcProxyFactoryBean"> <property name="integrationProxy" ref="integrationProxy"/> <property name="serviceInterface"> <value>com.baidu.beidou.navi.pbrpc.demo.service.DemoService</value> </property> </bean>   </beans>

5.5.2 使用长链接池 

<?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context" xmlns:aop="http://www.springframework.org/schema/aop" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop.xsd">   <aop:aspectj-autoproxy proxy-target-class="true"/>   <context:annotation-config/> <context:component-scan base-package="com.baidu.beidou"/>   <!-- properties配置文件，内含ip端口列表或者一些timeout设置 --> <bean id="propertyPlaceholderConfigurerConfig" class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer"> <property name="systemPropertiesModeName" value="SYSTEM_PROPERTIES_MODE_OVERRIDE"/> <property name="ignoreResourceNotFound" value="true"/> <property name="ignoreUnresolvablePlaceholders" value="true"/> <property name="locations"> <list> <value>classpath:ipportstring_pooled/application.properties</value> </list> </property> </bean>   <!-- 高可用相关配置，FailOverStrategy表明失败重试，FailFastStrategy表明失败当即退出 --> <!-- 负载均衡配置中，RRLoadBalanceStrategy表明轮训调用服务器，RandomLoadBalanceStrategy表明随机选择服务器调用 --> <!-- 默认transportCallback不作任何事情，能够配置AutoEvictTransportCallback作自动剔除失效连接 --> <bean id="failoverStrategy" class="com.baidu.beidou.navi.pbrpc.client.ha.FailOverStrategy"/> <bean id="roundRobinLoadBalanceStrategy" class="com.baidu.beidou.navi.pbrpc.client.ha.RRLoadBalanceStrategy"> <property name="failStrategy" ref="failoverStrategy"/> </bean>   <!-- Pbprc服务server定位locator工厂，这里使用BlockingIO长链接池 --> <bean id="pbrpcServerLocator" class="com.baidu.beidou.navi.pbrpc.client.IpPortPooledBlockingIOPbrpcServerLocator"/>   <!-- 经过Pbprc服务server定位locator工厂构造高可用客户端 --> <bean id="haPbrpcClient" factory-bean="pbrpcServerLocator" factory-method="factory"> <constructor-arg value="${pbrpc.client.server}"/> <constructor-arg value="${pbrpc.client.connect.timeout}"/> <constructor-arg value="${pbrpc.client.read.timeout}"/> <constructor-arg ref="roundRobinLoadBalanceStrategy"/> </bean>   <!-- Pbprc代理proxy生成器，须要指定高可用pbrpc客户端和provider标示 --> <bean id="integrationProxy" class="com.baidu.beidou.navi.pbrpc.spring.JdkDynamicIntegrationProxy"> <property name="pbrpcClient" ref="haPbrpcClient"/> <property name="provider" value="beidou"/> </bean>   <!-- 服务bean定义，使用Spring的FactoryBean来作代理 --> <bean id="demoService" class="com.baidu.beidou.navi.pbrpc.spring.PbrpcProxyFactoryBean"> <property name="integrationProxy" ref="integrationProxy"/> <property name="serviceInterface"> <value>com.baidu.beidou.navi.pbrpc.demo.service.DemoService</value> </property> </bean>   </beans>

 

6 附录

6.1 NsHead头结构

Byte/ 0 | 1 | 2 | 3 | / | | | | |0 1 2 3 4 5 6 7|0 1 2 3 4 5 6 7|0 1 2 3 4 5 6 7|0 1 2 3 4 5 6 7| +---------------+---------------+---------------+---------------+ 0| id | flags | +---------------+---------------+---------------+---------------+ 4| log id | +---------------+---------------+---------------+---------------+ 8| provider | + + | | + + 16| | + + 20| | +---------------+---------------+---------------+---------------+ 24| magic number | +---------------+---------------+---------------+---------------+ 28| method id | +---------------+---------------+---------------+---------------+ 32| body length | +---------------+---------------+---------------+---------------+ Total 36 bytes

Header各字段含义

1. id请求的id。目前未使用。建议设置为0。
2. flags本次请求的一些标志符。目前框架用于传输errorCode。
3. log-id。本次请求的日志id。Navi-rpc服务端用该id定位一个惟一的客户端请求。
4. provider标识调用方的表示。
5. magic-number特殊标识，用于标识一个包的完整性。目前未使用。
6. method-id是RPC方法的序列号。根据proto文件中定义的service顺序，从注册进入的起始值开始依次递增。
7. body-length消息体长度。

6.2 设计——UML类图

6.3 性能测试报告

测试环境： Linux内核版本：2.6.32_1-11-0-0 CPU：Intel(R) Xeon(R) CPU E5-2620 0 @ 2.00GHz processor_count : 12 内存：64G 在同一台物理机上测试。

JVM参数： -Xms512m -Xmx512m

测试压力： 10w请求，20并发，测试期间会有4个以上的核所有100%负荷。

测试case： 客户端发起请求，要求字符串长度以及数量，服务端返回一个指定数量的List给予客户端，字符串为随机生成。

测试结果： 能够看出在常见的请求区间10k左右数据大小，QPS能在18000+。 

传输数据大小 响应时间(毫秒) QPS 50byte 3186 31387 1k 4063 24612 10k 5354 18677 20k 7833 12766 50k 12658 7900

 

6.4 长链接池PooledConfiguration配置详解

/** * 控制池中空闲的对象的最大数量。 默认值是8，若是是负值表示没限制。 */ private int maxIdle = GenericObjectPool.DEFAULT_MAX_IDLE;   /** * whenExhaustedAction若是是WHEN_EXHAUSTED_BLOCK，指定等待的毫秒数。<br /> * 若是maxWait是正数，那么会等待maxWait的毫秒的时间，超时会抛出NoSuchElementException异常 ；<br /> * 若是maxWait为负值，会永久等待。 maxWait的默认值是-1。 */ private long maxWait = GenericObjectPool.DEFAULT_MAX_WAIT;   /** * 若是testOnBorrow被设置，pool会在borrowObject返回对象以前使用PoolableObjectFactory的validateObject来验证这个对象是否有效，要是对象没经过验证，这个对象会被丢弃， * 而后从新选择一个新的对象。 testOnBorrow的默认值是false，可使用GenericObjectPool.DEFAULT_TEST_ON_BORROW;。 * <p>&nbsp;</p> * 注意，对于长期idle的链接，服务端会默认关闭channel此时客户端并不知晓，所以不能使用已经失效的channel，为保证客户端可用，这里暂时使用这个策略每次borrow的时候都test */ private boolean testOnBorrow = true;   /** * 控制池中空闲的对象的最小数量。 默认值是0。 */ private int minIdle = GenericObjectPool.DEFAULT_MIN_IDLE;   /** * 控制池中对象的最大数量。 默认值是8，若是是负值表示没限制。 */ private int maxActive = GenericObjectPool.DEFAULT_MAX_ACTIVE;   /** * 若是testOnReturn被设置，pool会在returnObject的时候经过PoolableObjectFactory的validateObject方法验证对象，若是对象没经过验证，对象会被丢弃，不会被放到池中。 * testOnReturn的默认值是false。 */ private boolean testOnReturn = GenericObjectPool.DEFAULT_TEST_ON_RETURN;   /** * 指定idle对象是否应该使用PoolableObjectFactory的validateObject校验，若是校验失败，这个对象会从对象池中被清除。 * 这个设置仅在timeBetweenEvictionRunsMillis被设置成正值（&gt;0）的时候才会生效。 testWhileIdle的默认值是false。 */ private boolean testWhileIdle = GenericObjectPool.DEFAULT_TEST_WHILE_IDLE;   /** * 指定驱逐线程的休眠时间。若是这个值不是正数（&gt;0），不会有驱逐线程运行。 timeBetweenEvictionRunsMillis的默认值是-1。 */ private long timeBetweenEvictionRunsMillis = GenericObjectPool.DEFAULT_TIME_BETWEEN_EVICTION_RUNS_MILLIS;   /** * 设置驱逐线程每次检测对象的数量。 这个设置仅在timeBetweenEvictionRunsMillis被设置成正值（&gt;0）的时候才会生效。 numTestsPerEvictionRun的默认值是3。 */ private int numTestsPerEvictionRun = GenericObjectPool.DEFAULT_NUM_TESTS_PER_EVICTION_RUN;   /** * 指定最小的空闲驱逐的时间间隔（空闲超过指定的时间的对象，会被清除掉）。 这个设置仅在timeBetweenEvictionRunsMillis被设置成正值（&gt;0）的时候才会生效。 * minEvictableIdleTimeMillis默认值是30分钟。 */ private long minEvictableIdleTimeMillis = GenericObjectPool.DEFAULT_MIN_EVICTABLE_IDLE_TIME_MILLIS;   /** * 与minEvictableIdleTimeMillis相似，也是指定最小的空闲驱逐的时间间隔（空闲超过指定的时间的对象，会被清除掉），不过会参考minIdle的值，只有idle对象的数量超过minIdle的值，对象才会被清除。 * 这个设置仅在timeBetweenEvictionRunsMillis被设置成正值 * （&gt;0）的时候才会生效，而且这个配置能被minEvictableIdleTimeMillis配置取代（minEvictableIdleTimeMillis配置项的优先级更高）。 * softMinEvictableIdleTimeMillis的默认值是-1。 */ private long softMinEvictableIdleTimeMillis = GenericObjectPool.DEFAULT_SOFT_MIN_EVICTABLE_IDLE_TIME_MILLIS;   /** * 设置后进先出的池策略。pool能够被配置成LIFO队列（last-in-first-out）或FIFO队列（first-in-first-out），来指定空闲对象被使用的次序。 lifo的默认值是true。 */ private boolean lifo = GenericObjectPool.DEFAULT_LIFO;   /** * 指定池中对象被消耗完之后的行为，有下面这些选择： WHEN_EXHAUSTED_FAIL 0 WHEN_EXHAUSTED_GROW 2 WHEN_EXHAUSTED_BLOCK 1 * 若是是WHEN_EXHAUSTED_FAIL，当池中对象达到上限之后，继续borrowObject会抛出NoSuchElementException异常。 * 若是是WHEN_EXHAUSTED_GROW，当池中对象达到上限之后，会建立一个新对象，并返回它。 * 若是是WHEN_EXHAUSTED_BLOCK，当池中对象达到上限之后，会一直等待，直到有一个对象可用。这个行为还与maxWait有关 * ，若是maxWait是正数，那么会等待maxWait的毫秒的时间，超时会抛出NoSuchElementException异常；若是maxWait为负值，会永久等待。 * whenExhaustedAction的默认值是WHEN_EXHAUSTED_BLOCK，maxWait的默认值是-1。 */ private byte whenExhaustedAction = GenericObjectPool.WHEN_EXHAUSTED_BLOCK;

6.5 默认值手册

/** * 默认客户端链接超时时间，单位毫秒 */ public static final int DEFAULT_CLIENT_CONN_TIMEOUT = 4000;   /** * 默认客户端调用读超时时间，单位毫秒 */ public static final int DEFAULT_CLIENT_READ_TIMEOUT = 60000;   /** * 默认客户端超时调用检测器启动时间，单位毫秒 */ public static int CLIENT_TIMEOUT_EVICTOR_DELAY_START_TIME = 5000;   /** * 默认客户端超时调用检测器检测间隔，单位毫秒 */ public static int CLIENT_TIMEOUT_EVICTOR_CHECK_INTERVAL = 5000;

6.6 protoc生成原生proto代码方法

1）下载的protobuffer编译客户端： github：https://github.com/google/protobuf/releases 目前经常使用的是2.5.0版本

2）重命名问xxx.proto为本身想生成类名称

3）修改文件中package为本身的包前缀

4）调用命令：protoc –java_out=xxx.proto 

其余生成方法可使用各类IDE或者编辑器（如sublime text）直接生成。