智能摘要文章源自JAVA秀-https://www.javaxiu.com/66941.html

值得注意是，两个或多个应用程序都分布在不同的服务器上，它们之间的调用都像是本地方法调用一样。现在业界内比较流行的一些RPC框架，例如Dubbo提供的是基于接口的远程方法调用，即客户端只需要知道接口的定义即可调用远程服务。和解析器的方式来将服务实现类载入容器(当然也可以采用自定义注解的方式，此处不过多论述)并将服务接口信息注册到注册中心。至此服务实现类已被载入Spring容器中，且服务接口信息也注册到了注册中心。文章源自JAVA秀-https://www.javaxiu.com/66941.html

原文约 4208 字 | 图片 13 张 | 建议阅读 9 分钟 | 评价反馈文章源自JAVA秀-https://www.javaxiu.com/66941.html

100 行代码搞定了 RPC 原理，大家随便问！

点击关注 ? Java面试那些事儿 文章源自JAVA秀-https://www.javaxiu.com/66941.html

点击关注下方公众号，Java面试资料都在这里

文章源自JAVA秀-https://www.javaxiu.com/66941.html

作者：孙浩来源：https://xiaomi-info.github.io/2020/03/02/rpc-achieve/文章源自JAVA秀-https://www.javaxiu.com/66941.html

文章源自JAVA秀-https://www.javaxiu.com/66941.html

# 引言文章源自JAVA秀-https://www.javaxiu.com/66941.html

文章源自JAVA秀-https://www.javaxiu.com/66941.html

本文主要论述的是“RPC 实现原理”，那么首先明确一个问题什么是 RPC 呢？RPC 是 Remote Procedure Call 的缩写，即，远程过程调用。RPC 是一个计算机通信协议。该协议允许运行于一台计算机的程序调用另一台计算机的子程序，而开发人员无需额外地为这个交互编程。文章源自JAVA秀-https://www.javaxiu.com/66941.html

文章源自JAVA秀-https://www.javaxiu.com/66941.html

值得注意是，两个或多个应用程序都分布在不同的服务器上，它们之间的调用都像是本地方法调用一样。接下来我们便来分析一下一次 RPC 调用发生了些什么？文章源自JAVA秀-https://www.javaxiu.com/66941.html

# 一次基本的 RPC 调用会涉及到什么？文章源自JAVA秀-https://www.javaxiu.com/66941.html

文章源自JAVA秀-https://www.javaxiu.com/66941.html

现在业界内比较流行的一些 RPC 框架，例如 Dubbo 提供的是基于接口的远程方法调用，即客户端只需要知道接口的定义即可调用远程服务。文章源自JAVA秀-https://www.javaxiu.com/66941.html

文章源自JAVA秀-https://www.javaxiu.com/66941.html

在 Java 中接口并不能直接调用实例方法，必须通过其实现类对象来完成此操作，这意味着客户端必须为这些接口生成代理对象，对此 Java 提供了 Proxy、InvocationHandler 生成动态代理的支持；生成了代理对象，那么每个具体的发方法是怎么调用的呢？文章源自JAVA秀-https://www.javaxiu.com/66941.html

文章源自JAVA秀-https://www.javaxiu.com/66941.html

jdk 动态代理生成的代理对象调用指定方法时实际会执行 InvocationHandler 中定义的 #invoke 方法，在该方法中完成远程方法调用并获取结果。文章源自JAVA秀-https://www.javaxiu.com/66941.html

抛开客户端，回过头来看 RPC 是两台计算机间的调用，实质上是两台主机间的网络通信，涉及到网络通信又必然会有序列化、反序列化，编解码等一些必须要考虑的问题；同时实际上现在大多系统都是集群部署的，多台主机/容器对外提供相同的服务，如果集群的节点数量很大的话，那么管理服务地址也将是一件十分繁琐的事情，常见的做法是各个服务节点将自己的地址和提供的服务列表注册到一个 注册中心，由 注册中心 来统一管理服务列表；文章源自JAVA秀-https://www.javaxiu.com/66941.html

文章源自JAVA秀-https://www.javaxiu.com/66941.html

这样的做法解决了一些问题同时为客户端增加了一项新的工作——那就是服务发现，通俗来说就是从注册中心中找到远程方法对应的服务列表并通过某种策略从中选取一个服务地址来完成网络通信。文章源自JAVA秀-https://www.javaxiu.com/66941.html

聊了客户端和 注册中心，另外一个重要的角色自然是服务端，服务端最重要的任务便是提供服务接口的真正实现并在某个端口上监听网络请求，监听到请求后从网络请求中获取到对应的参数（比如服务接口、方法、请求参数等），再根据这些参数通过反射的方式调用接口的真正实现获取结果并将其写入对应的响应流中。文章源自JAVA秀-https://www.javaxiu.com/66941.html

综上所述，一次基本的 RPC 调用流程大致如下：文章源自JAVA秀-https://www.javaxiu.com/66941.html

文章源自JAVA秀-https://www.javaxiu.com/66941.html

文章源自JAVA秀-https://www.javaxiu.com/66941.html

# 基本实现文章源自JAVA秀-https://www.javaxiu.com/66941.html

文章源自JAVA秀-https://www.javaxiu.com/66941.html

服务端（生产者）文章源自JAVA秀-https://www.javaxiu.com/66941.html

文章源自JAVA秀-https://www.javaxiu.com/66941.html

• 服务接口文章源自JAVA秀-https://www.javaxiu.com/66941.html

在 RPC 中，生产者和消费者有一个共同的服务接口 API。如下，定义一个 HelloService 接口。文章源自JAVA秀-https://www.javaxiu.com/66941.html

/** * @author 孙浩 * @Descrption 服务接口 ***/publicinterface HelloService {String sayHello(String somebody);}

文章源自JAVA秀-https://www.javaxiu.com/66941.html

• 服务实现文章源自JAVA秀-https://www.javaxiu.com/66941.html

生产者要提供服务接口的实现，创建 HelloServiceImpl 实现类。文章源自JAVA秀-https://www.javaxiu.com/66941.html

/** * @author 孙浩 * @Descrption 服务实现 ***/publicclassHelloServiceImplimplementsHelloService{@Overridepublic String sayHello(String somebody){return"hello " + somebody + "!"; }}

文章源自JAVA秀-https://www.javaxiu.com/66941.html

• 服务注册文章源自JAVA秀-https://www.javaxiu.com/66941.html

文章源自JAVA秀-https://www.javaxiu.com/66941.html

本例使用 Spring 来管理 bean，采用自定义 xml 和解析器的方式来将服务实现类载入容器（当然也可以采用自定义注解的方式，此处不过多论述）并将服务接口信息注册到注册中心。文章源自JAVA秀-https://www.javaxiu.com/66941.html

文章源自JAVA秀-https://www.javaxiu.com/66941.html

首先自定义xsd,文章源自JAVA秀-https://www.javaxiu.com/66941.html

<xsd:elementname="service"><xsd:complexType><xsd:complexContent><xsd:extensionbase="beans:identifiedType"><xsd:attributename="interface"type="xsd:string"use="required"/><xsd:attributename="timeout"type="xsd:int"use="required"/><xsd:attributename="serverPort"type="xsd:int"use="required"/><xsd:attributename="ref"type="xsd:string"use="required"/><xsd:attributename="weight"type="xsd:int"use="optional"/><xsd:attributename="workerThreads"type="xsd:int"use="optional"/><xsd:attributename="appKey"type="xsd:string"use="required"/><xsd:attributename="groupName"type="xsd:string"use="optional"/></xsd:extension></xsd:complexContent></xsd:complexType></xsd:element>

文章源自JAVA秀-https://www.javaxiu.com/66941.html

分别指定 schema 和 xmd，schema 和对应 handler 的映射：文章源自JAVA秀-https://www.javaxiu.com/66941.html

schema文章源自JAVA秀-https://www.javaxiu.com/66941.html

http\://www.storm.com/schema/storm-service.xsd=META-INF/storm-service.xsdhttp\://www.storm.com/schema/storm-reference.xsd=META-INF/storm-reference.xsd

文章源自JAVA秀-https://www.javaxiu.com/66941.html

handler文章源自JAVA秀-https://www.javaxiu.com/66941.html

http\://www.storm.com/schema/storm-service=com.hsunfkqm.storm.framework.spring.StormServiceNamespaceHandlerhttp\://www.storm.com/schema/storm-reference=com.hsunfkqm.storm.framework.spring.StormRemoteReferenceNamespaceHandler

文章源自JAVA秀-https://www.javaxiu.com/66941.html

将编写好的文件放入 classpath 下的 META-INF 目录下：文章源自JAVA秀-https://www.javaxiu.com/66941.html

文章源自JAVA秀-https://www.javaxiu.com/66941.html

文章源自JAVA秀-https://www.javaxiu.com/66941.html

在 Spring 配置文件中配置服务类：文章源自JAVA秀-https://www.javaxiu.com/66941.html

文章源自JAVA秀-https://www.javaxiu.com/66941.html

<!-- 发布远程服务 --><beanid="helloService"class="com.hsunfkqm.storm.framework.test.HelloServiceImpl"/><storm:serviceid="helloServiceRegister"interface="com.hsunfkqm.storm.framework.test.HelloService"ref="helloService"groupName="default"weight="2"appKey="ares"workerThreads="100"serverPort="8081"timeout="600"/>

文章源自JAVA秀-https://www.javaxiu.com/66941.html

编写对应的 Handler 和 Parser：文章源自JAVA秀-https://www.javaxiu.com/66941.html

StormServiceNamespaceHandler文章源自JAVA秀-https://www.javaxiu.com/66941.html

import org.springframework.beans.factory.xml.NamespaceHandlerSupport;/** * @author 孙浩 * @Descrption 服务发布自定义标签 ***/publicclassStormServiceNamespaceHandlerextendsNamespaceHandlerSupport{@Overridepublicvoidinit(){ registerBeanDefinitionParser("service", new ProviderFactoryBeanDefinitionParser()); }}

文章源自JAVA秀-https://www.javaxiu.com/66941.html

ProviderFactoryBeanDefinitionParser：文章源自JAVA秀-https://www.javaxiu.com/66941.html

protected Class getBeanClass(Element element) {return ProviderFactoryBean.class;    }    protected void doParse(Element element, BeanDefinitionBuilder bean) {try { String serviceItf = element.getAttribute("interface"); String serverPort = element.getAttribute("serverPort"); String ref = element.getAttribute("ref");// .... bean.addPropertyValue("serverPort", Integer.parseInt(serverPort)); bean.addPropertyValue("serviceItf", Class.forName(serviceItf)); bean.addPropertyReference("serviceObject", ref);//...if (NumberUtils.isNumber(weight)) { bean.addPropertyValue("weight", Integer.parseInt(weight)); }//... } catch (Exception e) {// ...  } }

文章源自JAVA秀-https://www.javaxiu.com/66941.html

ProviderFactoryBean：文章源自JAVA秀-https://www.javaxiu.com/66941.html

/** * @author 孙浩 * @Descrption 服务发布 ***/publicclass ProviderFactoryBean implements FactoryBean, InitializingBean {//服务接口private Class<?> serviceItf;//服务实现privateObject serviceObject;//服务端口privateString serverPort;//服务超时时间private long timeout;//服务代理对象，暂时没有用到privateObject serviceProxyObject;//服务提供者唯一标识privateString appKey;//服务分组组名privateString groupName = "default";//服务提供者权重，默认为 1 , 范围为 [1-100]private int weight = 1;//服务端线程数，默认 10 个线程private int workerThreads = 10;@OverridepublicObject getObject() throws Exception {return serviceProxyObject; }@Overridepublic Class<?> getObjectType() {return serviceItf; }@Overridepublicvoid afterPropertiesSet() throws Exception {//启动 Netty 服务端 NettyServer.singleton().start(Integer.parseInt(serverPort));//注册到 zk, 元数据注册中心 List<ProviderService> providerServiceList = buildProviderServiceInfos(); IRegisterCenter4Provider registerCenter4Provider = RegisterCenter.singleton(); registerCenter4Provider.registerProvider(providerServiceList); }}//================RegisterCenter#registerProvider======================@Overridepublicvoid registerProvider(final List<ProviderService> serviceMetaData) {if (CollectionUtils.isEmpty(serviceMetaData)) {return; }//连接 zk, 注册服务 synchronized (RegisterCenter.class) {for (ProviderService provider : serviceMetaData) {String serviceItfKey = provider.getServiceItf().getName(); List<ProviderService> providers = providerServiceMap.get(serviceItfKey);if (providers == null) { providers = Lists.newArrayList(); } providers.add(provider); providerServiceMap.put(serviceItfKey, providers); }if (zkClient == null) { zkClient = new ZkClient(ZK_SERVICE, ZK_SESSION_TIME_OUT, ZK_CONNECTION_TIME_OUT, new SerializableSerializer()); }//创建 ZK 命名空间/当前部署应用 APP 命名空间/String APP_KEY = serviceMetaData.get(0).getAppKey();String ZK_PATH = ROOT_PATH + "/" + APP_KEY;boolean exist = zkClient.exists(ZK_PATH);if (!exist) { zkClient.createPersistent(ZK_PATH, true); }for (Map.Entry<String, List<ProviderService>> entry : providerServiceMap.entrySet()) {//服务分组String groupName = entry.getValue().get(0).getGroupName();//创建服务提供者String serviceNode = entry.getKey();String servicePath = ZK_PATH + "/" + groupName + "/" + serviceNode + "/" + PROVIDER_TYPE; exist = zkClient.exists(servicePath);if (!exist) { zkClient.createPersistent(servicePath, true); }//创建当前服务器节点 int serverPort = entry.getValue().get(0).getServerPort();//服务端口 int weight = entry.getValue().get(0).getWeight();//服务权重 int workerThreads = entry.getValue().get(0).getWorkerThreads();//服务工作线程String localIp = IPHelper.localIp();String currentServiceIpNode = servicePath + "/" + localIp + "|" + serverPort + "|" + weight + "|" + workerThreads + "|" + groupName; exist = zkClient.exists(currentServiceIpNode);if (!exist) {//注意，这里创建的是临时节点 zkClient.createEphemeral(currentServiceIpNode); }//监听注册服务的变化，同时更新数据到本地缓存 zkClient.subscribeChildChanges(servicePath, new IZkChildListener() {@Overridepublicvoid handleChildChange(String parentPath, List<String> currentChilds) throws Exception {if (currentChilds == null) { currentChilds = Lists.newArrayList(); }//存活的服务 IP 列表 List<String> activityServiceIpList = Lists.newArrayList(Lists.transform(currentChilds, newFunction<String, String>() {@OverridepublicString apply(String input) {return StringUtils.split(input, "|")[0]; } })); refreshActivityService(activityServiceIpList); } }); } }}

文章源自JAVA秀-https://www.javaxiu.com/66941.html

至此服务实现类已被载入 Spring 容器中，且服务接口信息也注册到了注册中心。文章源自JAVA秀-https://www.javaxiu.com/66941.html

• 网络通信文章源自JAVA秀-https://www.javaxiu.com/66941.html

作为生产者对外提供 RPC 服务，必须有一个网络程序来来监听请求和做出响应。在 Java 领域 Netty 是一款高性能的 NIO 通信框架，很多的框架的通信都是采用 Netty 来实现的，本例中也采用它当做通信服务器。文章源自JAVA秀-https://www.javaxiu.com/66941.html

构建并启动 Netty 服务监听指定端口：文章源自JAVA秀-https://www.javaxiu.com/66941.html

文章源自JAVA秀-https://www.javaxiu.com/66941.html

publicvoidstart(finalint port){synchronized (NettyServer.class) {if (bossGroup != null || workerGroup != null) {return;            } bossGroup = new NioEventLoopGroup(); workerGroup = new NioEventLoopGroup(); ServerBootstrap serverBootstrap = new ServerBootstrap(); serverBootstrap .group(bossGroup, workerGroup) .channel(NioServerSocketChannel.class) .option(ChannelOption.SO_BACKLOG, 1024) .childOption(ChannelOption.SO_KEEPALIVE, true) .childOption(ChannelOption.TCP_NODELAY, true) .handler(new LoggingHandler(LogLevel.INFO)) .childHandler(new ChannelInitializer<SocketChannel>() {@OverrideprotectedvoidinitChannel(SocketChannel ch)throws Exception {//注册解码器 NettyDecoderHandler ch.pipeline().addLast(new NettyDecoderHandler(StormRequest.class, serializeType));//注册编码器 NettyEncoderHandler ch.pipeline().addLast(new NettyEncoderHandler(serializeType));//注册服务端业务逻辑处理器 NettyServerInvokeHandler ch.pipeline().addLast(new NettyServerInvokeHandler()); } });try { channel = serverBootstrap.bind(port).sync().channel(); } catch (InterruptedException e) {thrownew RuntimeException(e); } } }

文章源自JAVA秀-https://www.javaxiu.com/66941.html

上面的代码中向 Netty 服务的 pipeline 中添加了编解码和业务处理器，当接收到请求时，经过编解码后，真正处理业务的是业务处理器，即NettyServerInvokeHandler, 该处理器继承自SimpleChannelInboundHandler, 当数据读取完成将触发一个事件，并调用NettyServerInvokeHandler#channelRead0方法来处理请求。文章源自JAVA秀-https://www.javaxiu.com/66941.html

@OverrideprotectedvoidchannelRead0(ChannelHandlerContext ctx, StormRequest request)throws Exception {if (ctx.channel().isWritable()) {//从服务调用对象里获取服务提供者信息 ProviderService metaDataModel = request.getProviderService();long consumeTimeOut = request.getInvokeTimeout();final String methodName = request.getInvokedMethodName();//根据方法名称定位到具体某一个服务提供者 String serviceKey = metaDataModel.getServiceItf().getName();//获取限流工具类int workerThread = metaDataModel.getWorkerThreads(); Semaphore semaphore = serviceKeySemaphoreMap.get(serviceKey);if (semaphore == null) {synchronized (serviceKeySemaphoreMap) { semaphore = serviceKeySemaphoreMap.get(serviceKey);if (semaphore == null) { semaphore = new Semaphore(workerThread); serviceKeySemaphoreMap.put(serviceKey, semaphore); } } }//获取注册中心服务 IRegisterCenter4Provider registerCenter4Provider = RegisterCenter.singleton(); List<ProviderService> localProviderCaches = registerCenter4Provider.getProviderServiceMap().get(serviceKey); Object result = null;boolean acquire = false;try { ProviderService localProviderCache = Collections2.filter(localProviderCaches, new Predicate<ProviderService>() {@Overridepublicbooleanapply(ProviderService input){return StringUtils.equals(input.getServiceMethod().getName(), methodName); } }).iterator().next(); Object serviceObject = localProviderCache.getServiceObject();//利用反射发起服务调用 Method method = localProviderCache.getServiceMethod();//利用 semaphore 实现限流 acquire = semaphore.tryAcquire(consumeTimeOut, TimeUnit.MILLISECONDS);if (acquire) { result = method.invoke(serviceObject, request.getArgs());//System.out.println("---------------"+result); } } catch (Exception e) { System.out.println(JSON.toJSONString(localProviderCaches) + " " + methodName+" "+e.getMessage()); result = e; } finally {if (acquire) { semaphore.release(); } }//根据服务调用结果组装调用返回对象 StormResponse response = new StormResponse(); response.setInvokeTimeout(consumeTimeOut); response.setUniqueKey(request.getUniqueKey()); response.setResult(result);//将服务调用返回对象回写到消费端 ctx.writeAndFlush(response); } else { logger.error("------------channel closed!---------------"); }}

文章源自JAVA秀-https://www.javaxiu.com/66941.html

此处还有部分细节如自定义的编解码器等，篇幅所限不在此详述，继承 MessageToByteEncoder 和 ByteToMessageDecoder 覆写对应的 encode 和 decode 方法即可自定义编解码器，使用到的序列化工具如 Hessian/Proto 等可参考对应的官方文档。文章源自JAVA秀-https://www.javaxiu.com/66941.html

文章源自JAVA秀-https://www.javaxiu.com/66941.html

• 请求和响应包装文章源自JAVA秀-https://www.javaxiu.com/66941.html

  为便于封装请求和响应，定义两个 bean 来表示请求和响应。文章源自JAVA秀-https://www.javaxiu.com/66941.html

请求：文章源自JAVA秀-https://www.javaxiu.com/66941.html

/** * @author 孙浩 * @Descrption ***/publicclassStormRequestimplementsSerializable{privatestaticfinallong serialVersionUID = -5196465012408804755L;//UUID, 唯一标识一次返回值private String uniqueKey;//服务提供者信息private ProviderService providerService;//调用的方法名称private String invokedMethodName;//传递参数private Object[] args;//消费端应用名private String appName;//消费请求超时时长privatelong invokeTimeout;// getter/setter}

文章源自JAVA秀-https://www.javaxiu.com/66941.html

响应：文章源自JAVA秀-https://www.javaxiu.com/66941.html

/** * @author 孙浩 * @Descrption ***/publicclassStormResponseimplementsSerializable{privatestaticfinallong serialVersionUID = 5785265307118147202L;//UUID, 唯一标识一次返回值private String uniqueKey;//客户端指定的服务超时时间privatelong invokeTimeout;//接口调用返回的结果对象private Object result;//getter/setter}

文章源自JAVA秀-https://www.javaxiu.com/66941.html

客户端（消费者）文章源自JAVA秀-https://www.javaxiu.com/66941.html

文章源自JAVA秀-https://www.javaxiu.com/66941.html

客户端（消费者）在 RPC 调用中主要是生成服务接口的代理对象，并从注册中心获取对应的服务列表发起网络请求。文章源自JAVA秀-https://www.javaxiu.com/66941.html

文章源自JAVA秀-https://www.javaxiu.com/66941.html

客户端和服务端一样采用 Spring 来管理 bean 解析 xml 配置等不再赘述，重点看下以下几点：文章源自JAVA秀-https://www.javaxiu.com/66941.html

• 通过 jdk 动态代理来生成引入服务接口的代理对象文章源自JAVA秀-https://www.javaxiu.com/66941.html

public Object getProxy() {returnProxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class<?>[]{targetInterface}, this);}

文章源自JAVA秀-https://www.javaxiu.com/66941.html

• 从注册中心获取服务列表并依据某种策略选取其中一个服务节点文章源自JAVA秀-https://www.javaxiu.com/66941.html

//服务接口名称String serviceKey = targetInterface.getName();//获取某个接口的服务提供者列表IRegisterCenter4Invoker registerCenter4Consumer = RegisterCenter.singleton();List<ProviderService> providerServices = registerCenter4Consumer.getServiceMetaDataMap4Consume().get(serviceKey);//根据软负载策略，从服务提供者列表选取本次调用的服务提供者ClusterStrategy clusterStrategyService = ClusterEngine.queryClusterStrategy(clusterStrategy);ProviderService providerService = clusterStrategyService.select(providerServices);

文章源自JAVA秀-https://www.javaxiu.com/66941.html

• 通过 Netty 建立连接，发起网络请求文章源自JAVA秀-https://www.javaxiu.com/66941.html

文章源自JAVA秀-https://www.javaxiu.com/66941.html

/** * @author 孙浩 * @Descrption Netty 消费端 bean 代理工厂 ***/publicclassRevokerProxyBeanFactoryimplementsInvocationHandler{private ExecutorService fixedThreadPool = null;//服务接口private Class<?> targetInterface;//超时时间privateint consumeTimeout;//调用者线程数privatestaticint threadWorkerNumber = 10;//负载均衡策略private String clusterStrategy;@Overridepublic Object invoke(Object proxy, Method method, Object[] args)throws Throwable { ...//复制一份服务提供者信息 ProviderService newProvider = providerService.copy();//设置本次调用服务的方法以及接口 newProvider.setServiceMethod(method); newProvider.setServiceItf(targetInterface);//声明调用 AresRequest 对象，AresRequest 表示发起一次调用所包含的信息final StormRequest request = new StormRequest();//设置本次调用的唯一标识 request.setUniqueKey(UUID.randomUUID().toString() + "-" + Thread.currentThread().getId());//设置本次调用的服务提供者信息 request.setProviderService(newProvider);//设置本次调用的方法名称 request.setInvokedMethodName(method.getName());//设置本次调用的方法参数信息 request.setArgs(args);try {//构建用来发起调用的线程池if (fixedThreadPool == null) {synchronized (RevokerProxyBeanFactory.class) {if (null == fixedThreadPool) { fixedThreadPool = Executors.newFixedThreadPool(threadWorkerNumber); } } }//根据服务提供者的 ip,port, 构建 InetSocketAddress 对象，标识服务提供者地址 String serverIp = request.getProviderService().getServerIp();int serverPort = request.getProviderService().getServerPort(); InetSocketAddress inetSocketAddress = new InetSocketAddress(serverIp, serverPort);//提交本次调用信息到线程池 fixedThreadPool, 发起调用 Future<StormResponse> responseFuture = fixedThreadPool.submit(RevokerServiceCallable.of(inetSocketAddress, request));//获取调用的返回结果 StormResponse response = responseFuture.get(request.getInvokeTimeout(), TimeUnit.MILLISECONDS);if (response != null) {return response.getResult(); } } catch (Exception e) {thrownew RuntimeException(e); }returnnull; }// ...}

文章源自JAVA秀-https://www.javaxiu.com/66941.html

Netty 的响应是异步的，为了在方法调用返回前获取到响应结果，需要将异步的结果同步化。文章源自JAVA秀-https://www.javaxiu.com/66941.html

• Netty 异步返回的结果存入阻塞队列文章源自JAVA秀-https://www.javaxiu.com/66941.html

@OverrideprotectedvoidchannelRead0(ChannelHandlerContext channelHandlerContext, StormResponse response)throws Exception {//将 Netty 异步返回的结果存入阻塞队列，以便调用端同步获取 RevokerResponseHolder.putResultValue(response);}

文章源自JAVA秀-https://www.javaxiu.com/66941.html

• 请求发出后同步获取结果文章源自JAVA秀-https://www.javaxiu.com/66941.html

//提交本次调用信息到线程池 fixedThreadPool, 发起调用Future<StormResponse> responseFuture = fixedThreadPool.submit(RevokerServiceCallable.of(inetSocketAddress, request));//获取调用的返回结果StormResponse response = responseFuture.get(request.getInvokeTimeout(), TimeUnit.MILLISECONDS);if (response != null) {return response.getResult();}//===================================================//从返回结果容器中获取返回结果，同时设置等待超时时间为 invokeTimeoutlong invokeTimeout = request.getInvokeTimeout();StormResponse response = RevokerResponseHolder.getValue(request.getUniqueKey(), invokeTimeout);

文章源自JAVA秀-https://www.javaxiu.com/66941.html

# 测试文章源自JAVA秀-https://www.javaxiu.com/66941.html

文章源自JAVA秀-https://www.javaxiu.com/66941.html

Server：文章源自JAVA秀-https://www.javaxiu.com/66941.html

/** * @author 孙浩 * @Descrption ***/publicclassMainServer{publicstaticvoidmain(String[] args)throws Exception {//发布服务final ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("storm-server.xml"); System.out.println(" 服务发布完成"); }}

文章源自JAVA秀-https://www.javaxiu.com/66941.html

Client：文章源自JAVA秀-https://www.javaxiu.com/66941.html

publicclassClient{    private static final Logger logger = LoggerFactory.getLogger(Client.class);    public static void main(String[] args) throws Exception {final ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("storm-client.xml");final HelloService helloService = (HelloService) context.getBean("helloService"); String result = helloService.sayHello("World"); System.out.println(result);        for (;;) { } }}

文章源自JAVA秀-https://www.javaxiu.com/66941.html

# 结果文章源自JAVA秀-https://www.javaxiu.com/66941.html

文章源自JAVA秀-https://www.javaxiu.com/66941.html

生产者：文章源自JAVA秀-https://www.javaxiu.com/66941.html

文章源自JAVA秀-https://www.javaxiu.com/66941.html

文章源自JAVA秀-https://www.javaxiu.com/66941.html

消费者：文章源自JAVA秀-https://www.javaxiu.com/66941.html

文章源自JAVA秀-https://www.javaxiu.com/66941.html

注册中心文章源自JAVA秀-https://www.javaxiu.com/66941.html

文章源自JAVA秀-https://www.javaxiu.com/66941.html

# 总结文章源自JAVA秀-https://www.javaxiu.com/66941.html

文章源自JAVA秀-https://www.javaxiu.com/66941.html

本文简单介绍了 RPC 的整个流程，并实现了一个简单的 RPC 调用。希望阅读完本文之后，能加深你对 RPC 的一些认识。文章源自JAVA秀-https://www.javaxiu.com/66941.html

• 生产者端流程：文章源自JAVA秀-https://www.javaxiu.com/66941.html

• 加载服务接口，并缓存文章源自JAVA秀-https://www.javaxiu.com/66941.html

• 服务注册，将服务接口以及服务主机信息写入注册中心（本例使用的是 zookeeper)文章源自JAVA秀-https://www.javaxiu.com/66941.html

• 启动网络服务器并监听文章源自JAVA秀-https://www.javaxiu.com/66941.html

• 反射，本地调用文章源自JAVA秀-https://www.javaxiu.com/66941.html

• 消费者端流程：文章源自JAVA秀-https://www.javaxiu.com/66941.html

• 代理服务接口生成代理对象文章源自JAVA秀-https://www.javaxiu.com/66941.html

• 服务发现（连接 zookeeper，拿到服务地址列表，通过客户端负载策略获取合适的服务地址）文章源自JAVA秀-https://www.javaxiu.com/66941.html

• 远程方法调用（本例通过 Netty，发送消息，并获取响应结果）文章源自JAVA秀-https://www.javaxiu.com/66941.html

技术交流群文章源自JAVA秀-https://www.javaxiu.com/66941.html

最后，D哥也建了一个技术群，主要探讨一些新的技术和开源项目值不值得去研究及IDEA使用的“骚操作”，有兴趣入群的同学，可长按扫描下方二维码，一定要备注：城市+昵称+技术方向，根据格式备注，可更快被通过且邀请进群。▲长按扫描

热门推荐：知乎热问：一个程序员的水平能差到什么程度？这个牛逼的低代码生成器，现在开源了！如何优雅的写 Controller 层代码？