共计 11109 个字符，预计需要花费 28 分钟才能阅读完成。
概要
RPC（Remote Procedure Call）是指近程过程调用，也就是说两台服务器 A，B，一个利用部署在 A 服务器上，想要调用 B 服务器上利用提供的函数 / 办法。在分布式系统中的零碎环境建设和利用程序设计中有着宽泛的利用。
常见的 RPC 框架有
- Apache Dubbo
- Google gRPC
- Apache Thrift
- Spring Cloud 的 Http 实现
优良的开源框架有高性能，能够像调用本地办法一样调用近程服务，本文着重探讨以下流程的实现
- 低侵入
- 利用 Netty 自定义网络协议实现近程调用
本文的代码可在 github 上自取，
链接：https://github.com/tangbu/myrpc
RPC 的流程
以上，咱们能够看到在实现 RPC 的过程中，咱们须要着重解决一下几点
- 低侵入(咱们应用动静代理来实现办法级别间接调用)
- 实现 RpcRequest 和 RpcResponse 的序列化和反序列化
- 基于 TCP 自定义报文，承载 RpcRequest 和 RpcResponse
- 解决网络连接，网络传输
代码实现
动静代理实现低侵入(咱们应用 jdk 动静代理)
假如应用层存在这样一个接口
public interface HelloWorldService {String helloWorld(String name);
}咱们在调用 HelloWorldService#helloWorld 的时候，心愿自定义外面的逻辑，应用 RPC 来调用，为此咱们就应用动静代理来实现
public class DynamicProxy implements InvocationHandler {
@Override
public Object invoke(Object proxy, Method method, Object[] args) throws ExecutionException, InterruptedException, JsonProcessingException {System.out.println("在调用办法时走到了动静代理外面");
return null;
}
}在真正调用 helloWorld 办法的时候应用如下代码
Class<?> helloWorldServiceClass = HelloWorldService.class;
// 创立代理类对象
HelloWorldService so = (HelloWorldService) Proxy.newProxyInstance(helloWorldServiceClass.getClassLoader(),
new Class[]{HelloWorldService.class}, new DynamicProxy());
String result1 = so.helloWorld("zhangsan");此时，原有接口的逻辑就调到了动静代理办法外面。之后，咱们会将 RPC 的实现封装在 DynamicProxy 这个办法里的实现中。
封装 RpcRequest 对象和 RpcResponse 对象
在执行近程调用的时候，必须通知近程服务，我须要调用那个类，那个办法，办法参数是哪些，入参是什么才能够让他返回后果给我，所以须要封装一下 RpcRequest 对象和 RpcResponse 对象
public class RpcRequest {
private String requestId;
private String className;
private String methodName;
private Class<?>[] parameterTypes;
private Object[] parameters;
private int version;
// getter setter...
}
public class RpcResponse {
private String requestId;
private boolean success;
private String message;
private Object result;
// getter setter...
}定义网络协议，将 RpcRequest 和 RpcResponse 写成字节放在网络报文中传输
自定义报文构造
0----7----15---23---31
| 1 | 2 |
---------------------
| 2 | 3 | 4 |..... 4 前面的是音讯体
---------------------
......5.......
---------------------
序号 1 0-7 version 1byte
序号 2 7-39 总报文长度 4byte
序号 3 39-47 type 音讯类型 1byte
序号 4 47-77 priority 音讯优先级 1byte
序号 5 依据报文总长度减掉 1 - 4 的长度就是 5 的长度相应的依据这个报文构造，能够形象出咱们的 TCP 的报文 Java 类
/**
* @author tangbu
*/
public final class NettyMessage {
private byte version = 1;
private int length;// 音讯长度
private byte type;// 音讯类型
private byte priority;// 音讯优先级;
private JsonNode body; // 目前全副用 json 传递申请
}针对报文构造编写 Netty 的编码器和解码器
编码器的实现 由 NettyMessage 对象变成网络字节
/**
* @author tangbu
*/
public class NettyMessageEncoder extends MessageToByteEncoder<NettyMessage> {private ObjectMapper mapper = new ObjectMapper();
@Override
protected void encode(ChannelHandlerContext ctx, NettyMessage msg, ByteBuf out) throws Exception {out.writeByte(msg.getVersion());
out.writeInt(msg.getLength());
out.writeByte(msg.getType());
out.writeByte(msg.getPriority());
JsonNode body = msg.getBody();
try {byte[] jsonBytes = mapper.writeValueAsBytes(body);
out.writeBytes(jsonBytes);
} catch (JsonProcessingException e) {throw new RuntimeException(e);
}
// 最初填充报文长度
out.setInt(1, out.readableBytes());
}
}
解码器的实现 由网络字节变成 NettyMessage 对象
网络报文接收端须要做两件事件
-
依据报文的 Length 字段的长度从 TCP 流中读取一整个 NettyMessage 对象长度。
发送到网络中的字节以流的模式传输，如果没有指定的拆包规定，报文就像没有标点符号一样字节发送到接收端，造成下层利用无奈辨认，所以须要拆包，netty 提供了针对固定报文构造的拆包器，对于咱们的报文来说，长度占 4 个字节、报文首部偏移量为 1，所以应用这个new LengthFieldBasedFrameDecoder(1460, 1, 4, -5, 0)这个来进行拆包, 通过这个解码器的报文就被拆成一整个 NettyMessage 的一段段字节了
-
读取到字节须要反序列化成一个 NettyMessage 对象。
相应的字节须要转换成 NettyMessage 对象，咱们就应用一个对象解码器来进行解码public class NettyMessageDecoder extends ChannelInboundHandlerAdapter {ObjectMapper mapper = new ObjectMapper(); @Override public void channelRead(ChannelHandlerContext ctx, Object obj) throws Exception {ByteBuf byteBuf= (ByteBuf) obj; if (byteBuf == null){return ;} NettyMessage message = new NettyMessage(); message.setVersion(byteBuf.readByte()); message.setLength(byteBuf.readInt()); message.setType(byteBuf.readByte()); message.setPriority(byteBuf.readByte()); byte[] bodyBytes = new byte[message.getLength() - 7]; byteBuf.readBytes(bodyBytes); JsonNode jsonNode = null; try {jsonNode = mapper.readValue(bodyBytes, JsonNode.class); } catch (IOException e) {throw new RuntimeException(e); } message.setBody(jsonNode); ctx.fireChannelRead(message); }这样，网络申请接收端就能够把承受到的字节读取成 NettyMessage 对象了。
解决 RPC 的客户端和服务端逻辑
RPC 客户端的申请
解决逻辑封装在动静代理类中
public class DynamicProxy implements InvocationHandler { @Override public Object invoke(Object proxy, Method method, Object[] args) throws ExecutionException, InterruptedException, JsonProcessingException {if ("toString".equals(method.getName())){return proxy.toString(); } RpcRequest request = new RpcRequest(); request.setRequestId(UUID.randomUUID().toString()); request.setClassName(method.getDeclaringClass().getName()); request.setMethodName(method.getName()); request.setParameterTypes(method.getParameterTypes()); request.setParameters(args); request.setVersion(1); System.out.println("动静代理封装的 request"+ request); System.out.println("---------------- 执行近程调用 --------"); RpcResponse response = invokeRpc(request); System.out.println("近程调用返回的后果"+ response); return response.getResult();} int count = 0; private RpcResponse invokeRpc(RpcRequest request) throws ExecutionException, InterruptedException, JsonProcessingException { count++; NettyClientHandler nettyClientHandler = ChannelHandlerManager.chooseHandler("127.0.0.1", count % 2 == 0 ? 8888 : 8889); RpcReqResponseFuture rpcReqResponseFuture = nettyClientHandler.sendRpcRequest(request); return rpcReqResponseFuture.get();} }与此同时，在客户端记录连贯，和 request 编号和返回的 response 对应造成回调。
public class NettyClientHandler extends ChannelInboundHandlerAdapter {private ObjectMapper mapper = new ObjectMapper(); private ChannelHandlerContext ctx; private Map<String, RpcReqResponseFuture> reqRespFutures = new HashMap<>(); private Executor executor = Executors.newFixedThreadPool(5); @Override public void channelActive(ChannelHandlerContext ctx) throws Exception {System.out.println("连贯服务提供者" + ctx.channel().remoteAddress() + "胜利"); this.ctx = ctx; ChannelHandlerManager.register(this); } public RpcReqResponseFuture sendRpcRequest(RpcRequest request) throws JsonProcessingException {RpcReqResponseFuture future = new RpcReqResponseFuture(request, executor); reqRespFutures.put(request.getRequestId(), future); NettyMessage nettyMessage = new NettyMessage(); nettyMessage.setType((byte) 1); nettyMessage.setPriority((byte) 2); nettyMessage.setBody(mapper.readValue(mapper.writeValueAsString(request), JsonNode.class)); ctx.channel().writeAndFlush(nettyMessage); return future; } @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {NettyMessage message = (NettyMessage) msg; System.out.println("服务器回复的 Frame:" + message); JsonNode body = message.getBody(); RpcResponse response = mapper.readValue(body.toString(), RpcResponse.class); String requestId = response.getRequestId(); RpcReqResponseFuture future = reqRespFutures.get(requestId); future.done(response); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {cause.printStackTrace(); ctx.close(); ctx.fireExceptionCaught(cause); } public ChannelHandlerContext getCtx() {return ctx;} }
在服务端注册好真正的 HelloWorld 实现类来执行后果，返回 RpcResponse
/**
* @author tangbu
*/
public class NettyServerHandler extends ChannelInboundHandlerAdapter {public static Map<String,Object> serviceMap = new HashMap<>();
private ObjectMapper mapper = new ObjectMapper();
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {NettyMessage message = (NettyMessage) msg;
System.out.println("服务端收到的音讯是:" + message);
JsonNode rpcRequestBody = message.getBody();
RpcRequest rpcRequest = mapper.readValue(rpcRequestBody.toString(), RpcRequest.class);
RpcResponse rpcResponse = handleRpcRequest(rpcRequest);
NettyMessage response = new NettyMessage();
response.setType((byte) 1);
response.setPriority((byte) 2);
response.setBody(mapper.readValue(mapper.writeValueAsString(rpcResponse), JsonNode.class));
ctx.channel().writeAndFlush(response);
}
private RpcResponse handleRpcRequest(RpcRequest rpcRequest) {String requestId = rpcRequest.getRequestId();
RpcResponse response = new RpcResponse();
response.setRequestId(requestId);
try {String className = rpcRequest.getClassName();
String methodName = rpcRequest.getMethodName();
Class<?>[] parameterTypes = rpcRequest.getParameterTypes();
Object[] parameters = rpcRequest.getParameters();
Object o = serviceMap.get(className);
if (o == null){throw new DkRuntimeException("服务不存在");
}
Class clazz = Class.forName(className);
Method method = clazz.getMethod(methodName, parameterTypes);
Object result = method.invoke(o, parameters);
response.setSuccess(true);
response.setResult(result);
} catch (Exception e) {response.setSuccess(false);
response.setMessage(e.getMessage());
}
return response;
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {cause.printStackTrace();
ctx.close();
ctx.fireExceptionCaught(cause);
}
}测试
启动两个 NettyServer，
绑定 8888 和 8889，别离注册 HelloWorldImpl1 和 HelloWorldImpl2 的实现
public class NettyServer1 {public static void main(String[] args) throws InterruptedException {EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup(1);
Map<String, Object> serviceMap = new HashMap<>();
// HelloWorldServiceImpl1
serviceMap.put(HelloWorldService.class.getName(), new HelloWorldServiceImpl1());
NettyServerHandler.serviceMap = serviceMap;
try {ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.childOption(ChannelOption.SO_KEEPALIVE,true)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(1460, 1, 4, -5, 0));
ch.pipeline().addLast(new NettyMessageDecoder());
ch.pipeline().addLast(new NettyMessageEncoder());
ch.pipeline().addLast(new NettyServerHandler());
}
});
// 一个服务绑定 8888，一个绑定 8889 端口
ChannelFuture cf = bootstrap.bind(8888).sync();
cf.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture channelFuture) throws Exception {if (channelFuture.isSuccess()) {System.out.println("监听端口" + 8888 + "胜利");
} else {System.out.println("监听端口" + 8888 + "失败");
}
}
});
cf.channel().closeFuture().sync();} finally {bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();}
}
}构建 Netty 客户端，测试时同时和两个 NettyServer 建设连贯
public class NettyClient implements Runnable {
private String ip;
private int port;
public NettyClient(String ip, int port) {
this.ip = ip;
this.port = port;
}
@Override
public void run() {EventLoopGroup group = new NioEventLoopGroup(1);
try {Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group)
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(1460, 4, 4, -8, 0));
ch.pipeline().addLast(new NettyMessageDecoder());
ch.pipeline().addLast(new NettyMessageEncoder());
ch.pipeline().addLast(new NettyClientHandler());
}
});
System.out.println("客户端 ok..");
ChannelFuture connect = bootstrap.connect(new InetSocketAddress(ip, port));
try {connect.channel().closeFuture().sync();
} catch (InterruptedException e) {e.printStackTrace();
}
} finally {group.shutdownGracefully();
}
}
}执行程序
@Test
public void test2() throws InterruptedException {new Thread(new NettyClient("127.0.0.1",8888)).start();
new Thread(new NettyClient("127.0.0.1",8889)).start();
Thread.sleep(3000);
Class<?> helloWorldServiceClass = HelloWorldService.class;
System.out.println();
System.out.println();
// 创立代理类对象
HelloWorldService so = (HelloWorldService) Proxy.newProxyInstance(helloWorldServiceClass.getClassLoader(),
new Class[]{HelloWorldService.class}, new DynamicProxy());
String result1 = so.helloWorld("zhangsan");
Thread.sleep(1000000L);
}如图，近程调用胜利，繁难的 RPC 代码失去了实现
瞻望
后续能够改良的计划
- 减少 Netty 的异步实现，缩小收发申请的阻塞
- 联合 spring 在 BeanPostProcessor 中对 Bean 进行加强，对立动静代理。
- 能够独自抽出 RpcClient 局部和 RpcServer 局部，不对利用裸露细节
- 减少注册核心，容错，负载平衡，实现高可用
正文完
