引言
本文主要论述的是“RPC 实现原理”,那么首先明确一个问题什么是 RPC 呢?RPC 是 Remote Procedure Call 的缩写,即,远程过程调用。RPC 是一个计算机通信协议。该协议允许运行于一台计算机的程序调用另一台计算机的子程序,而开发人员无需额外地为这个交互编程。
值得注意是,两个或多个应用程序都分布在不同的服务器上,它们之间的调用都像是本地方法调用一样。接下来我们便来分析一下一次 RPC 调用发生了些什么?
一次基本的 RPC 调用会涉及到什么?
现在业界内比较流行的一些 RPC 框架,例如 Dubbo 提供的是基于接口的远程方法调用
,即客户端只需要知道接口的定义即可调用远程服务。在 Java 中接口并不能直接调用实例方法,必须通过其实现类对象来完成此操作,这意味着客户端必须为这些接口生成代理对象
,对此 Java 提供了 Proxy
、InvocationHandler
生成动态代理的支持;生成了代理对象,那么每个具体的发方法是怎么调用的呢?jdk 动态代理生成的代理对象调用指定方法时实际会执行 InvocationHandler
中定义的 #invoke
方法,在该方法中完成远程方法调用并获取结果。
抛开客户端,回过头来看 RPC 是两台计算机间的调用,实质上是两台主机间的网络通信
,涉及到网络通信又必然会有序列化、反序列化
,编解码
等一些必须要考虑的问题;同时实际上现在大多系统都是集群部署的,多台主机/容器对外提供相同的服务,如果集群的节点数量很大的话,那么管理服务地址也将是一件十分繁琐的事情,常见的做法是各个服务节点将自己的地址和提供的服务列表注册到一个 注册中心
,由 注册中心
来统一管理服务列表;这样的做法解决了一些问题同时为客户端增加了一项新的工作——那就是服务发现
,通俗来说就是从注册中心中找到远程方法对应的服务列表并通过某种策略从中选取一个服务地址来完成网络通信。
聊了客户端和 注册中心
,另外一个重要的角色自然是服务端,服务端最重要的任务便是提供服务接口的真正实现并在某个端口上监听网络请求,监听到请求后从网络请求中获取到对应的参数(比如服务接口、方法、请求参数等),再根据这些参数通过反射
的方式调用接口的真正实现获取结果并将其写入对应的响应流中。
综上所述,一次基本的 RPC 调用流程大致如下:
基本实现
服务端(生产者)
在 RPC 中,生产者和消费者有一个共同的服务接口 API。如下,定义一个 HelloService 接口。
1 2 3 4 5 6 7 public interface HelloService { String sayHello (String somebody) ; }
生产者要提供服务接口的实现,创建 HelloServiceImpl 实现类。
1 2 3 4 5 6 7 8 9 10 public class HelloServiceImpl implements HelloService { @Override public String sayHello (String somebody) { return "hello " + somebody + "!" ; } }
本例使用 Spring 来管理 bean,采用自定义 xml 和解析器的方式来将服务实现类载入容器(当然也可以采用自定义注解的方式,此处不过多论述)并将服务接口信息注册到注册中心。
首先自定义xsd
,
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 <xsd:element name ="service" > <xsd:complexType > <xsd:complexContent > <xsd:extension base ="beans:identifiedType" > <xsd:attribute name ="interface" type ="xsd:string" use ="required" /> <xsd:attribute name ="timeout" type ="xsd:int" use ="required" /> <xsd:attribute name ="serverPort" type ="xsd:int" use ="required" /> <xsd:attribute name ="ref" type ="xsd:string" use ="required" /> <xsd:attribute name ="weight" type ="xsd:int" use ="optional" /> <xsd:attribute name ="workerThreads" type ="xsd:int" use ="optional" /> <xsd:attribute name ="appKey" type ="xsd:string" use ="required" /> <xsd:attribute name ="groupName" type ="xsd:string" use ="optional" /> </xsd:extension > </xsd:complexContent > </xsd:complexType > </xsd:element >
分别指定 schema 和 xmd,schema 和对应 handler 的映射:
1 2 3 4 5 6 schema http\://www.storm.com/schema/storm-service.xsd=META-INF/storm-service.xsd http\://www.storm.com/schema/storm-reference.xsd=META-INF/storm-reference.xsd handler http\://www.storm.com/schema/storm-service=com.hsunfkqm.storm.framework.spring.StormServiceNamespaceHandler http\://www.storm.com/schema/storm-reference=com.hsunfkqm.storm.framework.spring.StormRemoteReferenceNamespaceHandler
将编写好的文件放入 classpath
下的 META-INF
目录下:
在 Spring 配置文件中配置服务类:
1 2 3 4 5 6 7 8 9 10 11 <bean id ="helloService" class ="com.hsunfkqm.storm.framework.test.HelloServiceImpl" /> <storm:service id ="helloServiceRegister" interface ="com.hsunfkqm.storm.framework.test.HelloService" ref ="helloService" groupName ="default" weight ="2" appKey ="ares" workerThreads ="100" serverPort ="8081" timeout ="600" />
编写对应的 Handler 和 Parser:
StormServiceNamespaceHandler
1 2 3 4 5 6 7 8 9 10 11 12 import org.springframework.beans.factory.xml.NamespaceHandlerSupport;public class StormServiceNamespaceHandler extends NamespaceHandlerSupport { @Override public void init () { registerBeanDefinitionParser("service" , new ProviderFactoryBeanDefinitionParser()); } }
ProviderFactoryBeanDefinitionParser
:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 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) { } }
ProviderFactoryBean
:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 public class ProviderFactoryBean implements FactoryBean , InitializingBean { private Class<?> serviceItf; private Object serviceObject; private String serverPort; private long timeout; private Object serviceProxyObject; private String appKey; private String groupName = "default" ; private int weight = 1 ; private int workerThreads = 10 ; @Override public Object getObject () throws Exception { return serviceProxyObject; } @Override public Class<?> getObjectType() { return serviceItf; } @Override public void afterPropertiesSet () throws Exception { NettyServer.singleton().start(Integer.parseInt(serverPort)); List<ProviderService> providerServiceList = buildProviderServiceInfos(); IRegisterCenter4Provider registerCenter4Provider = RegisterCenter.singleton(); registerCenter4Provider.registerProvider(providerServiceList); } } @Override public void registerProvider (final List<ProviderService> serviceMetaData) { if (CollectionUtils.isEmpty(serviceMetaData)) { return ; } 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()); } 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() { @Override public void handleChildChange (String parentPath, List<String> currentChilds) throws Exception { if (currentChilds == null ) { currentChilds = Lists.newArrayList(); } List<String> activityServiceIpList = Lists.newArrayList(Lists.transform(currentChilds, new Function<String, String>() { @Override public String apply (String input) { return StringUtils.split(input, "|" )[0 ]; } })); refreshActivityService(activityServiceIpList); } }); } } }
至此服务实现类已被载入 Spring 容器中,且服务接口信息也注册到了注册中心。
作为生产者对外提供 RPC 服务,必须有一个网络程序来来监听请求和做出响应。在 Java 领域 Netty 是一款高性能的 NIO 通信框架,很多的框架的通信都是采用 Netty 来实现的,本例中也采用它当做通信服务器。
构建并启动 Netty 服务监听指定端口:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 public void start (final int 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>() { @Override protected void initChannel (SocketChannel ch) throws Exception { ch.pipeline().addLast(new NettyDecoderHandler(StormRequest.class, serializeType)); ch.pipeline().addLast(new NettyEncoderHandler(serializeType)); ch.pipeline().addLast(new NettyServerInvokeHandler()); } }); try { channel = serverBootstrap.bind(port).sync().channel(); } catch (InterruptedException e) { throw new RuntimeException(e); } } }
上面的代码中向 Netty 服务的 pipeline 中添加了编解码和业务处理器,当接收到请求时,经过编解码后,真正处理业务的是业务处理器,即NettyServerInvokeHandler
, 该处理器继承自SimpleChannelInboundHandler
, 当数据读取完成将触发一个事件,并调用NettyServerInvokeHandler#channelRead0
方法来处理请求。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 @Override protected void channelRead0 (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>() { @Override public boolean apply (ProviderService input) { return StringUtils.equals(input.getServiceMethod().getName(), methodName); } }).iterator().next(); Object serviceObject = localProviderCache.getServiceObject(); Method method = localProviderCache.getServiceMethod(); acquire = semaphore.tryAcquire(consumeTimeOut, TimeUnit.MILLISECONDS); if (acquire) { result = method.invoke(serviceObject, request.getArgs()); } } 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!---------------" ); } }
此处还有部分细节如自定义的编解码器等,篇幅所限不在此详述,继承 MessageToByteEncoder
和 ByteToMessageDecoder
覆写对应的 encode
和 decode
方法即可自定义编解码器,使用到的序列化工具如 Hessian/Proto 等可参考对应的官方文档。
请求和响应包装
为便于封装请求和响应,定义两个 bean 来表示请求和响应。
请求:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 public class StormRequest implements Serializable { private static final long serialVersionUID = -5196465012408804755L ; private String uniqueKey; private ProviderService providerService; private String invokedMethodName; private Object[] args; private String appName; private long invokeTimeout; }
响应:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 public class StormResponse implements Serializable { private static final long serialVersionUID = 5785265307118147202L ; private String uniqueKey; private long invokeTimeout; private Object result; }
客户端(消费者)
客户端(消费者)在 RPC 调用中主要是生成服务接口的代理对象,并从注册中心获取对应的服务列表发起网络请求。
客户端和服务端一样采用 Spring 来管理 bean 解析 xml 配置等不再赘述,重点看下以下几点:
通过 jdk 动态代理来生成引入服务接口的代理对象
1 2 3 public Object getProxy () { return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class<?>[]{targetInterface}, this ); }
从注册中心获取服务列表并依据某种策略选取其中一个服务节点
1 2 3 4 5 6 7 8 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);
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 public class RevokerProxyBeanFactory implements InvocationHandler { private ExecutorService fixedThreadPool = null ; private Class<?> targetInterface; private int consumeTimeout; private static int threadWorkerNumber = 10 ; private String clusterStrategy; @Override public Object invoke (Object proxy, Method method, Object[] args) throws Throwable { ... ProviderService newProvider = providerService.copy(); newProvider.setServiceMethod(method); newProvider.setServiceItf(targetInterface); 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); } } } String serverIp = request.getProviderService().getServerIp(); int serverPort = request.getProviderService().getServerPort(); InetSocketAddress inetSocketAddress = new InetSocketAddress(serverIp, serverPort); 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) { throw new RuntimeException(e); } return null ; } }
Netty 的响应是异步的,为了在方法调用返回前获取到响应结果,需要将异步的结果同步化。
1 2 3 4 5 @Override protected void channelRead0 (ChannelHandlerContext channelHandlerContext, StormResponse response) throws Exception { RevokerResponseHolder.putResultValue(response); }
1 2 3 4 5 6 7 8 9 10 11 12 Future<StormResponse> responseFuture = fixedThreadPool.submit(RevokerServiceCallable.of(inetSocketAddress, request)); StormResponse response = responseFuture.get(request.getInvokeTimeout(), TimeUnit.MILLISECONDS); if (response != null ) { return response.getResult(); } long invokeTimeout = request.getInvokeTimeout();StormResponse response = RevokerResponseHolder.getValue(request.getUniqueKey(), invokeTimeout);
测试
Server
:
1 2 3 4 5 6 7 8 9 10 11 public class MainServer { public static void main (String[] args) throws Exception { final ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("storm-server.xml" ); System.out.println(" 服务发布完成" ); } }
Client
:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 public class Client { 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 (;;) { } } }
结果
生产者:
消费者:
注册中心
总结
本文简单介绍了 RPC 的整个流程,并实现了一个简单的 RPC 调用。希望阅读完本文之后,能加深你对 RPC 的一些认识。
生产者端流程:
加载服务接口,并缓存
服务注册,将服务接口以及服务主机信息写入注册中心(本例使用的是 zookeeper)
启动网络服务器并监听
反射,本地调用
消费者端流程:
代理服务接口生成代理对象
服务发现(连接 zookeeper,拿到服务地址列表,通过客户端负载策略获取合适的服务地址)
远程方法调用(本例通过 Netty,发送消息,并获取响应结果)
限于篇幅,本文代码并不完整,如有需要,访问:https://github.com/fankongqiumu/storm.git,获取完整代码。
如有错误之处,还望大家指正。
本文转自:https://xiaomi-info.github.io/2020/03/02/rpc-achieve/
注:文中所用到的代码已上传至 github: https://github.com/fankongqiumu/storm.git
[作者简介] 孙浩,信息部售后组研发工程师