java高性能rpc,企业级rpc,zk调度,负载均衡,泛化调用一体的rpc服务框架

​​一:项目介绍

koalas-RPC个人作品,提供大家交流学习,有意见请私信,欢迎拍砖。客户端采用thrift协议,服务端支持netty和thrift的TThreadedSelectorServer半同步半异步线程模型,支持动态扩容,服务上下线,权重动态,可用性配置,页面流量统计等,QPS统计,TP90,TP99,TP95等丰富可视化数据,持续为个人以及中小型公司提供可靠的RPC框架技术方案。

1:为什么要写这个RPC

市面上常见的RPC框架很多,grpc,motan,dubbo等,但是随着越来越多的元素加入,复杂的架构设计等因素似使得这些框架和spring一样,虽然号称是轻量级,但是用起来却是让我们很蹩脚,大量的配置,繁杂的API设计,其实,我们根本用不上这些东西!!!我也算得上是在很多个互联网企业厮杀过,见过很多很多的内部RPC框架,有些优秀的设计让我非常赞赏,有一天我突然想着,为什么不对这些设计原型进行聚合归类,自己搞一套【轻量级】RPC框架呢,碍于工作原因,一直没有时间倒腾出空,十一期间工作闲暇,说搞就搞吧,落地不易,很多细节性问题,比如tcp中怎么解决大量的wait-time,如何做到thrift和netty的兼容等等大量细节的优化,希望源码对大家对认识RPC框架起到推进的作用。东西越写越多,有各种问题欢迎随时拍砖

2:为什么叫koalas

树袋熊英文翻译,希望考拉RPC给那些不太喜欢动手自己去造轮子的人提供可靠的RPC使用环境

3:技术栈

thrift0.8.0

spring-core-4.2.5,spring-context-4.2.5,spring-beans-4.2.5

log4j,slf4j

org.apache.commons(v2.0+)

io.netty4

fastJson

zookeeper

点评cat(V3.0.0+做数据大盘统计上报等使用,可不配置)

AOP,反射代理等

4:关于技术选型

序列化篇考察了很多个序列化组件,其中包括jdk原生,kryo、hessian、protoStuff,thrift,json等,最终选择了Thrift,原因如下原生JDK序列化反序列化效率堪忧,其序列化内容太过全面kryo和hessian,json相对来说比原生JDK强一些,但是对跨语言支持一般,所以舍弃了,最终想在protoBuf和Thrift协议里面选择一套框架,这俩框架很相通,支持跨语言,需要静态编译等等。但是protoBuf不带RPC服务,本着提供多套服务端模式(thriftrpc,netty)的情况下,最终选择了Thrift协议。

IO线程模型篇原生socket可以模拟出简单的RPC框架,但是对于大规模并发,要求吞吐量的系统来说,也就算得上是一个demo级别的,所以BIO肯定是不考虑了,NIO的模型在序列化技术选型的时候已经说了,Thrift本身支持很多个io线程模型,同步,异步,半同步异步等(SimpleServer,TNonblockingServer,THsHaServer,TThreadedSelectorServer,TThreadPoolServer),其中吞吐量最高的肯定是半同步半异步的IO模TThreadedSelectorServer了,具体原因大家可自行google,这次不做多的阐述,选择好了模型之后,发现thrift简直就是神器一样的存在,再一想,对于服务端来说,IO模型怎么能少得了Netty啊,所以下决心也要支持Netty,但是很遗憾Netty目前没有对Thrift的序列化解析,拆包粘包的处理,但是有protoBuf,和http协议的封装,怎么办,自己在netty上写对thrift的支持呗,虽然工作量大了一些,但是一想netty不就是干这个事儿的嘛--!

服务发现支持集群的RPC框架里面,像dubbo,或者是其他三方框架,对服务发现都进行的封装,那么自研RPC的话,服务发现就要自己来写了,那么简单小巧容易上手的zookeeper肯定是首选了。

5:安装教程

考拉RPC确保精简,轻量的原则,只需要zk服务器进行服务发现(后续版本服务治理可能需要Datasource),对于zookeeper的各个环境安装教程请自行google,不在本安装教程内特意说明如果需要cat的数据大盘功能,想更方便的查看服务的调用情况,需要安装cat服务,至于cat的安装就更简单了,就是war包扔在tomcat里面运行,然后配置一些参数即可,当然你也可以不接入cat,单独的作为RPC框架来使用。CAT接入参考:https://github.com/dianping/cat

二:使用说明

1:前期准以及依赖

maven依赖

<dependency>

<groupId>koalas.rpc</groupId>

<artifactId>com.Koalas.rpc</artifactId>

<version>Koalas-1.0-SNAPSHOT</version>

</dependency>

关于私服的引用问题,记得全局文件不要把全局的依赖都代理掉,因为这么做只能从aliyun的私服上下载项目,由于koalas-rpc中的Cat依赖只在美团点评的私有仓库中存在,这么做会下载依赖失败,所以不要暴力的设置下面的代理做法。

<mirror>

<id>nexus-aliyun</id>

<mirrorOf>*</mirrorOf>

<name>Nexusaliyun</name>

<url>http://maven.aliyun.com/nexus/content/groups/public</url>

</mirror>

正确的做法是将代理去掉,直接按照作者在pom.xml文件中给定的依赖仓库地址就可以了。

首先需要编写自己的thriftidl文件了,这里多说一句,在群里的小伙伴曾经说过idl文件编写不熟悉,有可能出错这里顺带说一嘴,thrift的ldl文件和写java的请求体和service几乎没有任何区别,熟能生巧,上手之后非常简单这里推荐几篇thrift的文章,有兴趣可以看一看https://blog.csdn.net/lk10207160511/article/details/50450541,https://blog.csdn.net/hrn1216/article/details/51306395下面截图为测试的thrift文件

更新于2019年06月10日

如果大家实在不乐意手写idl文件,那么作者给大家提供了一个简单的插件。链接:https://pan.baidu.com/s/1d_Raox39zSdFrMGw--VUsQ提取码:y7yu,下载之后在src/test/java下面写自己的普通java接口对象,然后一键生成thrfit文件和便后之后的文件(前提条件是需要使用者把thrift编译环境设置到path中,否则不能正常运行),使用方式如下:写好了自己的接口文件之后直接运行ThriftFileBuilderTest测试类中方法。

1@Test

2publicvoidtestToOutputstream()throwsException{

3

4StringbaseDir="src/test/java";

5Classclazz=ICommonUserService.class;

6StringoutPutFile=baseDir.concat("/").concat(clazz.getPackage().getName().replaceAll("\\.","/")).concat("/");

7outPutFile=outPutFile.concat(clazz.getSimpleName()).concat("/");

8outPutFile=outPutFile.concat(clazz.getSimpleName()+".thrift");

9

10Filefile=newFile(outPutFile);

11if(file.getParentFile()!=null&&!file.getParentFile().exists()){

12file.getParentFile().mkdirs();

13file.createNewFile();

14}

15

16this.fileBuilder.setSourceDir(baseDir);

17

18FileOutputStreamfileOutputStream=newFileOutputStream(file);

19this.fileBuilder.buildToOutputStream(clazz,fileOutputStream);

20

21excuteThriftCommand(file.getAbsolutePath());

22}

只需要修改clazz的接口就可以了,执行过后在当前包下会生成一个thrift文件和编译过后的class文件,直接使用即可。test0包是作者的测试包名,改成自己实际的包名就可以了。最后说明的是作者还是推荐自己练习写idl文件,熟练过后就可以不依赖这个工具了。

1namespacejavathrift.service

2

3include'WmCreateAccountRequest.thrift'

4include'WmCreateAccountRespone.thrift'

5

6serviceWmCreateAccountService{

7WmCreateAccountRespone.WmCreateAccountResponegetRPC(1:WmCreateAccountRequest.WmCreateAccountRequestwmCreateAccountRequest);

8WmCreateAccountRespone.WmCreateAccountResponekoaloasTest1(1:WmCreateAccountRequest.WmCreateAccountRequestwmCreateAccountRequest);

9WmCreateAccountRespone.WmCreateAccountResponekoaloasTest2(1:WmCreateAccountRequest.WmCreateAccountRequestwmCreateAccountRequest);

10WmCreateAccountRespone.WmCreateAccountResponekoaloasTest3(1:WmCreateAccountRequest.WmCreateAccountRequestwmCreateAccountRequest);

11WmCreateAccountRespone.WmCreateAccountResponekoaloasTest4(1:WmCreateAccountRequest.WmCreateAccountRequestwmCreateAccountRequest);

12WmCreateAccountRespone.WmCreateAccountResponekoaloasTest5(1:WmCreateAccountRequest.WmCreateAccountRequestwmCreateAccountRequest);

13WmCreateAccountRespone.WmCreateAccountResponekoaloasTest6(1:WmCreateAccountRequest.WmCreateAccountRequestwmCreateAccountRequest);

14WmCreateAccountRespone.WmCreateAccountResponekoaloasTest7(1:WmCreateAccountRequest.WmCreateAccountRequestwmCreateAccountRequest);

15WmCreateAccountRespone.WmCreateAccountResponekoaloasTest8(1:WmCreateAccountRequest.WmCreateAccountRequestwmCreateAccountRequest);

16WmCreateAccountRespone.WmCreateAccountResponekoaloasTest9(1:WmCreateAccountRequest.WmCreateAccountRequestwmCreateAccountRequest);

17WmCreateAccountRespone.WmCreateAccountResponekoaloasTest10(1:WmCreateAccountRequest.WmCreateAccountRequestwmCreateAccountRequest);

18WmCreateAccountRespone.WmCreateAccountResponekoaloasTest11(1:WmCreateAccountRequest.WmCreateAccountRequestwmCreateAccountRequest);

19WmCreateAccountRespone.WmCreateAccountResponekoaloasTest12(1:WmCreateAccountRequest.WmCreateAccountRequestwmCreateAccountRequest);

20}

1namespacejavathrift.domain

2/**

3*测试类

4**/

5structWmCreateAccountRequest{

6

71:i32source,

8

92:i32accountType,

10

113:i64partnerId,

12

134:i32partnerType,

14

155:stringpartnerName,

16

176:i32poiFlag,

18}

19namespacejavathrift.domain

20/**

21*测试类

22**/

23structWmCreateAccountRespone{

241:i32code,

252:stringmessage,

26}

编译器需要大家去下载对应的版本windows和linux下不同的编译器,下载地址http://archive.apache.org/dist/thrift/0.8.0/下载0.8.0版本即可,0.8.0版本是很老的版本了,但是相对稳定,后续会把thirft版本升级。如果上面地址下载不下来或者失效,可以上作者的网盘上下载zip包,上面有win版本和mac,linux版本的0.8.0的thrift编译器,链接:https://pan.baidu.com/s/1JpLqVbmokTOe30nU_TznWw提取码:ntye,编译上面三个文件thrift-genjavaWmCreateAccountService.thrift,thrift-genjavaWmCreateAccountRequest.thrift,thrift-genjavaWmCreateAccountRespone.thrift在当前目录下会生成3个java文件这三个文件分别是请求体,返回体,和服务类,就这么简单Ok作为开发者而言,所有的准备工作都结束了。下面就开始进入实际开发~

2:xml配置方式

1.客户端同步调用

首先在你的xml里面配置一下引用

<?xmlversion="1.0"encoding="UTF-8"?>

<beansxmlns="http://www.springframework.org/schema/beans"

xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

xmlns:koalas="http://www.koalas.com/schema/ch"

xsi:schemaLocation="http://www.springframework.org/schema/beans

http://www.springframework.org/schema/beans/spring-beans-4.2.xsd

http://www.koalas.com/schema/ch

http://www.koalas.com/schema/ch.xsd">

<koalas:clientid="wmCreateAccountService1"

serviceInterface="thrift.service.WmCreateAccountService"

zkPath="127.0.0.1:2181"/>

</beans>

首先引用koalas的自定义schema,xmlns:koalas和xsi:schemaLocation,其中serviceInterface为thrift自动生成的java类,zkPath为zk的服务地址,默认是同步调用,接下来就是在java里面的远程调用了。

packagethrift.service;

importorg.apache.thrift.TException;

importorg.springframework.beans.factory.annotation.Autowired;

importorg.springframework.stereotype.Service;

importthrift.domain.WmCreateAccountRequest;

importthrift.domain.WmCreateAccountRespone;

@Service("testService")

publicclassTestService{

@Autowired

WmCreateAccountService.IfacewmCreateAccountService;

publicvoidgetRemoteRpc()throwsTException{

WmCreateAccountRequestrequest=newWmCreateAccountRequest();

//request.setSource(10);

request.setAccountType(1);

request.setPartnerId(1);

request.setPartnerType(1);

request.setPartnerName("你好");

request.setPoiFlag(1);

WmCreateAccountResponerespone=wmCreateAccountService.getRPC(request);

System.out.println(respone);

}

}

就这么简单一个高性能的RPC框架就诞生了。WmCreateAccountService是thrift自动生成的,作为使用者而言不需要做任何事情,只需要在springbean中注入xxx.Iface即可。

2.客户端异步调用

刚刚我们看了客户端的同步调用方式,下面我们一起来看看异步的使用方式,首先在你的xml里面配置一下引用

<?xmlversion="1.0"encoding="UTF-8"?>

<beansxmlns="http://www.springframework.org/schema/beans"

xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

xmlns:koalas="http://www.koalas.com/schema/ch"

xsi:schemaLocation="http://www.springframework.org/schema/beans

http://www.springframework.org/schema/beans/spring-beans-4.2.xsd

http://www.koalas.com/schema/ch

http://www.koalas.com/schema/ch.xsd">

<koalas:clientid="wmCreateAccountService2"

serviceInterface="thrift.service.WmCreateAccountService"

zkPath="127.0.0.1:2181"

async="true"/>

</beans>

和同步的区别async=true,代表异步使用,接下来就是在java里面的异步远程调用了

packagethrift.service;

importclient.async.KoalasAsyncCallBack;

importorg.apache.thrift.TException;

importorg.springframework.beans.factory.annotation.Autowired;

importorg.springframework.stereotype.Service;

importthrift.domain.WmCreateAccountRequest;

importthrift.domain.WmCreateAccountRespone;

importjava.util.concurrent.ExecutionException;

importjava.util.concurrent.Future;

@Service("testService")

publicclassTestService2{

@Autowired

WmCreateAccountService.AsyncIfacewmCreateAccountService;

publicvoidgetRemoteRpc()throwsTException{

KoalasAsyncCallBack<WmCreateAccountRespone,WmCreateAccountService.AsyncClient.getRPC_call>

koalasAsyncCallBack=newKoalasAsyncCallBack<>();

WmCreateAccountRequestrequest=newWmCreateAccountRequest();

request.setAccountType(1);

request.setPartnerId(1);

request.setPartnerType(1);

request.setPartnerName("你好啊");

request.setPoiFlag(1);

wmCreateAccountService.getRPC(request,koalasAsyncCallBack);

Future<WmCreateAccountRespone>future=koalasAsyncCallBack.getFuture();

try{

//togetotherthings

System.out.println(future.get());

}catch(InterruptedExceptione){

e.printStackTrace();

}catch(ExecutionExceptione){

e.printStackTrace();

}

}

}

这次调用getRpc方法不会阻塞等待server同步结果了。而是可以去干一些自己的其他事情,然后在调用future.get()来获得返回resopne,当然future.get()支持最大等待时间的,超时之后会抛出TimeOutException,当然这仅仅是client超时而已不会影响server的执行结果。

3.服务端实现

<?xmlversion="1.0"encoding="UTF-8"?>

<beansxmlns="http://www.springframework.org/schema/beans"

xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

xmlns:context="http://www.springframework.org/schema/context"

xmlns:koalas="http://www.koalas.com/schema/ch"

xsi:schemaLocation="http://www.springframework.org/schema/beans

http://www.springframework.org/schema/beans/spring-beans-4.2.xsd

http://www.springframework.org/schema/context

http://www.springframework.org/schema/context/spring-context-4.2.xsd

http://www.koalas.com/schema/ch

http://www.koalas.com/schema/ch.xsd">

<koalas:serverid="WmCreateAccountService"

serviceInterface="thrift.service.WmCreateAccountService"

serviceImpl="wmCreateAccountServiceImpl"

port="8001"

zkpath="127.0.0.1:2181"/>

</beans>

服务端只需要指定暴露的端口,zk服务地址和服务端实现即可。

@Service

publicclassWmCreateAccountServiceImplimplementsWmCreateAccountService.Iface{

@Override

publicWmCreateAccountResponegetRPC(WmCreateAccountRequestwmCreateAccountRequest)throwsTException{

WmCreateAccountResponewmCreateAccountRespone=newWmCreateAccountRespone();

wmCreateAccountRespone.setCode(1);

wmCreateAccountRespone.setMessage("你好");

if(newRandom().nextInt(5)>100){

thrownewRuntimeException("测试错误");

}

System.out.println("getRPCstart...."+wmCreateAccountRequest+"------"+atomicInteger.incrementAndGet());

returnwmCreateAccountRespone;

}

}

只需要实现xxxx.Iface即可

3:注解配置方式

有的小伙伴会觉得配置xml有点麻烦,koalas-rpc也提供了纯注解的使用方式

1.客户端调用

xml中的配置

<?xmlversion="1.0"encoding="UTF-8"?>

<beansxmlns="http://www.springframework.org/schema/beans"

xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

xmlns:koalas="http://www.koalas.com/schema/ch"

xsi:schemaLocation="http://www.springframework.org/schema/beans

http://www.springframework.org/schema/beans/spring-beans-4.2.xsd

http://www.koalas.com/schema/ch

http://www.koalas.com/schema/ch.xsd">

<koalas:annotationpackage="thrift.annotation.client.impl"/>

</beans>

一个扫描标签就行了,如果你在springbean里想通过调用rpc远程服务,那么扫描一下就行了

java中使用

@Service("testServiceSync")

publicclassTestServiceSync{

@KoalasClient(zkPath="127.0.0.1:2181",readTimeout=5000*1000)

WmCreateAccountService.IfacewmCreateAccountService;

publicvoidgetRemoteRpc()throwsTException{

WmCreateAccountRequestrequest=newWmCreateAccountRequest();

//request.setSource(10);

request.setAccountType(1);

request.setPartnerId(1);

request.setPartnerType(1);

request.setPartnerName("你好啊-我是注解实现的");

request.setPoiFlag(1);

WmCreateAccountResponerespone=wmCreateAccountService.getRPC(request);

System.out.println(respone);

}

}

只需要在你想远程调用的类上加一个@KoalasClient注解就可以了,远程调用就这么简单,当然异步使用方式也类似

@Service("testServiceAsync")

publicclassTestServiceAsync{

@KoalasClient(zkPath="127.0.0.1:2181",readTimeout=5000*1000)

WmCreateAccountService.AsyncIfacewmCreateAccountService;

publicvoidgetRemoteRpc()throwsTException{

KoalasAsyncCallBack<WmCreateAccountRespone,WmCreateAccountService.AsyncClient.getRPC_call>koalasAsyncCallBack=newKoalasAsyncCallBack<>();

WmCreateAccountRequestrequest=newWmCreateAccountRequest();

//request.setSource(10);

request.setAccountType(1);

request.setPartnerId(1);

request.setPartnerType(1);

request.setPartnerName("你好啊-我是注解实现的");

request.setPoiFlag(1);

wmCreateAccountService.getRPC(request,koalasAsyncCallBack);

Future<WmCreateAccountRespone>future=koalasAsyncCallBack.getFuture();

try{

System.out.println(future.get());

}catch(InterruptedExceptione){

e.printStackTrace();

}catch(ExecutionExceptione){

e.printStackTrace();

}

}

}

注意和同步调用不同的是自定义注解注入的接口是xxxx.AsyncIface,同步是xxxx.Iface。KoalasAsyncCallBack回调使用方式和上面的xml一样。有一点需要说明

<koalas:annotationpackage="thrift.annotation.client.impl"/>

如果package属性设置为空,那么所有的@KoalasClient都会生效,也就是说所有在springbean中的自定义注解@KoalasClient都会自动注入。这里说另外一种用法

privateWmCreateAccountService.IfacewmCreateAccountService;

@KoalasClient(zkPath="127.0.0.1:2181",readTimeout=5000*1000)

publicvoidsetWmCreateAccountService(WmCreateAccountService.IfacewmCreateAccountService){

this.wmCreateAccountService=wmCreateAccountService;

}

直接注入方法的方式也是可以的。

2.服务端实现

xml中的配置

<?xmlversion="1.0"encoding="UTF-8"?>

<beansxmlns="http://www.springframework.org/schema/beans"

xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

xmlns:koalas="http://www.koalas.com/schema/ch"

xsi:schemaLocation="http://www.springframework.org/schema/beans

http://www.springframework.org/schema/beans/spring-beans-4.2.xsd

http://www.koalas.com/schema/ch

http://www.koalas.com/schema/ch.xsd">

<koalas:annotationpackage="thrift.annotation.server.impl"/>

</beans>

配置和client中一样只需要配置一个自定义标签即可,java中的使用方式如下:

packagethrift.annotation.server.impl;

importannotation.KoalasServer;

importorg.apache.thrift.TException;

importthrift.domain.WmCreateAccountRequest;

importthrift.domain.WmCreateAccountRespone;

importthrift.service.WmCreateAccountService;

importjava.util.Random;

importjava.util.concurrent.atomic.AtomicInteger;

@KoalasServer(port=8801,zkpath="127.0.0.1:2181")

publicclassWmCreateAccountServiceNettyImplimplementsWmCreateAccountService.Iface{

privateAtomicIntegeratomicInteger=newAtomicInteger(0);

@Override

publicWmCreateAccountResponegetRPC(WmCreateAccountRequestwmCreateAccountRequest)throwsTException{

WmCreateAccountResponewmCreateAccountRespone=newWmCreateAccountRespone();

wmCreateAccountRespone.setCode(1);

wmCreateAccountRespone.setMessage("你好啊");

if(newRandom().nextInt(5)>100){

try{

Thread.sleep(5000);

}catch(InterruptedExceptione){

e.printStackTrace();

}

}

System.out.println("getRPCstart...."+wmCreateAccountRequest+"------"+atomicInteger.incrementAndGet());

returnwmCreateAccountRespone;

}

}

这样服务实现就会主从注册到zookeeper中提供给client端使用了。值得说明的是被扫描到并且类上有@KoalasServer的类会被加载到spring上下文中,可以当成一个普通的springbean来处理,还有一点如果你不指定package,配置成如下情况

<koalas:annotationpackage=""/>

这样配置会以spring的bean为基础实现,那么使用方式需要改成

packagethrift.annotation.server.impl;

importannotation.KoalasServer;

importorg.apache.thrift.TException;

importthrift.domain.WmCreateAccountRequest;

importthrift.domain.WmCreateAccountRespone;

importthrift.service.WmCreateAccountService;

importjava.util.Random;

importjava.util.concurrent.atomic.AtomicInteger;

@KoalasServer(port=8801,zkpath="127.0.0.1:2181")

@Service

publicclassWmCreateAccountServiceNettyImplimplementsWmCreateAccountService.Iface{

privateAtomicIntegeratomicInteger=newAtomicInteger(0);

@Override

publicWmCreateAccountResponegetRPC(WmCreateAccountRequestwmCreateAccountRequest)throwsTException{

WmCreateAccountResponewmCreateAccountRespone=newWmCreateAccountRespone();

wmCreateAccountRespone.setCode(1);

wmCreateAccountRespone.setMessage("你好啊");

if(newRandom().nextInt(5)>100){

try{

Thread.sleep(5000);

}catch(InterruptedExceptione){

e.printStackTrace();

}

}

System.out.println("getRPCstart...."+wmCreateAccountRequest+"------"+atomicInteger.incrementAndGet());

returnwmCreateAccountRespone;

}

}

就这么简单即可。

3.泛化调用

为什么需要泛化调用?1:有一个通用压测平台,想去压测不同的server。那么现在就有一个问题了,不可能让压测平台服务端去依赖所有的下游服务,这样依赖会很繁杂,这时候如果说只配置serviceName,request模型和request请求json就可以进行远程调用,那么将大大的减少头疼的依赖。2:假设php同事对java代码不熟悉,不可能让他们去依赖spring,一共一套简单的api来使用是很有必要的。3:上游服务不想依赖下游服务的数据模型。

对于泛化调用来说,dubbo已经提供,soft-rpc也有提供。当然koalas-rpc也不会例外,并且支持xml,注解和javaapi的使用方式。下面几个例子来说明一下使用方式。更多demo去源码中查看,作者已经写好,开箱即用。

xml使用方式

<koalas:clientid="wmCreateAccountService3"

serviceInterface="thrift.service.WmCreateAccountService"

zkPath="127.0.0.1:2181"

generic="true"

readTimeout="50000000"/>

@Autowired

@Qualifier("wmCreateAccountService3")

GenericService.IfacewmGenericService;

publicvoidgetGenericRpc()throwsTException{

GenericRequestrequest=newGenericRequest();

request.setMethodName("getRPC");

request.setClassType(newArrayList<String>(){{

add("thrift.domain.WmCreateAccountRequest");

}});

request.setRequestObj(newArrayList<String>(){{

add("{\"accountType\":1,\"partnerId\":1,\"partnerName\":\"你好\",\"partnerType\":1,\"poiFlag\":1,\"source\":0}");

}});

Stringstr=wmGenericService.invoke(request);

System.out.println(str);

}

简单说明一下,GenericService.Iface是通用服务,有三个参数,第一个是方法名称,第二个是请求体类型集合,第三个是请求体内容。直接调用即可,返回值是server端的json类型,使用json工具为阿里巴巴的Fast-json

注解使用方式

@KoalasClient(zkPath="127.0.0.1:2181",readTimeout=5000*1000,genericService="thrift.service.WmCreateAccountService")

GenericService.IfacegenericService;

publicvoidgetGenericRemoteRpc()throwsTException{

GenericRequestrequest=newGenericRequest();

request.setMethodName("getRPC");

request.setClassType(newArrayList<String>(){{

add("thrift.domain.WmCreateAccountRequest");

}});

request.setRequestObj(newArrayList<String>(){{

add("{\"accountType\":1,\"partnerId\":1,\"partnerName\":\"你好\",\"partnerType\":1,\"poiFlag\":1,\"setAccountType\":true,\"setPartnerId\":true,\"setPartnerName\":true,\"setPartnerType\":true,\"setPoiFlag\":true,\"setSource\":false,\"source\":0}");

}});

Stringstr=genericService.invoke(request);

System.out.println(str);

}

唯一区别的是注解要指定genericService,当genericService不为空时,默认开启泛化调用当然,javaapi方式也是支持的。

KoalasClientProxykoalasClientProxy=newKoalasClientProxy();

koalasClientProxy.setServiceInterface("thrift.service.WmCreateAccountService");

koalasClientProxy.setZkPath("127.0.0.1:2181");

koalasClientProxy.setGeneric(true);

koalasClientProxy.setReadTimeout(50000000);

koalasClientProxy.afterPropertiesSet();

GenericService.IfacegenericService=(GenericService.Iface)koalasClientProxy.getObject();

GenericRequestrequest=newGenericRequest();

request.setMethodName("getRPC");

request.setClassType(newArrayList<String>(){{

add("thrift.domain.WmCreateAccountRequest");

}});

request.setRequestObj(newArrayList<String>(){{

add("{\"accountType\":1,\"partnerId\":1,\"partnerName\":\"你好\",\"partnerType\":1,\"poiFlag\":1,\"setAccountType\":true,\"setPartnerId\":true,\"setPartnerName\":true,\"setPartnerType\":true,\"setPoiFlag\":true,\"setSource\":false,\"source\":0}");

}});

Stringstr=genericService.invoke(request);

System.out.println(str);

koalasClientProxy.destroy();

特别注意的是KoalasClientProxy对象非常非常重,一定要在服务关闭的时候执行koalasClientProxy.destroy();方法,并且需要带应用程序中缓存该对象,千万不要每次使用都要创建,这样会极大的浪费资源,每个服务对应一个KoalasClientProxy,同步和异步也是不同的对象,这些使用者需要注意。

4.原生调用支持

koalas-rpc在原生基础上封装了自定义协议和特定的传输类型,看过源码的朋友一定觉得处理非常非常麻烦,但是在自定义协议的过程中koalas-rpc也同时支持原生的thrift请求,可以在本地做测试等等。请求调用demo:

packagexml.client;

importorg.apache.thrift.TException;

importorg.apache.thrift.protocol.TBinaryProtocol;

importorg.apache.thrift.protocol.TProtocol;

importorg.apache.thrift.transport.TFramedTransport;

importorg.apache.thrift.transport.TSocket;

importorg.apache.thrift.transport.TTransport;

importthrift.domain.WmCreateAccountRequest;

importthrift.domain.WmCreateAccountRespone;

importthrift.service.WmCreateAccountService;

publicclassThriftNative{

publicstaticfinalStringSERVER_IP="localhost";

publicstaticfinalintSERVER_PORT=8001;

publicstaticfinalintTIMEOUT=3000000;

publicstaticvoidmain(String[]args)throwsTException{

TTransporttransport=newTFramedTransport(newTSocket(SERVER_IP,SERVER_PORT,TIMEOUT));

TProtocolprotocol=newTBinaryProtocol(transport);

WmCreateAccountService.Clientclient=newWmCreateAccountService.Client(protocol);

transport.open();

WmCreateAccountRequestrequest=newWmCreateAccountRequest();

//request.setSource(10);

request.setAccountType(1);

request.setPartnerId(1);

request.setPartnerType(1);

request.setPartnerName("你好啊-我是ThriftNative实现的服务端getRemoteRpc");

request.setPoiFlag(1);

WmCreateAccountResponerespone=client.getRPC(request);

System.out.println(respone);

}

}

三:参数配置文档

1:客户端

参数名说明是否必须

serviceInterfacethrift生成的接口类Y

zkPathzk的服务地址,集群中间逗号分隔Y

serverIpPorts不实用zk发现直接连接服务器server,格式ip:端口#权重。多个逗号分隔N

async是否异步N,默认false同步

generic是否泛化调用(xml配置中使用)N,默认false

genericService泛化调用的serviceName(注解配置中使用)使用方法参照代码中demoN,默认false

cat是否开启CAT数据大盘,需要配置CAT服务,即可查看详细调用情况)N,默认false

connTimeout连接超时N,默认3000ms

readTimeout读取超时N,默认5000ms,按照服务端指定时间适当调整

localMockServiceImpl本地测试的实现N

retryRequest是否错误重试N,默认true

retryTimes重试次数N,默认3次

maxTotalTCP长连接池,参照ApachePool参数100

maxIdleTCP长连接池,参照ApachePool参数50

minIdleTCP长连接池,参照ApachePool参数10

lifoTCP长连接池,参照ApachePool参数true

fairnessTCP长连接池,参照ApachePool参数false

maxWaitMillisTCP长连接池,参照ApachePool参数30*1000

timeBetweenEvictionRunsMillisTCP长连接池,参照ApachePool参数3*60*1000

minEvictableIdleTimeMillisTCP长连接池,参照ApachePool参数5*60*1000

softMinEvictableIdleTimeMillisTCP长连接池,参照ApachePool参数10*60*1000

numTestsPerEvictionRunTCP长连接池,参照ApachePool参数20

testOnCreateTCP长连接池,参照ApachePool参数false

testOnBorrowTCP长连接池,参照ApachePool参数false

testOnReturnTCP长连接池,参照ApachePool参数false

testWhileIdleTCP长连接池,参照ApachePool参数true

iLoadBalancer负载略侧,默认随机N

env环境N,默认dev

removeAbandonedOnBorrowTCP长连接池,参照ApachePool参数true

removeAbandonedOnMaintenanceTCP长连接池,参照ApachePool参数true

removeAbandonedTimeoutTCP长连接池,参照ApachePool参数30000ms

maxLength_允许发送最大字节数N,10*1024*1024

coresselecter核心数量N,默认当前cpu数量

asyncSelectorThreadCount异步请求时线程数量N,默认当前CPU核心数量*2

privateKey私钥N

publicKey公钥N

2:服务端

参数说明是否必须

serviceImpl服务端实现Y

serviceInterfacethrift自动生成的类Y

port暴露的服务端口Y

zkpath服务端的zk路径Y

cat(是否开启CAT数据大盘,需要配置CAT服务,即可查看详细调用情况)N,默认false

bossThreadCount处理连接线程N,当前CPU核心数

workThreadCount读取线程N,当前CPU核心数*2

koalasThreadCount业务线程数256

maxLength最大接收字节数Integer.MAX_VALUE

env环境N,dev

weight权重N,10

serverType采用哪些服务端,可以选NETTY和THRIFT,默认NETTYN

workQueue当server超载时,可以容纳等待任务的队列长度0

privateKey私钥N

publicKey公钥N

3:客户端服务端RSA双向加密

源码中utils.KoalasRsaUtil的main方法已经为大家写好生成私钥和公钥的代码,执行即可,下面为核心源码展示

publicstaticStringsign(byte[]data,StringprivateKey)throwsException{

byte[]keyBytes=Base64.decodeBase64(privateKey.getBytes("UTF-8"));

PKCS8EncodedKeySpecpkcs8KeySpec=newPKCS8EncodedKeySpec(keyBytes);

KeyFactorykeyFactory=KeyFactory.getInstance(KEY_ALGORITHM);

PrivateKeyprivateK=keyFactory.generatePrivate(pkcs8KeySpec);

Signaturesignature=Signature.getInstance(SIGNATURE_ALGORITHM);

signature.initSign(privateK);

signature.update(data);

returnnewString(Base64.encodeBase64(signature.sign()),"UTF-8");

}

publicstaticbooleanverify(byte[]data,StringpublicKey,Stringsign)

throwsException{

byte[]keyBytes=Base64.decodeBase64(publicKey.getBytes("UTF-8"));

X509EncodedKeySpeckeySpec=newX509EncodedKeySpec(keyBytes);

KeyFactorykeyFactory=KeyFactory.getInstance(KEY_ALGORITHM);

PublicKeypublicK=keyFactory.generatePublic(keySpec);

Signaturesignature=Signature.getInstance(SIGNATURE_ALGORITHM);

signature.initVerify(publicK);

signature.update(data);

returnsignature.verify(Base64.decodeBase64(sign.getBytes("UTF-8")));

}

执行main方法之后,会得到4个长长的字符串

下面四个字符串为koalas-rpc中客户端和服务端使用的rsa非对称秘钥,复制使用即可

MIICdQIBADANBgkqhkiG9w0BAQEFAASCAl8wggJbAgEAAoGBAIPQIc8/+wl5hTDT8fT4rCEA//pwSqdX8djur+UDwR/qg5iW3xBHUuxTGXRko/3SXYKJLugRmT2gV4ZggSHLpToSFYJZwATIbVD2p3oqZx4ZC5g3mZdTCScHbTb4CITFPacJCKads75Plrk8ryW7wP9dWlSmrF8f3CzReKUTjf5dAgMBAAECgYBRigXwK9cCNG8lFmc9sDriq7it1psHzApqtLSQifME6FCBqwrQCh8M3BcJ/lvH30NDRdODcaeHDNI36SjYnB5X25mMG95OEgLqPm7T8oB3DBY/BhJbAY43FbZSU3Lb+El5zknpTtH0M8DTlul1EmLbe+TJVL/x/SkpDx/HSS3GAQJBALtSSBeskQ4P+Pn5M4F2+GZJmFDxaOQHIuy/RdfckxV1aEMN425ieSrinSCXyBC8uTN0zF1NlJsfWLAUhtfSQ90CQQC0I+mEXsxWtTDT+fd3bDgiJtfOwPpyNT4HSObdq+aAqO44NL7fqD2plNZ3vBULfDbdbnTlvKJJnPUdt457WjyBAkAiM63SFMIPbT8qdSPAWbaVBo73CHz8VYk87NeVyEJawqscwyZpezVgbSv/TXdMBwlRqdu+lXGyuRB6ZeUQ9uVJAkAscjfpqyIruqUDiEdgtdjbxE22+7JPf4eAcKJVy1YiJIwyXgFCWdZtAwYvoL5oiQtYcypwjKxWEV4BKQsEsG0BAkBmlDi0wSPA2x7YjudQNWv+H51CsYDWMjOQ7AzUYABfkWVnbeYS/3uf7W56AHl3Rmdo7zUTBJFCyM/Rt28yZVLj

MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQDAAxbccTLuu12V2Le1mI5b+0kZMiQwN/WTSv8d2y0J/wVl+yMWgjZi4c8/kAs8pACEiFQ8hUUovmoAwceKEd5h3ISSV5lEPyBt+68DzinOrSGv7bZhGm5bwkRG7MMpSgAVSJj2lWTkf63fp2e/FwHs3WM64sSlbdlUN/57YtUC6QIDAQAB

MIICdwIBADANBgkqhkiG9w0BAQEFAASCAmEwggJdAgEAAoGBAMADFtxxMu67XZXYt7WYjlv7SRkyJDA39ZNK/x3bLQn/BWX7IxaCNmLhzz+QCzykAISIVDyFRSi+agDBx4oR3mHchJJXmUQ/IG37rwPOKc6tIa/ttmEablvCREbswylKABVImPaVZOR/rd+nZ78XAezdYzrixKVt2VQ3/nti1QLpAgMBAAECgYEApwwI/4+b+AYZzRvV967Zazyaw8jTov+MLrC4cokUDfZIBAkQ5awzFKPPYkU3AXLM4ICaiGyJVoESR8ZOitgw1wB6tbI2DhP4FD5dqJkIOdUNujo+gAda3kfeCjAgWbtUL3Zhj7Ff+xFvSDDxUYKGG4fZwge3CFwyQ2vjxhPTXGECQQDpAkS6AW17LvWAiiu2924MEicJQW/s3w+chjuQ3VaauzotAHoSMi8VjBSlINbKxpklthKB4vubfA6AtTHae3hPAkEA0vVBKk9Qz8TkraN3QcILJwHjcjqP8+51n1jimSpZeZQL4BJxStdqqMP2nUzAVnh4ncEoFZ/3QA0sSwcdPtDLRwJBAIDpMmC+HXYDWuvMhbbqWUXwXQxv2Z5xIk/0q8vPyPQ+FUeEdgTPIuGG6H0bF/qDuYL1onOdwpoZHmTy2iwIF10CQBiVNdvNVFhx1EgbtWj3SL9p6+xCwMWnMxO3kuhQVA7j3qJk48jZ43b5JwLbj8pDzaJsgNRMSM6w+klf8duBDz8CQBMIMmhU84An2nv/CPNPArCC8BN8YhY1AH685zgRQBLv5untRhfZ+hJtqjSzTJlY7JHybMzc6wt2FZXrhvuopO4=

MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQCD0CHPP/sJeYUw0/H0+KwhAP/6cEqnV/HY7q/lA8Ef6oOYlt8QR1LsUxl0ZKP90l2CiS7oEZk9oFeGYIEhy6U6EhWCWcAEyG1Q9qd6KmceGQuYN5mXUwknB202+AiExT2nCQimnbO+T5a5PK8lu8D/XVpUpqxfH9ws0XilE43+XQIDAQAB

上面四个字符串为koalas-rpc中客户端和服务端使用的rsa非对称秘钥,复制使用即可

得到上面的四个长长的字符串,可以由server端给client端提供。其中字符串1,字符串2分别对应client的privateKey,和publicKey,字符串3和字符串4分别对应server端的privateKey,和publicKey,提供rsa双向加密的初衷是为了将非常重要的项目保护起来,不允许其他项目随意调用,但是RSA双向加密会对性能有所影响。当RSA验证失败的时候,client会抛RsaException。RSA对称加密适合给三方系统进行调用,对称加密会影响传输性能。

实际性能压测

8C16Gmac开发本,单机10000次请求耗时截图

10w次请求,大约耗时12s,平均qps在8000左右,在集群环境下会有不错的性能表现

数据大盘展示

开启数据大盘,需要设置客户端或者服务端的cat参数为true,默认为false。koalas2.0已经接入了cat服务,cat服务支持qps统计,可用率,tp90line,tp99line,丰富自定义监控报警等,接入效果图丰富的可视参数,流量统计,日,周,月报表展示等。

链路跟踪

对RPC服务来说,系统间的调用和排查异常接口,确定耗时代码是非常重要的,只要接入了cat,koalsa-rpc天然的支持链路跟踪,一切尽在眼前!

代码下载后如何测试

作者在src/test/java和resource下面有已经写好了的丰富的xml配置和注解配置,下载后直接运行测试即可,注意测试的时候需要安装zookeeper服务,如果不想通过zk做服务发现,那么客户端可以进行直连,指定的server列表,逗号分隔,#分隔权重,格式,192.168.3.253:6666#10,192.168.3.253:6667#10详情见参数配置列表,但是这种办法作者是不推荐的,在生产环境下没有心跳和动态上下线功能。

CAT服务按需配置,不需要数据大盘不需要配置,不会影响RPC功能,CAT接入参考:https://github.com/dianping/cat

开源协议:

ApacheLicenseVersion2.0seehttp://www.apache.org/licenses/LICENSE-2.0.html