基于Protobuf的分布式高性能RPC框架——Navi-Pbrpc
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各字段含义
- id请求的id。目前未使用。建议设置为0。
- flags本次请求的一些标志符。目前框架用于传输errorCode。
- log-id。本次请求的日志id。Navi-rpc服务端用该id定位一个唯一的客户端请求。
- provider标识调用方的表示。
- magic-number特殊标识，用于标识一个包的完整性。目前未使用。
- method-id是RPC方法的序列号。根据proto文件中定义的service顺序，从注册进入的起始值开始依次递增。
- 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> </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被设置成正值（>0）的时候才会生效。 testWhileIdle的默认值是false。 */ private boolean testWhileIdle = GenericObjectPool.DEFAULT_TEST_WHILE_IDLE; /** * 指定驱逐线程的休眠时间。如果这个值不是正数（>0），不会有驱逐线程运行。 timeBetweenEvictionRunsMillis的默认值是-1。 */ private long timeBetweenEvictionRunsMillis = GenericObjectPool.DEFAULT_TIME_BETWEEN_EVICTION_RUNS_MILLIS; /** * 设置驱逐线程每次检测对象的数量。 这个设置仅在timeBetweenEvictionRunsMillis被设置成正值（>0）的时候才会生效。 numTestsPerEvictionRun的默认值是3。 */ private int numTestsPerEvictionRun = GenericObjectPool.DEFAULT_NUM_TESTS_PER_EVICTION_RUN; /** * 指定最小的空闲驱逐的时间间隔（空闲超过指定的时间的对象，会被清除掉）。 这个设置仅在timeBetweenEvictionRunsMillis被设置成正值（>0）的时候才会生效。 * minEvictableIdleTimeMillis默认值是30分钟。 */ private long minEvictableIdleTimeMillis = GenericObjectPool.DEFAULT_MIN_EVICTABLE_IDLE_TIME_MILLIS; /** * 与minEvictableIdleTimeMillis类似，也是指定最小的空闲驱逐的时间间隔（空闲超过指定的时间的对象，会被清除掉），不过会参考minIdle的值，只有idle对象的数量超过minIdle的值，对象才会被清除。 * 这个设置仅在timeBetweenEvictionRunsMillis被设置成正值 * （>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）直接生成。