一篇文章玩转RPC通信原理，并使用Netty实现一个PRC（

　　1。什么是RPC
　　RPC一般指远程过程调用。RPC是远程过程调用（RemoteProcedureCall）的缩写形式。首先看下服务的演变过程：单一应用架构MVC三层架构PRC分布式服务弹性计算架构
　　接口请求也在慢慢演变：TCPIP报文协议RMI（仅JAVA可用）WebServiceHTTPGPRC（Thrift，Dubbo）SpringRestful（路径风格）
　　总体而言就是随着服务的增多，也伴随着服务之间的调用频繁和繁琐，这就有了PRC这代名词。
　　PRC普通应用在分布式架构中，先看下分布式服务派系阿里系：dubbozookeepernginxspring生态：cloudeurekagateway
　　RPC的核心职能，以dubbo图解为例
　　这个机制现在用的很广泛了，例如cloud中的注册中心和配置中心。大概了解一下理论后，接下来我们用代码来实操，以便更深入的认识PRC。2。Netty实现一个RPC2。1原理概述客户端1。通过bean的初始化回调判断是否需要注入动态代理2。在动态代理回调类中使用Netty调用远程服务，并发送约定协议的消息3。使用回调机制返回服务端响应，并返回原始类服务端1。在bean的回调判断是否为发布的服务，是的话保存在公共map中，初始化时启动Rpc服务2。调用服务解析消息后，通过请求的service获取指定的service，通过反射调用，并将结果返回2。2pom。xml依赖
　　基于springboot2。5。6版本，额外引入lombok和fastjsonnetty依赖dependencygroupIdio。nettygroupIdnettyallartifactIdversion4。1。42。Finalversiondependency2。3apijar包
　　自定义注解，api目录为待发布的API接口，protocol为公用的协议和工具包
　　2。4。客户端架构2。4。1rpc目录下为公用代码，可以单独抽离的
　　2。4。2Controller代码注意这两个声明，并没有加Autowired或ResourceRpcReferenceHellServicehellService；RpcReferenceOrderServiceorderService；GetMapping（hello）publicStringhello（RequestParamStringorderId）｛returnorderService。getOrder（orderId）；｝GetMapping（add）publicintadd（RequestParamIntegera，RequestParamIntegerb）｛returnhellService。add（a，b）；｝
　　PS说明：上面的两个声明没有加Autowired或Resource，所以spring容器在注入的时候不会处理这里两个，本文使用的是反射注入。如果想交由spring处理可以参考mybatis第九话手写实现一个简单的mybatis版本中的Mapper接口注入原理2。4。3核心动态代理处理类RpcBeanPostProcessor实现环境配置回调EnvironmentAware该类为初始化类之后的回调还没到注入阶段因此在这里接收环境的回调，读取RPC的配置传递到代理类中Environmentenvironment；注册之前设置坏境变量OverridepublicvoidsetEnvironment（Environmentenvironment）｛this。environmentenvironment；｝实现了InstantiationAwareBeanPostProcessor接口，重写postProcessAfterInitialization方法可以在bean初始化之前后返回继承类或者代理类，aop就是典型的例子OverridepublicObjectpostProcessAfterInitialization（Objectbean，StringbeanName）throwsBeansException｛Classlt；？clazzbean。getClass（）；遍历所有的声明for（Fieldfield：clazz。getDeclaredFields（））｛如果包含这个注解就创建代理类，并用反射注入if（field。isAnnotationPresent（RpcReference。class））｛Objectinstance；StringbeanClassNamefield。getType（）。getName（）；try｛单例缓存if（cacheProxyMap。containsKey（beanClassName））｛instancecacheProxyMap。get（beanClassName）；｝else｛根据不同的服务名称参数传递不同的rpc调用地址RpcReferenceannotationfield。getAnnotation（RpcReference。class）；生成动态代理instanceProxy。newProxyInstance（field。getType（）。getClassLoader（），newClass〔〕｛field。getType（）｝，可以配置注解参数以获取不同的RPC连接配置newProxyHandler（bean，beanClassName，this。environment。getProperty（annotation。name（）。rpcHost），Integer。valueOf（this。environment。getProperty（annotation。name（）。rpcPort））））；｝log。info（createproxybean：｛｝，beanClassName）；反射注入field。setAccessible（true）；field。set（bean，instance）；cacheProxyMap。put（field。getType（）。getName（），instance）；｝catch（IllegalAccessExceptione）｛log。error（createbeanerror，beanClassName｛｝，beanClassName）；｝｝｝returnbean；｝2。4。4动态代理调用类ProxyHandlerinvoke方法OverridepublicObjectinvoke（Objectproxy，Methodmethod，Object〔〕args）throwsThrowable｛组装协议RpcRequestrequestnewRpcRequest（）；设置一个唯一ID，用来回调request。setReqId（UUID。randomUUID（）。toString（））；request。setService（this。service）；request。setMethod（method。getName（））；request。setParamterType（method。getParameterTypes（））；request。setArgs（args）；发起服务调用NettyClientnettyClientnewNettyClient（）；nettyClient。start（rpcHost，rpcPort，newMyRpcClientHandler（））；返回结果returnnettyClient。sendRequest（request）；｝2。4。5NettyClient公共类该类不是单例的，但是保存通道和回调的Map是单例的publicChannelchannel；publicvoidstart（Stringhost，intport，RpcHandlerrpcHandler）｛StringmapKeyhost：port；if（NettyConstans。clientMap。containsKey（mapKey））｛this。channelNettyConstans。clientMap。get（mapKey）；return；｝NioEventLoopGroupb1newNioEventLoopGroup（）；BootstrapbsnewBootstrap（）。group（b1）。channel（NioSocketChannel。class）。handler（newChannelInitializerChannel（）｛OverrideprotectedvoidinitChannel（Channelchannel）throwsException｛ChannelPipelinepipelinechannel。pipeline（）；这里偷懒就直接用string的编解码了pipeline。addLast（newStringEncoder（））；pipeline。addLast（newStringDecoder（））；pipeline。addLast（rpcHandler）；｝｝）；try｛客户端连接服务端ChannelFuturefuturebs。connect（host，port）。sync（）；future。addListener（listen｛if（listen。isSuccess（））｛log。info（connectrpcservicesuccess，｛｝：｛｝，host，port）；｝｝）；channelfuture。channel（）；保存为单例NettyConstans。clientMap。put（mapKey，channel）；｝catch（Exceptione）｛b1。shutdownGracefully（）；log。error（connectrpcserviceerror，｛｝：｛｝，host，port）；｝｝publicObjectsendRequest（RpcRequestrpcRequest）throwsException｛自定义一个返回结果的回调保存到单例Map中RpcFutureRpcResponserpcFuturenewRpcFuture（newDefaultPromiseRpcResponse（newDefaultEventLoop（）））；NettyConstans。rpcFutureMap。put（rpcRequest。getReqId（），rpcFuture）；消息发送，编解码为string，所以发送的是stringchannel。writeAndFlush（JSONObject。toJSONString（rpcRequest））；实际上为阻塞等待回调由接收消息那里回调其实还有一个熔断线程处理这些超时或者一直没有回调的returnrpcFuture。getPromise（）。get（）。getContent（）；｝2。4。6客户端接收消息handlerMyRpcClientHandler协议RpcResponseOverrideprotectedvoidchannelRead0（ChannelHandlerContextctx，Stringmsg）throwsException｛log。info（RpcResponsereceivemsg：｛｝，msg）；RpcResponseresponseJSONObject。parseObject（msg，RpcResponse。class）；未知的消息直接忽略if（responsenull！NettyConstans。rpcFutureMap。containsKey（response。getReqId（）））return；给指定的ReqId回调NettyConstans。rpcFutureMap。get（response。getReqId（））。getPromise（）。setSuccess（response）；NettyConstans。rpcFutureMap。remove（response。getReqId（））；｝OverridepublicvoidexceptionCaught（ChannelHandlerContextctx，Throwablecause）throwsException｛log。error（连接出现异常，重置连接：｛｝，ctx。channel（）。remoteAddress（））；异常重连服务端重启之类的NettyConstans。clientMap。remove（ctx。channel（）。remoteAddress（）。toString（））；｝
　　客户端的代码基本上贴完了，比较复杂，服务端会比较简单，接下来看看服务端的代码2。5服务端架构2。5。2。1bean的初始化回调RpcBeanPostProcessorstaticMapString，ObjectbeanMapnewConcurrentHashMap（）；OverridepublicObjectpostProcessAfterInitialization（Objectbean，StringbeanName）throwsBeansException｛Classlt；？clazzbean。getClass（）；只要包含该注解的就报保存到Map中if（clazz。isAnnotationPresent（RpcService。class））｛存的是服务发布的接口类名称beanMap。put（clazz。getInterfaces（）〔0〕。getName（），bean）；log。info（registerrpcservice：｛｝，clazz。getInterfaces（）〔0〕。getName（））；｝returnbean；｝
　　这里没有往注册中心上发布了，直接以本地Map的形式保存的。主要是为弄懂原理2。5。2NettyService初始化使用springboot的启动回调开始一个RPC服务Overridepublicvoidrun（String。。。args）throwsException｛启动代码就不贴了编解码为StringNettyService。start（port，newMyRpcHandler（））；｝自定义handler类MyRpcHandler协议RpcRequestOverrideprotectedvoidchannelRead0（ChannelHandlerContextctx，Stringmsg）throwsException｛log。info（RpcRequestreceivemsg：｛｝，msg）；RpcRequestrequestJSONObject。parseObject（msg，RpcRequest。class）；if（requestnullrequest。getReqId（）null）return；Stringservicerequest。getService（）；ObjectbeanRpcBeanPostProcessor。beanMap。get（service）；根据方法名称和参数类型获取类中的方法Methodmethodbean。getClass（）。getMethod（request。getMethod（），request。getParamterType（））；Objectresultmethod。invoke（bean，request。getArgs（））；响应协议RpcResponseresponsenewRpcResponse（）；response。setReqId（request。getReqId（））；response。setContent（result）；写出和发送同理ctx。writeAndFlush（JSONObject。toJSONString（response））；｝3。RPC测试
　　分别启动客户端和服务端3。1客户端调用控制台日志createproxybean：com。exmaple。demo。api。HellServicecreateproxybean：com。exmaple。demo。api。OrderService执行http：127。0。0。1：8080hello？orderId1234567connectrpcservicesuccess，127。0。0。1：18080RpcResponsereceivemsg：｛content：selectorderservicebyorderId：1234567，reqId：61a37ef56a974fe79ba9d8c3a955c8c0｝3。2服务端调用日志startremoteservice：18080RpcRequestreceivemsg：｛args：〔1234567〕，method：getOrder，paramterType：〔java。lang。String〕，reqId：61a37ef56a974fe79ba9d8c3a955c8c0，service：com。exmaple。demo。api。OrderService｝第二次调用http：127。0。0。1：8080add？a4545b12日志RpcRequestreceivemsg：｛args：〔4545，12〕，method：add，paramterType：〔int，int〕，reqId：4f312678b4634db9a861d8b4b9c9fc4a，service：com。exmaple。demo。api。HellService｝4。总结4。1关于反射注入
　　正常应该使用的是FactoryBean的方式注入的，这里只是为了搞懂原理，忽略！4。2关于Rpc服务地址
　　正常的RPC服务，会先从注册中心获取这个服务发布的地址，也就是我们配置中的地址实际上是注册中心的地址建立连接后，应该会保持心跳，第二次调用不再重新建立连接4。3关于阻塞异步回调
　　实际上还有熔断机制，应该处理掉一直等待的回调