徒手撸框架--实现 RPC 远程调用

1. server
   server 是负责处理客户端请求的组件。在互联网高并发的环境下，使用 Nio 非阻塞的方式可以相对轻松的应付高并发的场景。netty 是一个优秀的 Nio 处理框架。Server 的关键代码如下：

netty 是基于 Recotr 模型的。所以需要初始化两组线程 boss 和 worker 。boss 负责分发请求，worker 负责执行相应的 handler：

@Bean
public ServerBootstrap serverBootstrap() throws InterruptedException {
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.group(bossGroup(), workerGroup())
.channel(NioServerSocketChannel.class)
.handler(new LoggingHandler(LogLevel.DEBUG))
.childHandler(serverInitializer);
Map<ChannelOption<?>, Object> tcpChannelOptions = tcpChannelOptions();
Set<ChannelOption<?>> keySet = tcpChannelOptions.keySet();
for (@SuppressWarnings("rawtypes") ChannelOption option : keySet) {
serverBootstrap.option(option, tcpChannelOptions.get(option));
}
return serverBootstrap;
}
netty 的操作是基于 pipeline 的。所以我们需要把在 protocol 实现的几个 coder 注册到 netty 的 pipeline 中。

ChannelPipeline pipeline = ch.pipeline();
// 处理 tcp 请求中粘包的 coder，具体作用可以自行 google
pipeline.addLast(new LengthFieldBasedFrameDecoder(65535,0,4));
// protocol 中实现的 序列化和反序列化 coder
pipeline.addLast(new RpcEncoder(RpcResponse.class,new JsonSerialization()));
pipeline.addLast(new RpcDecoder(RpcRequest.class,new JsonSerialization()));
// 具体处理请求的 handler 下文具体解释
pipeline.addLast(serverHandler);
实现具体的 ServerHandler 用于处理真正的调用。
ServerHandler 继承 SimpleChannelInboundHandler<RpcRequest>。简单来说这个 InboundHandler 会在数据被接受时或者对于的 Channel 的状态发生变化的时候被调用。当这个 handler 读取数据的时候方法 channelRead0() 会被用，所以我们就重写这个方法就够了。

@Override
protected void channelRead0(ChannelHandlerContext ctx, RpcRequest msg) throws Exception {
RpcResponse rpcResponse = new RpcResponse();
rpcResponse.setRequestId(msg.getRequestId());
try{
// 收到请求后开始处理请求
Object handler = handler(msg);
rpcResponse.setResult(handler);
}catch (Throwable throwable){
// 如果抛出异常也将异常存入 response 中
rpcResponse.setThrowable(throwable);
throwable.printStackTrace();
}
// 操作完以后写入 netty 的上下文中。netty 自己处理返回值。
ctx.writeAndFlush(rpcResponse);
}
handler(msg) 实际上使用的是 cglib 的 Fastclass 实现的，其实根本原理，还是反射。学好 java 中的反射真的可以为所欲为。

private Object handler(RpcRequest request) throws Throwable {
Class<?> clz = Class.forName(request.getClassName());
Object serviceBean = applicationContext.getBean(clz);
Class<?> serviceClass = serviceBean.getClass();
String methodName = request.getMethodName();
Class<?>[] parameterTypes = request.getParameterTypes();
Object[] parameters = request.getParameters();
// 根本思路还是获取类名和方法名，利用反射实现调用
FastClass fastClass = FastClass.create(serviceClass);
FastMethod fastMethod = fastClass.getMethod(methodName,parameterTypes);
// 实际调用发生的地方
return fastMethod.invoke(serviceBean,parameters);
}
总体上来看，server 的实现不是很困难。核心的知识点是 netty 的 channel 的使用和 cglib 的反射机制。

1. client
   future

其实，对于我来说，client 的实现难度，远远大于 server 的实现。netty 是一个异步框架，所有的返回都是基于 Future 和 Callback 的机制。

所以在阅读以下文字前强烈推荐，我之前写的一篇文章 Future 研究。利用经典的 wite 和 notify 机制，实现异步的获取请求的结果。

/**

• @author zhengxin
  */
  public class DefaultFuture {
  private RpcResponse rpcResponse;
  private volatile boolean isSucceed = false;
  private final Object object = new Object();
  public RpcResponse getResponse(int timeout){
  synchronized (object){
  while (!isSucceed){
  try {
  //wait
  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;
  //notiy
  object.notify();
  }
  }
  }
  复用资源

为了能够提升 client 的吞吐量，可提供的思路有以下几种：

使用对象池：建立多个 client 以后保存在对象池中。但是代码的复杂度和维护 client 的成本会很高。
尽可能的复用 netty 中的 channel。
之前你可能注意到，为什么要在 RpcRequest 和 RpcResponse 中增加一个 ID。因为 netty 中的 channel 是会被多个线程使用的。当一个结果异步的返回后，你并不知道是哪个线程返回的。这个时候就可以考虑利用一个 Map，建立一个 ID 和 Future 映射。这样请求的线程只要使用对应的 ID 就能获取，相应的返回结果。

/**

使用 Transporters 对请求进行封装。

public class Transporters {
public static RpcResponse send(RpcRequest request){
NettyClient nettyClient = new NettyClient("127.0.0.1", 8080);
nettyClient.connect(nettyClient.getInetSocketAddress());
RpcResponse send = nettyClient.send(request);
return send;
}
}
动态代理的实现

动态代理技术最广为人知的应用，应该就是 Spring Aop，面向切面的编程实现。动态的在原有方法Before 或者 After 添加代码。而 RPC 框架中动态代理的作用就是彻底替换原有方法，直接调用远程方法。

代理工厂类：

生成 RequestId。
拼装 RpcRequest。
调用 Transports 发送请求，获取结果。
至此终于，整个调用链完整了。我们终于完成了一次 RPC 调用。

与 Spring 集成

为了使我们的 client 能够易于使用我们需要考虑，定义一个自定义注解 @RpcInterface 当我们的项目接入 Spring 以后，Spring 扫描到这个注解之后，自动的通过我们的 ProxyFactory 创建代理对象，并存放在 spring 的 applicationContext 中。这样我们就可以通过 @Autowired 注解直接注入使用了。

1. Demo
   api

@RpcInterface
public interface IHelloService {
String sayHi(String name);
}
server

IHelloSerivce 的实现：

@SpringBootApplication
public class Application {
public static void main(String[] args) throws InterruptedException {
ConfigurableApplicationContext context = SpringApplication.run(Application.class);
TcpService tcpService = context.getBean(TcpService.class);
tcpService.start();
}
}
`
client

@SpringBootApplication()
public class ClientApplication {
public static void main(String[] args) {
ConfigurableApplicationContext context = SpringApplication.run(ClientApplication.class);
IHelloService helloService = context.getBean(IHelloService.class);
System.out.println(helloService.sayHi("doudou"));
}
}
运行以后输出的结果：

Hello doudou

总结
终于我们实现了一个最简版的 RPC 远程调用的模块。

原文地址：http://blog.51cto.com/13952955/2287615