使用Netty实现RPC

Netty作为一个异步事件驱动的网络应用框架，可以用于快速开发可维护的高性能服务器和客户端。国内著名的RPC框架Dubbo底层使用的是Netty作为网络通信的。本篇文章我们来探索一下RPC框架的本质以及使用Netty来实现一个简单地RPC框架。

1. RPC是什么

RPC（Remote Procedure Call），翻译成中文就是远程过程调用。远程过程就是相对于本地方法而言的，是运行在某个远程的地方而不是本地。通过RPC可以实现像本地函数调用一样调用远程服务，是一种进程间的通信方式。RPC调用的本质可以用下图表示：

通过上面的描述，好像RPC与Socket非常像，都是调用远程的方法，都是client/server模式。但是值得注意的是，RPC并不等同于Socket。Socket是RPC经常采用的通信手段之一，RPC是在Socket的基础上实现的，它比socket需要更多的网络和系统资源。除了Socket，RPC还有其他的通信方法，比如：http、操作系统自带的管道等技术来实现对于远程程序的调用。微软的Windows系统中，RPC就是采用命名管道进行通信。需要了解Socket相关概念的，可以参考之前的这篇文章golang socket编程。

1.1 本地方法调用

本地方法调用使我们开发中最常见的，如下定义一个方法：

public String sayHello(String name) {return "hello, " + name;}

只需要传入一个参数，调用sayHello方法就可以得到一个输出，入参、出参以及方法体都在同一个进程空间中，这就是本地方法调用

1.2 Socket通信

那有没有办法实现不同进程之间通信呢？调用方在进程A，需要调用方法B，但是方法B在进程B中。

最容易想到的实现方式

就是使用Socket通信，使用Socket可以完成跨进程调用，我们需要约定一个进程通信协议，来进行传参，调用函数，出参。写过Socket应该都知道，Socket是比较原始的方式，我们需要更多的去关注一些细节问题，比如参数和函数需要转换成字节流进行网络传输，也就是序列化操作，然后出参时需要反序列化。

假如RPC就是让我们在客户端直接使用Socket远程调用，那无疑是个灾难。所以有没有什么简单方法，让我们的调用方不需要关注细节问题，让调用方像调用本地函数一样，只要传入参数，调用方法，然后坐等返回结果就可以了呢？而这个诉求的解决方案就是RPC框架——为使用方屏蔽底层网络通信的细节。

1.3 RPC框架

RPC框架就是用来解决上面的问题的，它能够让调用方像调用本地函数一样调用远程服务，底层通讯细节对调用方是透明的，将各种复杂性都给屏蔽掉，给予调用方极致体验。

当server需要对方法内实现修改时，client完全感知不到，不用做任何变更。这种方式在跨部门，跨公司合作的时候是非常方便的。

1.4 RPC调用需要关注的技术细节

前面就已经说到RPC框架，可以让调用方像调用本地函数一样调用远程服务。原理就是RPC框架屏蔽Socket通信的相关细节，使调用方可以向调用本地方法一样调用远程方法。

在使用的时候，调用方是直接调用本地函数，传入相应参数，其他细节它不用管，至于通讯细节交给RPC框架来实现。实际上RPC框架采用代理类的方式，具体来说是动态代理的方式，在运行时动态创建新的类，也就是代理类，在该类中实现通讯的细节问题，比如与服务端的连接、参数序列化、结果反序列化等。

除了上述动态代理，还需要约定一个双方通信的协议格式，规定好协议格式，比如请求方法的类名、请求的方法名、请求参数的数据类型，请求的参数等，这样根据格式进行序列化后进行网络传输，然后服务端收到请求对象后按照指定格式进行解码，这样服务端才知道具体该调用哪个方法，传入什么样的参数。

刚才又提到网络传输，RPC框架重要的一环也就是网络传输，服务是部署在不同主机上的，如何高效的进行网络传输，尽量不丢包，保证数据完整无误的快速传递出去？实际上，就是利用我们今天的主角——Netty，Netty是一个高性能的网络通讯框架，它足以胜任我们的任务。

前面说了这么多，再次总结下一个RPC框架需要重点关注哪几个点：

动态代理通信协议序列化网络传输

当然一个优秀的RPC框架需要关注的不止上面几点，只不过本篇文章旨在做一个简易的RPC框架，理解了上面关键的几点就够了

2. 基于Netty实现RPC框架

上面提到，RPC框架的几个技术细节：动态代理、通信协议、序列化以及网络传输，下面我们分别来实现。

2.1 通信协议

通信协议其实就是客户端和服务端约定的通信规则，本质就是用来约定客户端如何将需要调用的远程方法的信息通知给服务端，比如请求方法的类名、请求的方法名、请求参数的数据类型，请求的参数以及服务端返回结果等。所以需要约定一个通信协议用来交互上述信息。

请求对象

@Data@ToStringpublic class RpcRequest {/*** 请求对象的ID，客户端用来验证服务器请求和响应是否匹配*/private String requestId;/*** 类名*/private String className;/*** 方法名*/private String methodName;/*** 参数类型*/private Class<?>[] parameterTypes;/*** 入参*/private Object[] parameters;}

响应对象

@Datapublic class RpcResponse {/*** 响应ID*/private String requestId;/*** 错误信息*/private String error;/*** 返回的结果*/private Object result;}

2.2 序列化

市面上序列化协议很多，比如jdk序列化工具（ObjectInputStream/ObjectOuputStream）、protobuf，kyro、Hessian等，只要不选择jdk自带的序列化方法，（因为其性能太差，序列化后产生的码流太大），其他方式其实都可以，这里为了方便起见，选用JSON作为序列化协议，使用fastjson作为JSON框架。

为了后续扩展方便，先定义序列化接口：

public interface Serializer {/*** java对象转换为二进制** @param object* @return*/byte[] serialize(Object object) throws IOException;/*** 二进制转换成java对象** @param clazz* @param bytes* @param <T>* @return*/<T> T deserialize(Class<T> clazz, byte[] bytes) throws IOException;}

我们采用JSON的方式，这里定义实现类JSONSerializer：

public class JSONSerializer implements Serializer{@Overridepublic byte[] serialize(Object object) {return JSON.toJSONBytes(object);}@Overridepublic <T> T deserialize(Class<T> clazz, byte[] bytes) {return JSON.parseObject(bytes, clazz);}}

如果需要使用其他序列化方式，可以自行实现序列化接口。

2.3 编解码器

约定好协议格式和序列化方式之后，我们还需要编解码器，编码器将请求对象转换为适合于传输的格式（一般来说是字节流），而对应的解码器是将网络字节流转换回应用程序的消息格式。这里我们通过继承Netty提供的抽象类MessageToByteEncoder实现编码器，继承Netty提供的抽象类ByteToMessageDecoder实现解码器，上述抽象类继承关系如下：

编码器

public class RpcEncoder extends MessageToByteEncoder {private Class<?> clazz;private Serializer serializer;public RpcEncoder(Class<?> clazz, Serializer serializer) {this.clazz = clazz;this.serializer = serializer;}@Overrideprotected void encode(ChannelHandlerContext channelHandlerContext, Object msg, ByteBuf byteBuf) throws Exception {if (clazz != null && clazz.isInstance(msg)) {byte[] bytes = serializer.serialize(msg);byteBuf.writeInt(bytes.length);byteBuf.writeBytes(bytes);}}}

解码器

public class RpcDecoder extends ByteToMessageDecoder {private Class<?> clazz;private Serializer serializer;public RpcDecoder(Class<?> clazz, Serializer serializer) {this.clazz = clazz;this.serializer = serializer;}@Overrideprotected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception {//因为之前编码的时候写入一个Int型，4个字节来表示长度if (byteBuf.readableBytes() < 4) {return;}//标记当前读的位置byteBuf.markReaderIndex();int dataLength = byteBuf.readInt();if (byteBuf.readableBytes() < dataLength) {byteBuf.resetReaderIndex();return;}byte[] data = new byte[dataLength];//将byteBuf中的数据读入data字节数组byteBuf.readBytes(data);Object obj = serializer.deserialize(clazz, data);list.add(obj);}}

2.4 Netty客户端

下面来看看Netty客户端是如何实现的，也就是如何使用Netty开启客户端，我们需要注意以下几点：

编写启动方法，指定传输使用Channel指定ChannelHandler，对网络传输中的数据进行读写处理添加编解码器添加失败重试机制添加发送请求消息的方法

@Slf4jpublic class NettyClient {private EventLoopGroup eventLoopGroup;private Channel channel;private ClientHandler clientHandler;private String host;private Integer port;public NettyClient(String host, Integer port) {this.host = host;this.port = port;}public void connect() throws InterruptedException {clientHandler = new ClientHandler();eventLoopGroup = new NioEventLoopGroup();//启动类Bootstrap bootstrap = new Bootstrap();bootstrap.group(eventLoopGroup)//指定传输使用的Channel.channel(NioSocketChannel.class).option(ChannelOption.SO_KEEPALIVE, true).option(ChannelOption.TCP_NODELAY, true).option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000).handler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();//添加编码器pipeline.addLast(new RpcEncoder(RpcRequest.class, new JSONSerializer()));//添加解码器pipeline.addLast(new RpcDecoder(RpcResponse.class, new JSONSerializer()));//请求处理类pipeline.addLast(clientHandler);}});/*** 同步获取Netty连接*/channel = bootstrap.connect(host, port).sync().channel();}/*** 发送消息** @param request* @return*/public RpcResponse send(final RpcRequest request) {try {channel.writeAndFlush(request).await();} catch (InterruptedException e) {e.printStackTrace();}return clientHandler.getRpcResponse(request.getRequestId());}@PreDestroypublic void close() {eventLoopGroup.shutdownGracefully();channel.closeFuture().syncUninterruptibly();}}

我们对于数据的处理重点在于ClientHandler类上，它继承了ChannelDuplexHandler类，可以对出站和入站的数据进行处理。

public class ClientHandler extends ChannelDuplexHandler {/*** 使用Map维护请求对象ID与响应结果Future的映射关系*/private final Map<String, DefaultFuture> futureMap = new ConcurrentHashMap<>();@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {if (msg instanceof RpcResponse) {//获取响应对象RpcResponse response = (RpcResponse) msg;DefaultFuture defaultFuture =futureMap.get(response.getRequestId());//将结果写入DefaultFuturedefaultFuture.setResponse(response);}super.channelRead(ctx, msg);}@Overridepublic void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {if (msg instanceof RpcRequest) {RpcRequest request = (RpcRequest) msg;//发送请求对象之前，先把请求ID保存下来，并构建一个与响应Future的映射关系futureMap.putIfAbsent(request.getRequestId(), new DefaultFuture());}super.write(ctx, msg, promise);}/*** 获取响应结果** @param requestId* @return*/public RpcResponse getRpcResponse(String requestId) {try {DefaultFuture future = futureMap.get(requestId);return future.getRpcResponse(10);} finally {//获取成功以后，从map中移除futureMap.remove(requestId);}}}

从上面实现可以看出，我们定义了一个Map，维护请求ID与响应结果的映射关系，目的是为了客户端用来验证服务端响应是否与请求相匹配，因为Netty的channel可能被多个线程使用，当结果返回时，你不知道是从哪个线程返回的，所以需要一个映射关系。

而我们的结果是封装在DefaultFuture中的，因为Netty是异步框架，所有的返回都是基于Future和Callback机制的，我们这里自定义Future来实现客户端“异步调用”。

public class DefaultFuture {private RpcResponse rpcResponse;private volatile boolean isSucceed = false;private final Object object = new Object();public RpcResponse getRpcResponse(int timeout) {synchronized (object) {while (!isSucceed) {try {object.wait(timeout);} catch (InterruptedException e) {e.printStackTrace();}}return rpcResponse;}}public void setResponse(RpcResponse response) {if (isSucceed) {return;}synchronized (object) {this.rpcResponse = response;this.isSucceed = true;object.notify();}}}

2.5 Netty服务端

Netty服务端的实现跟客户端的实现差不多，只不过要注意的是，当对请求进行解码过后，需要通过代理的方式调用本地函数。下面是服务端实现。

@Slf4jpublic class NettyServer implements InitializingBean {private ServerHandler serverHandler;private EventLoopGroup boss;private EventLoopGroup worker;private Integer serverPort;public NettyServer(ServerHandler serverHandler, Integer serverPort) {this.serverHandler = serverHandler;this.serverPort = serverPort;}@Overridepublic void afterPropertiesSet() throws Exception {//使用zookeeper做注册中心，本文不涉及，可忽略ServiceRegistry registry = null;if (Objects.nonNull(serverPort)) {start(registry);}}public void start(ServiceRegistry registry) throws Exception {//负责处理客户端连接的线程池boss = new NioEventLoopGroup();//负责处理读写操作的线程池worker = new NioEventLoopGroup();ServerBootstrap serverBootstrap = new ServerBootstrap();serverBootstrap.group(boss, worker).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG, 1024).childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();//添加解码器pipeline.addLast(new RpcEncoder(RpcResponse.class, new JSONSerializer()));//添加编码器pipeline.addLast(new RpcDecoder(RpcRequest.class, new JSONSerializer()));//添加请求处理器pipeline.addLast(serverHandler);}});bind(serverBootstrap, serverPort);}/*** 如果端口绑定失败，端口数+1，重新绑定*/public void bind(final ServerBootstrap serverBootstrap, int port) {serverBootstrap.bind(port).addListener(future -> {if (future.isSuccess()) {log.info("端口[ {} ] 绑定成功", port);} else {log.error("端口[ {} ] 绑定失败", port);bind(serverBootstrap, port + 1);}});}@PreDestroypublic void close() throws InterruptedException {boss.shutdownGracefully().sync();worker.shutdownGracefully().sync();log.info("关闭Netty");}}

下面是服务端核心代码，处理读写操作的Handler类：

@Slf4jpublic class ServerHandler extends SimpleChannelInboundHandler<RpcRequest> implements ApplicationContextAware {private ApplicationContext applicationContext;@Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {this.applicationContext = applicationContext;}@Overrideprotected void channelRead0(ChannelHandlerContext ctx, RpcRequest msg) {RpcResponse rpcResponse = new RpcResponse();rpcResponse.setRequestId(msg.getRequestId());try {Object handler = handler(msg);log.info("获取返回结果: {} ", handler);rpcResponse.setResult(handler);} catch (Throwable throwable) {rpcResponse.setError(throwable.toString());throwable.printStackTrace();}ctx.writeAndFlush(rpcResponse);}/*** 服务端使用代理处理请求** @param request* @return*/private Object handler(RpcRequest request) throws ClassNotFoundException, InvocationTargetException {//使用Class.forName进行加载Class文件Class<?> clazz = Class.forName(request.getClassName());Object serviceBean = applicationContext.getBean(clazz);log.info("serviceBean: {}", serviceBean);Class<?> serviceClass = serviceBean.getClass();log.info("serverClass:{}", serviceClass);String methodName = request.getMethodName();Class<?>[] parameterTypes = request.getParameterTypes();Object[] parameters = request.getParameters();//使用CGLIB ReflectFastClass fastClass = FastClass.create(serviceClass);FastMethod fastMethod = fastClass.getMethod(methodName, parameterTypes);log.info("开始调用CGLIB动态代理执行服务端方法...");return fastMethod.invoke(serviceBean, parameters);}}

2.6 客户端代理

客户端使用Java动态代理（要求所有的RPC接口都有实现的接口），需要了解动态代理相关细节的可以参考之前的文章彻底搞懂动态代理。客户端Java动态代理实现如下：

@Slf4jpublic class RpcClientDynamicProxy<T> implements InvocationHandler {private Class<T> interfaceClazz;private String host;private Integer port;public RpcClientDynamicProxy(Class<T> interfaceClazz, String host, Integer port) {this.interfaceClazz = interfaceClazz;this.host = host;this.port = port;}@Overridepublic Object invoke(Object proxy, Method method, Object[] args) throws Throwable {RpcRequest request = new RpcRequest();String requestId = UUID.randomUUID().toString();String className = method.getDeclaringClass().getName();String methodName = method.getName();Class<?>[] parameterTypes = method.getParameterTypes();request.setRequestId(requestId);request.setClassName(className);request.setMethodName(methodName);request.setParameterTypes(parameterTypes);request.setParameters(args);log.info("请求内容: {}", request);//开启Netty 客户端，直连//这里直接指定了server的host和port，正常的RPC框架会从注册中心获取NettyClient nettyClient = new NettyClient(host, port);log.info("开始连接服务端：{}", new Date());nettyClient.connect();RpcResponse send = nettyClient.send(request);log.info("请求调用返回结果：{}", send.getResult());return send.getResult();}@SuppressWarnings("unchecked")public T getProxy() {return (T) Proxy.newProxyInstance(interfaceClazz.getClassLoader(),new Class<?>[]{interfaceClazz},this);}}

在代理方法中，封装请求对象，构建NettyClient对象，并开启客户端，发送请求消息。

2.7 RPC远程调用测试

上面所有代码，就是对RPC的实现。如果要使用我们这个自己实现的RPC框架，我们可以把上述代码打成一个jar包，在分别在client和server端引入。为了模拟这个过程，我这里把上述所有代码作为一个单独的module，然后分别定义两个module来实现服务端和客户端。

其中netty-rpc-server模块依赖netty-rpc模块，netty-rpc-client模块依赖netty-rpc-server和netty-rpc模块。这里贴一下pom配置：

netty-rpc

<?xml version="1.0" encoding="UTF-8"?><project xmlns="/POM/4.0.0"xmlns:xsi="/2001/XMLSchema-instance"xsi:schemaLocation="/POM/4.0.0 /xsd/maven-4.0.0.xsd"><parent><artifactId>netty-explore</artifactId><groupId>com.zhuoli.service</groupId><version>1.0-SNAPSHOT</version></parent><modelVersion>4.0.0</modelVersion><artifactId>netty-rpc</artifactId><dependencies><!-- /artifact/org.springframework/spring-context --><dependency><groupId>org.springframework</groupId><artifactId>spring-context</artifactId><version>5.2.3.RELEASE</version></dependency></dependencies></project>

netty-rpc-server

<?xml version="1.0" encoding="UTF-8"?><project xmlns="/POM/4.0.0"xmlns:xsi="/2001/XMLSchema-instance"xsi:schemaLocation="/POM/4.0.0 /xsd/maven-4.0.0.xsd"><parent><artifactId>netty-explore</artifactId><groupId>com.zhuoli.service</groupId><version>1.0-SNAPSHOT</version></parent><modelVersion>4.0.0</modelVersion><artifactId>netty-rpc-server</artifactId><dependencies><!-- /artifact/org.springframework.boot/spring-boot-starter --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter</artifactId><version>2.2.4.RELEASE</version></dependency><dependency><groupId>com.zhuoli.service</groupId><artifactId>netty-rpc</artifactId><version>1.0-SNAPSHOT</version><scope>compile</scope></dependency></dependencies></project>

netty-rpc-client

<?xml version="1.0" encoding="UTF-8"?><project xmlns="/POM/4.0.0"xmlns:xsi="/2001/XMLSchema-instance"xsi:schemaLocation="/POM/4.0.0 /xsd/maven-4.0.0.xsd"><parent><artifactId>netty-explore</artifactId><groupId>com.zhuoli.service</groupId><version>1.0-SNAPSHOT</version></parent><modelVersion>4.0.0</modelVersion><artifactId>netty-rpc-client</artifactId><dependencies><!-- /artifact/org.springframework.boot/spring-boot-starter --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter</artifactId><version>2.2.4.RELEASE</version></dependency><dependency><groupId>com.zhuoli.service</groupId><artifactId>netty-rpc</artifactId><version>1.0-SNAPSHOT</version><scope>compile</scope></dependency><dependency><groupId>com.zhuoli.service</groupId><artifactId>netty-rpc-server</artifactId><version>1.0-SNAPSHOT</version><scope>compile</scope></dependency></dependencies></project>

2.7.1 RPC服务端

服务端RPC接口：

package com.ty.rpc.server.contract;public interface HelloService {String hello(String name);}

服务端RPC接口实现：

package com.ty.rpc.server.impl;import com.ty.rpc.server.contract.HelloService;import org.springframework.stereotype.Service;@Servicepublic class HelloServiceImpl implements HelloService {@Overridepublic String hello(String name) {return "hello, " + name;}}

服务端启动入口

@SpringBootApplication@Slf4jpublic class RpcServerApplicationContext {@Value("${netty.rpc.server.port}")private Integer port;public static void main(String[] args) throws Exception {SpringApplication.run(RpcServerApplicationContext.class, args);log.info("服务端启动成功");}@Beanpublic NettyServer nettyServer() {return new NettyServer(serverHandler(), port);}@Beanpublic ServerHandler serverHandler() {return new ServerHandler();}}

2.7.2 RPC客户端

客户端调用RPC：

@SpringBootApplication@Slf4jpublic class NettyRpcClientApplicationContext {public static void main(String[] args) throws Exception {SpringApplication.run(NettyRpcClientApplicationContext.class, args);//这里直接指定服务端host和port了HelloService helloService = new RpcClientDynamicProxy<>(HelloService.class, "127.0.0.1", 3663).getProxy();String result = helloService.hello("zhuoli");log.info("响应结果“: {}", result);}}

分别启动server端和client端，服务端日志：

客户端日志：

以上我们基于Netty实现了一个非常简陋的RPC框架，比起成熟的RPC框架来说相差甚远，甚至说基本的注册中心都没有实现，但是通过本次实践，可以说我对于RPC的理解更深了，了解了一个RPC框架到底需要关注哪些方面，未来当我们使用成熟的RPC框架时，比如Dubbo，能够做到心中有数，能明白其底层不过也是使用Netty作为基础通讯框架。如果更深入翻看开源RPC框架源码时，也相对比较容易。

参考链接：

1.基于Netty实现简易RPC框架

2.RPC基本原理以及如何用Netty来实现RPC