Java 从零开始手写 RPC—Reflect 反射实现通用调用之服务端

前面我们的例子是一个固定的出参和入参，固定的方法实现。

本节将实现通用的调用，让框架具有更广泛的实用性。

基本思路

所有的方法调用，基于反射进行相关处理实现。

服务端

核心类

• RpcServer

调整如下:

1. serverBootstrap.group(workerGroup,bossGroup)
2. .channel(NioServerSocketChannel.class)
3. //打印日志
4. .handler(newLoggingHandler(LogLevel.INFO))
5. .childHandler(newChannelInitializer(){
6. @Override
7. protectedvoidinitChannel(Channelch)throwsException{
8. ch.pipeline()
9. //解码bytes=>resp
10. .addLast(newObjectDecoder(Integer.MAX_VALUE,ClassResolvers.cacheDisabled(null)))
11. //request=>bytes
12. .addLast(newObjectEncoder())
13. .addLast(newRpcServerHandler());
14. }
15. })
16. //这个参数影响的是还没有被accept取出的连接
17. .option(ChannelOption.SO_BACKLOG,128)
18. //这个参数只是过一段时间内客户端没有响应，服务端会发送一个ack包，以判断客户端是否还活着。
19. .childOption(ChannelOption.SO_KEEPALIVE,true);

其中 ObjectDecoder 和 ObjectEncoder 都是 netty 内置的实现。

RpcServerHandler

1. packagecom.github.houbb.rpc.server.handler;
4. importcom.github.houbb.log.integration.core.Log;
5. importcom.github.houbb.log.integration.core.LogFactory;
6. importcom.github.houbb.rpc.common.rpc.domain.RpcRequest;
7. importcom.github.houbb.rpc.common.rpc.domain.impl.DefaultRpcResponse;
8. importcom.github.houbb.rpc.server.service.impl.DefaultServiceFactory;
9. importio.netty.channel.ChannelHandlerContext;
10. importio.netty.channel.SimpleChannelInboundHandler;
13. /**
14. *@authorbinbin.hou
15. *@since0.0.1
16. */
17. publicclassRpcServerHandlerextendsSimpleChannelInboundHandler{
20. privatestaticfinalLoglog=LogFactory.getLog(RpcServerHandler.class);
23. @Override
24. publicvoidchannelActive(ChannelHandlerContextctx)throwsException{
25. finalStringid=ctx.channel().id().asLongText();
26. log.info("[Server]channel{}connected"+id);
27. }
30. @Override
31. protectedvoidchannelRead0(ChannelHandlerContextctx,Objectmsg)throwsException{
32. finalStringid=ctx.channel().id().asLongText();
33. log.info("[Server]channelreadstart:{}",id);
36. //接受客户端请求
37. RpcRequestrpcRequest=(RpcRequest)msg;
38. log.info("[Server]receivechannel{}request:{}",id,rpcRequest);
41. //回写到client端
42. DefaultRpcResponserpcResponse=handleRpcRequest(rpcRequest);
43. ctx.writeAndFlush(rpcResponse);
44. log.info("[Server]channel{}response{}",id,rpcResponse);
45. }
48. @Override
49. publicvoidexceptionCaught(ChannelHandlerContextctx,Throwablecause)throwsException{
50. cause.printStackTrace();
51. ctx.close();
52. }
55. /**
56. *处理请求信息
57. *@paramrpcRequest请求信息
58. *@return结果信息
59. *@since0.0.6
60. */
61. privateDefaultRpcResponsehandleRpcRequest(finalRpcRequestrpcRequest){
62. DefaultRpcResponserpcResponse=newDefaultRpcResponse();
63. rpcResponse.seqId(rpcRequest.seqId());
66. try{
67. //获取对应的service实现类
68. //rpcRequest=>invocationRequest
69. //执行invoke
70. Objectresult=DefaultServiceFactory.getInstance()
71. .invoke(rpcRequest.serviceId(),
72. rpcRequest.methodName(),
73. rpcRequest.paramTypeNames(),
74. rpcRequest.paramValues());
75. rpcResponse.result(result);
76. }catch(Exceptione){
77. rpcResponse.error(e);
78. log.error("[Server]executemeetexforrequest",rpcRequest,e);
79. }
82. //构建结果值
83. returnrpcResponse;
84. }
87. }

和以前类似，不过 handleRpcRequest 要稍微麻烦一点。

这里需要根据发射，调用对应的方法。

pojo

其中使用的出参、入参实现如下：

RpcRequest

1. packagecom.github.houbb.rpc.common.rpc.domain;
4. importjava.util.List;
7. /**
8. *序列化相关处理
9. *（1）调用创建时间-createTime
10. *（2）调用方式callType
11. *（3）超时时间timeOut
12. *
13. *额外信息：
14. *（1）上下文信息
15. *
16. *@authorbinbin.hou
17. *@since0.0.6
18. */
19. publicinterfaceRpcRequestextendsBaseRpc{
22. /**
23. *创建时间
24. *@return创建时间
25. *@since0.0.6
26. */
27. longcreateTime();
30. /**
31. *服务唯一标识
32. *@return服务唯一标识
33. *@since0.0.6
34. */
35. StringserviceId();
38. /**
39. *方法名称
40. *@return方法名称
41. *@since0.0.6
42. */
43. StringmethodName();
46. /**
47. *方法类型名称列表
48. *@return名称列表
49. *@since0.0.6
50. */
51. ListparamTypeNames();
54. //调用参数信息列表
57. /**
58. *调用参数值
59. *@return参数值数组
60. *@since0.0.6
61. */
62. Object[]paramValues();
65. }

RpcResponse

1. packagecom.github.houbb.rpc.common.rpc.domain;
4. /**
5. *序列化相关处理
6. *@authorbinbin.hou
7. *@since0.0.6
8. */
9. publicinterfaceRpcResponseextendsBaseRpc{
12. /**
13. *异常信息
14. *@return异常信息
15. *@since0.0.6
16. */
17. Throwableerror();
20. /**
21. *请求结果
22. *@return请求结果
23. *@since0.0.6
24. */
25. Objectresult();
28. }

BaseRpc

1. packagecom.github.houbb.rpc.common.rpc.domain;
4. importjava.io.Serializable;
7. /**
8. *序列化相关处理
9. *@authorbinbin.hou
10. *@since0.0.6
11. */
12. publicinterfaceBaseRpcextendsSerializable{
15. /**
16. *获取唯一标识号
17. *（1）用来唯一标识一次调用，便于获取该调用对应的响应信息。
18. *@return唯一标识号
19. */
20. StringseqId();
23. /**
24. *设置唯一标识号
25. *@paramtraceId唯一标识号
26. *@returnthis
27. */
28. BaseRpcseqId(finalStringtraceId);
31. }

ServiceFactory-服务工厂

为了便于对所有的 service 实现类统一管理，这里定义 service 工厂类。

ServiceFactory

1. packagecom.github.houbb.rpc.server.service;
4. importcom.github.houbb.rpc.server.config.service.ServiceConfig;
5. importcom.github.houbb.rpc.server.registry.ServiceRegistry;
8. importjava.util.List;
11. /**
12. *服务方法类仓库管理类-接口
13. *
14. *
15. *（1）对外暴露的方法，应该尽可能的少。
16. *（2）对于外部的调用，后期比如telnet治理，可以使用比如有哪些服务列表？
17. *单个服务有哪些方法名称？
18. *
19. *等等基础信息的查询，本期暂时全部隐藏掉。
20. *
21. *（3）前期尽可能的少暴露方法。
22. *@authorbinbin.hou
23. *@since0.0.6
24. *@seeServiceRegistry服务注册，将服务信息放在这个类中，进行统一的管理。
25. *@seeServiceMethod方法信息
26. */
27. publicinterfaceServiceFactory{
30. /**
31. *注册服务列表信息
32. *@paramserviceConfigList服务配置列表
33. *@returnthis
34. *@since0.0.6
35. */
36. ServiceFactoryregisterServices(finalListserviceConfigList);
39. /**
40. *直接反射调用
41. *（1）此处对于方法反射，为了提升性能，所有的class.getFullName()进行拼接然后放进key中。
42. *
43. *@paramserviceId服务名称
44. *@parammethodName方法名称
45. *@paramparamTypeNames参数类型名称列表
46. *@paramparamValues参数值
47. *@return方法调用返回值
48. *@since0.0.6
49. */
50. Objectinvoke(finalStringserviceId,finalStringmethodName,
51. ListparamTypeNames,finalObject[]paramValues);
54. }

DefaultServiceFactory

作为默认实现，如下：

1. packagecom.github.houbb.rpc.server.service.impl;
4. importcom.github.houbb.heaven.constant.PunctuationConst;
5. importcom.github.houbb.heaven.util.common.ArgUtil;
6. importcom.github.houbb.heaven.util.lang.reflect.ReflectMethodUtil;
7. importcom.github.houbb.heaven.util.util.CollectionUtil;
8. importcom.github.houbb.rpc.common.exception.RpcRuntimeException;
9. importcom.github.houbb.rpc.server.config.service.ServiceConfig;
10. importcom.github.houbb.rpc.server.service.ServiceFactory;
13. importjava.lang.reflect.InvocationTargetException;
14. importjava.lang.reflect.Method;
15. importjava.util.HashMap;
16. importjava.util.List;
17. importjava.util.Map;
20. /**
21. *默认服务仓库实现
22. *@authorbinbin.hou
23. *@since0.0.6
24. */
25. publicclassDefaultServiceFactoryimplementsServiceFactory{
28. /**
29. *服务map
30. *@since0.0.6
31. */
32. privateMapserviceMap;
35. /**
36. *直接获取对应的method信息
37. *（1）key:serviceId:methodName:param1@param2@param3
38. *（2）value:对应的method信息
39. */
40. privateMapmethodMap;
43. privatestaticfinalDefaultServiceFactoryINSTANCE=newDefaultServiceFactory();
46. privateDefaultServiceFactory(){}
49. publicstaticDefaultServiceFactorygetInstance(){
50. returnINSTANCE;
51. }
54. /**
55. *服务注册一般在项目启动的时候，进行处理。
56. *属于比较重的操作，而且一个服务按理说只应该初始化一次。
57. *此处加锁为了保证线程安全。
58. *@paramserviceConfigList服务配置列表
59. *@returnthis
60. */
61. @Override
62. publicsynchronizedServiceFactoryregisterServices(ListserviceConfigList){
63. ArgUtil.notEmpty(serviceConfigList,"serviceConfigList");
66. //集合初始化
67. serviceMap=newHashMap<>(serviceConfigList.size());
68. //这里只是预估，一般为2个服务。
69. methodMap=newHashMap<>(serviceConfigList.size()*2);
72. for(ServiceConfigserviceConfig:serviceConfigList){
73. serviceMap.put(serviceConfig.id(),serviceConfig.reference());
74. }
77. //存放方法名称
78. for(Map.Entryentry:serviceMap.entrySet()){
79. StringserviceId=entry.getKey();
80. Objectreference=entry.getValue();
83. //获取所有方法列表
84. Method[]methods=reference.getClass().getMethods();
85. for(Methodmethod:methods){
86. StringmethodName=method.getName();
87. if(ReflectMethodUtil.isIgnoreMethod(methodName)){
88. continue;
89. }
92. ListparamTypeNames=ReflectMethodUtil.getParamTypeNames(method);
93. Stringkey=buildMethodKey(serviceId,methodName,paramTypeNames);
94. methodMap.put(key,method);
95. }
96. }
99. returnthis;
100. }
105. @Override
106. publicObjectinvoke(StringserviceId,StringmethodName,ListparamTypeNames,Object[]paramValues){
107. //参数校验
108. ArgUtil.notEmpty(serviceId,"serviceId");
109. ArgUtil.notEmpty(methodName,"methodName");
112. //提供cache，可以根据前三个值快速定位对应的method
113. //根据method进行反射处理。
114. //对于paramTypes进行string连接处理。
115. finalObjectreference=serviceMap.get(serviceId);
116. finalStringmethodKey=buildMethodKey(serviceId,methodName,paramTypeNames);
117. finalMethodmethod=methodMap.get(methodKey);
120. try{
121. returnmethod.invoke(reference,paramValues);
122. }catch(IllegalAccessException|InvocationTargetExceptione){
123. thrownewRpcRuntimeException(e);
124. }
125. }
128. /**
129. *（1）多个之间才用:分隔
130. *（2）参数之间采用@分隔
131. *@paramserviceId服务标识
132. *@parammethodName方法名称
133. *@paramparamTypeNames参数类型名称
134. *@return构建完整的key
135. *@since0.0.6
136. */
137. privateStringbuildMethodKey(StringserviceId,StringmethodName,ListparamTypeNames){
138. Stringparam=CollectionUtil.join(paramTypeNames,PunctuationConst.AT);
139. returnserviceId+PunctuationConst.COLON+methodName+PunctuationConst.COLON
140. +param;
141. }
144. }

ServiceRegistry-服务注册类

接口

1. packagecom.github.houbb.rpc.server.registry;
4. /**
5. *服务注册类
6. *（1）每个应用唯一
7. *（2）每个服务的暴露协议应该保持一致
8. *暂时不提供单个服务的特殊处理，后期可以考虑添加
9. *
10. *@authorbinbin.hou
11. *@since0.0.6
12. */
13. publicinterfaceServiceRegistry{
16. /**
17. *暴露的rpc服务端口信息
18. *@paramport端口信息
19. *@returnthis
20. *@since0.0.6
21. */
22. ServiceRegistryport(finalintport);
25. /**
26. *注册服务实现
27. *@paramserviceId服务标识
28. *@paramserviceImpl服务实现
29. *@returnthis
30. *@since0.0.6
31. */
32. ServiceRegistryregister(finalStringserviceId,finalObjectserviceImpl);
35. /**
36. *暴露所有服务信息
37. *（1）启动服务端
38. *@returnthis
39. *@since0.0.6
40. */
41. ServiceRegistryexpose();
44. }

实现

1. packagecom.github.houbb.rpc.server.registry.impl;
4. importcom.github.houbb.heaven.util.common.ArgUtil;
5. importcom.github.houbb.rpc.common.config.protocol.ProtocolConfig;
6. importcom.github.houbb.rpc.server.config.service.DefaultServiceConfig;
7. importcom.github.houbb.rpc.server.config.service.ServiceConfig;
8. importcom.github.houbb.rpc.server.core.RpcServer;
9. importcom.github.houbb.rpc.server.registry.ServiceRegistry;
10. importcom.github.houbb.rpc.server.service.impl.DefaultServiceFactory;
13. importjava.util.ArrayList;
14. importjava.util.List;
17. /**
18. *默认服务端注册类
19. *@authorbinbin.hou
20. *@since0.0.6
21. */
22. publicclassDefaultServiceRegistryimplementsServiceRegistry{
25. /**
26. *单例信息
27. *@since0.0.6
28. */
29. privatestaticfinalDefaultServiceRegistryINSTANCE=newDefaultServiceRegistry();
32. /**
33. *rpc服务端端口号
34. *@since0.0.6
35. */
36. privateintrpcPort;
39. /**
40. *协议配置
41. *（1）默认只实现tcp
42. *（2）后期可以拓展实现web-service/http/https等等。
43. *@since0.0.6
44. */
45. privateProtocolConfigprotocolConfig;
48. /**
49. *服务配置列表
50. *@since0.0.6
51. */
52. privateListserviceConfigList;
55. privateDefaultServiceRegistry(){
56. //初始化默认参数
57. this.serviceConfigList=newArrayList<>();
58. this.rpcPort=9527;
59. }
62. publicstaticDefaultServiceRegistrygetInstance(){
63. returnINSTANCE;
64. }
67. @Override
68. publicServiceRegistryport(intport){
69. ArgUtil.positive(port,"port");
72. this.rpcPort=port;
73. returnthis;
74. }
77. /**
78. *注册服务实现
79. *（1）主要用于后期服务调用
80. *（2）如何根据id获取实现？非常简单，id是唯一的。
81. *有就是有，没有就抛出异常，直接返回。
82. *（3）如果根据{@linkcom.github.houbb.rpc.common.rpc.domain.RpcRequest}获取对应的方法。
83. *
84. *3.1根据serviceId获取唯一的实现
85. *3.2根据{@linkClass#getMethod(String,Class[])}方法名称+参数类型唯一获取方法
86. *3.3根据{@linkjava.lang.reflect.Method#invoke(Object,Object...)}执行方法
87. *
88. *@paramserviceId服务标识
89. *@paramserviceImpl服务实现
90. *@returnthis
91. *@since0.0.6
92. */
93. @Override
94. @SuppressWarnings("unchecked")
95. publicsynchronizedDefaultServiceRegistryregister(finalStringserviceId,finalObjectserviceImpl){
96. ArgUtil.notEmpty(serviceId,"serviceId");
97. ArgUtil.notNull(serviceImpl,"serviceImpl");
100. //构建对应的其他信息
101. ServiceConfigserviceConfig=newDefaultServiceConfig();
102. serviceConfig.id(serviceId).reference(serviceImpl);
103. serviceConfigList.add(serviceConfig);
106. returnthis;
107. }
110. @Override
111. publicServiceRegistryexpose(){
112. //注册所有服务信息
113. DefaultServiceFactory.getInstance()
114. .registerServices(serviceConfigList);
117. //暴露nettyserver信息
118. newRpcServer(rpcPort).start();
119. returnthis;
120. }
123. }

ServiceConfig 是一些服务的配置信息，接口定义如下：

1. packagecom.github.houbb.rpc.server.config.service;
4. /**
5. *单个服务配置类
6. *
7. *简化用户使用：
8. *在用户使用的时候，这个类应该是不可见的。
9. *直接提供对应的服务注册类即可。
10. *
11. *后续拓展
12. *（1）版本信息
13. *（2）服务端超时时间
14. *
15. *@authorbinbin.hou
16. *@since0.0.6
17. *@param实现类泛型
18. */
19. publicinterfaceServiceConfig{
22. /**
23. *获取唯一标识
24. *@return获取唯一标识
25. *@since0.0.6
26. */
27. Stringid();
30. /**
31. *设置唯一标识
32. *@paramid标识信息
33. *@returnthis
34. *@since0.0.6
35. */
36. ServiceConfigid(Stringid);
39. /**
40. *获取引用实体实现
41. *@return实体实现
42. *@since0.0.6
43. */
44. Treference();
47. /**
48. *设置引用实体实现
49. *@paramreference引用实现
50. *@returnthis
51. *@since0.0.6
52. */
53. ServiceConfigreference(Treference);
56. }

测试

maven 引入

引入服务端的对应 maven 包：

2. com.github.houbb
3. rpc-server
4. 0.0.6

服务端启动

1. //启动服务
2. DefaultServiceRegistry.getInstance()
3. .register(ServiceIdConst.CALC,newCalculatorServiceImpl())
4. .expose();

这里注册了一个计算服务，并且设置对应的实现。

和以前实现类似，此处不再赘述。

启动日志：

1. [DEBUG][2021-10-0513:39:42.638][main][c.g.h.l.i.c.LogFactory.setImplementation]-Logginginitializedusing'classcom.github.houbb.log.integration.adaptors.stdout.StdOutExImpl'adapter.
2. [INFO][2021-10-0513:39:42.645][Thread-0][c.g.h.r.s.c.RpcServer.run]-RPC服务开始启动服务端
3. 十月05,20211:39:43下午io.netty.handler.logging.LoggingHandlerchannelRegistered
4. 信息:[id:0xec4dc74f]REGISTERED
5. 十月05,20211:39:43下午io.netty.handler.logging.LoggingHandlerbind
6. 信息:[id:0xec4dc74f]BIND:0.0.0.0/0.0.0.0:9527
7. 十月05,20211:39:43下午io.netty.handler.logging.LoggingHandlerchannelActive
8. 信息:[id:0xec4dc74f,L:/0:0:0:0:0:0:0:0:9527]ACTIVE
9. [INFO][2021-10-0513:39:43.893][Thread-0][c.g.h.r.s.c.RpcServer.run]-RPC服务端启动完成，监听【9527】端口

ps: 写到这里忽然发现忘记添加对应的 register 日志了，这里可以添加对应的 registerListener 拓展。

原文链接：https://www.toutiao.com/a7017765348539892256/