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

时间：2022-09-11 08:30:36

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

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

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

基本思路

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

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/

标签：

相关文章