关于java:100-行代码透彻解析-RPC-原理

本文次要阐述的是“RPC 实现原理”，那么首先明确一个问题什么是 RPC 呢？RPC 是 Remote Procedure Call 的缩写，即，近程过程调用。RPC 是一个计算机通信协议。该协定容许运行于一台计算机的程序调用另一台计算机的子程序，而开发人员无需额定地为这个交互编程。

值得注意的是，两个或多个应用程序都散布在不同的服务器上，它们之间的调用都像是本地办法调用一样。接下来咱们便来剖析一下一次 RPC 调用产生了些什么？

当初业界内比拟风行的一些 RPC 框架，例如 Dubbo 提供的是 基于接口的近程办法调用 ，即客户端只须要晓得接口的定义即可调用近程服务。在 Java 中接口并不能间接调用实例办法，必须通过其实现类对象来实现此操作，这意味着客户端必须为这些接口生成 代理对象，对此 Java 提供了 Proxy、InvocationHandler 生成动静代理的反对；生成了代理对象，那么每个具体的办法是怎么调用的呢？jdk 动静代理生成的代理对象调用指定办法时理论会执行 InvocationHandler 中定义的 #invoke 办法，在该办法中实现近程办法调用并获取后果。

抛开客户端，回过头来看 RPC 是两台计算机间的调用，本质上是两台主机间的 网络通信 ，波及到网络通信又必然会有 序列化 、 反序列化 ， 编解码 等一些必须要思考的问题；同时实际上当初大多零碎都是集群部署的，多台主机 / 容器对外提供雷同的服务，如果集群的节点数量很大的话，那么治理服务地址也将是一件非常繁琐的事件，常见的做法是各个服务节点将本人的地址和提供的服务列表注册到一个 注册核心 ，由 注册核心 来对立治理服务列表；这样的做法解决了一些问题同时为客户端减少了一项新的工作——那就是 服务发现，艰深来说就是从注册核心中找到近程办法对应的服务列表并通过某种策略从中选取一个服务地址来实现网络通信。

聊了客户端和 注册核心，另外一个重要的角色天然是服务端，服务端最重要的工作便是提供服务接口的真正实现并在某个端口上监听网络申请，监听到申请后从网络申请中获取到对应的参数（比方服务接口、办法、申请参数等），再依据这些参数通过反射的形式调用接口的真正实现获取后果并将其写入对应的响应流中。

综上所述，一次根本的 RPC 调用流程大抵如下：

| 服务端（生产者）

服务接口

在 RPC 中，生产者和消费者有一个独特的服务接口 API。如下，定义一个 HelloService 接口。

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

服务实现

生产者要提供服务接口的实现，创立 HelloServiceImpl 实现类。

/**
* @author 孙浩
* @Descrption 服务实现
***/
public class HelloServiceImpl implements HelloService {
@Override
public String sayHello(String somebody) {return "hello" + somebody + "!";}
}

服务注册

本例应用 Spring 来治理 bean，采纳自定义 xml 和解析器的形式来将服务实现类载入容器（当然也能够采纳自定义注解的形式，此处不过多阐述）并将服务接口信息注册到注册核心。

首先自定义xsd,

<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 的映射：

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 配置文件中的配置服务类：

<!-- 公布近程服务 -->
<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

import org.springframework.beans.factory.xml.NamespaceHandlerSupport;

/**
* @author 孙浩
* @Descrption 服务公布自定义标签
***/
public class StormServiceNamespaceHandler extends NamespaceHandlerSupport {
@Override
public void init() {registerBeanDefinitionParser("service", new ProviderFactoryBeanDefinitionParser());
}
}

ProviderFactoryBeanDefinitionParser

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

/**
* @author 孙浩
* @Descrption 服务公布
***/
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";
// 服务提供者权重，默认为 1 , 范畴为 [1-100]
private int weight = 1;
// 服务端线程数，默认 10 个线程
private int workerThreads = 10;

@Override
public Object getObject() throws Exception {return serviceProxyObject;}

@Override
public Class<?> getObjectType() {return serviceItf;}

@Override
public void afterPropertiesSet() throws Exception {
// 启动 Netty 服务端
NettyServer.singleton().start(Integer.parseInt(serverPort));
// 注册到 zk, 元数据注册核心
List<ProviderService> providerServiceList = buildProviderServiceInfos();
IRegisterCenter4Provider registerCenter4Provider = RegisterCenter.singleton();
registerCenter4Provider.registerProvider(providerServiceList);
}
}

//================RegisterCenter#registerProvider======================
@Override
public void 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() {
@Override
public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {if (currentChilds == null) {currentChilds = Lists.newArrayList();
}
// 存活的服务 IP 列表
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 服务监听指定端口：

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 {
// 注册解码器 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) {throw new RuntimeException(e);
}
}
}

下面的代码中向 Netty 服务的 pipeline 中增加了编解码和业务处理器，当接管到申请时，通过编解码后，真正解决业务的是业务处理器，即 NettyServerInvokeHandler, 该处理器继承自SimpleChannelInboundHandler, 当数据读取实现将触发一个事件，并调用NettyServerInvokeHandler#channelRead0 办法来解决申请。

@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();
// 利用 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!---------------");
}
}

此处还有局部细节如自定义的编解码器等，篇幅所限不在此详述，继承 MessageToByteEncoder 和 ByteToMessageDecoder 覆写对应的 encode 和 decode 办法即可自定义编解码器，应用到的序列化工具如 Hessian/Proto 等可参考对应的官网文档。

申请和响应包装

为便于封装申请和响应，定义两个 bean 来示意申请和响应。

申请：

/**
* @author 孙浩
* @Descrption
***/
public class StormRequest implements Serializable {

private static final long serialVersionUID = -5196465012408804755L;
//UUID, 惟一标识一次返回值
private String uniqueKey;
// 服务提供者信息
private ProviderService providerService;
// 调用的办法名称
private String invokedMethodName;
// 传递参数
private Object[] args;
// 生产端利用名
private String appName;
// 生产申请超时时长
private long invokeTimeout;
// getter/setter
}

响应：

/**
* @author 孙浩
* @Descrption
***/
public class StormResponse implements Serializable {
private static final long serialVersionUID = 5785265307118147202L;
//UUID, 惟一标识一次返回值
private String uniqueKey;
// 客户端指定的服务超时工夫
private long invokeTimeout;
// 接口调用返回的后果对象
private Object result;
//getter/setter
}

| 客户端（消费者）

客户端（消费者）在 RPC 调用中次要是生成服务接口的代理对象，并从注册核心获取对应的服务列表发动网络申请。

客户端和服务端一样采纳 Spring 来治理 bean 解析 xml 配置等不再赘述，重点看下以下几点：

通过 jdk 动静代理来生成引入服务接口的代理对象

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

从注册核心获取服务列表并根据某种策略选取其中一个服务节点

// 服务接口名称
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);

通过 Netty 建设连贯，发动网络申请

/**
* @author 孙浩
* @Descrption Netty 生产端 bean 代理工厂
***/
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);

// 申明调用 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) {throw new RuntimeException(e);
}
return null;
}
// ...
}

Netty 的响应是异步的，为了在办法调用返回前获取到响应后果，须要将异步的后果同步化。

Netty 异步返回的后果存入阻塞队列

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

申请收回后同步获取后果

// 提交本次调用信息到线程池 fixedThreadPool, 发动调用
Future<StormResponse> responseFuture = fixedThreadPool.submit(RevokerServiceCallable.of(inetSocketAddress, request));
// 获取调用的返回后果
StormResponse response = responseFuture.get(request.getInvokeTimeout(), TimeUnit.MILLISECONDS);
if (response != null) {return response.getResult();
}

//===================================================
// 从返回后果容器中获取返回后果，同时设置期待超时工夫为 invokeTimeout
long invokeTimeout = request.getInvokeTimeout();
StormResponse response = RevokerResponseHolder.getValue(request.getUniqueKey(), invokeTimeout);

Server

/**
* @author 孙浩
* @Descrption
***/
public class MainServer {public static void main(String[] args) throws Exception {
// 公布服务
final ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("storm-server.xml");
System.out.println("服务公布实现");
}
}

Client

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，发送音讯，并获取响应后果）

如有谬误之处，还望大家斧正。

关于java:100-行代码透彻解析-RPC-原理

引言

一次根本的 RPC 调用会波及到什么？

根本实现

测试

结果

总结

Just My Socks（注册教程内含优惠码）

关于java:100-行代码透彻解析-RPC-原理

引 言

一次根本的 RPC 调用会波及到什么？

根本实现

测 试

结 果

总 结

Just My Socks（注册教程 内含优惠码）

引言

测试

结果

总结

Just My Socks（注册教程内含优惠码）