swa con code face != nco 語言 dom policy

微服務,已經是每個互聯網開發者必須掌握的一項技術。而 RPC 框架，是構成微服務最重要的組成部分之一。趁最近有時間。又看了看 dubbo 的源碼。dubbo 為了做到靈活和解耦，使用了大量的設計模式和 SPI機制，要看懂 dubbo 的代碼也不太容易。

按照《徒手擼框架》系列文章的套路，我還是會極簡的實現一個 RPC 框架。幫助大家理解 RPC 框架的原理。

廣義的來講一個完整的 RPC 包含了很多組件，包括服務發現，服務治理，遠程調用，調用鏈分析，網關等等。我將會慢慢的實現這些功能，這篇文章主要先講解的是 RPC 的基石，遠程調用 的實現。

相信，讀完這篇文章你也一定可以自己實現一個可以提供 RPC 調用的框架。

1. RPC 的調用過程
   通過一圖我們來了解一下 RPC 的調用過程，從宏觀上來看看到底一次 RPC 調用經過些什麽過程。

當一次調用開始：

client 會調用本地動態代理 proxy
這個代理會將調用通過協議轉序列化字節流
通過 netty 網絡框架，將字節流發送到服務端
服務端在受到這個字節流後，會根據協議，反序列化為原始的調用，利用反射原理調用服務方提供的方法
如果請求有返回值，又需要把結果根據協議序列化後，再通過 netty 返回給調用方

2. 框架概覽和技術選型
   看一看框架的組件:

clinet就是調用方。servive是服務的提供者。protocol包定義了通信協議。common包含了通用的一些邏輯組件。

技術選型項目使用 maven 作為包管理工具，json 作為序列化協議，使用spring boot管理對象的生命周期，netty作為 nio 的網路組件。所以要閱讀這篇文章，你需要對spring boot和netty有基本的了解。

下面就看看每個組件的具體實現：

3. protocol
   其實作為 RPC 的協議，需要考慮只有一個問題–就是怎麽把一次方法的調用，變成能夠被網絡傳輸的字節流。

首先我們需要定義方法的調用和返回兩個實體：

請求：

@Data
public class RpcRequest {
// 調用編號
private String requestId;
// 類名
private String className;

// 方法名
private String methodName;
// 請求參數的數據類型
private Class<?>[] parameterTypes;
// 請求的參數
private Object[] parameters;
}
結果：

@Data
public class RpcResponse {
// 調用編號
private String requestId;
// 拋出的異常
private Throwable throwable;
// 返回結果
private Object result;
}
確定了，需要序列化的對象，就要確定序列化的協議，實現兩個方法，序列化和反序列化兩個方法。

public interface Serialization {
<T> byte[] serialize(T obj);
<T> T deSerialize(byte[] data,Class<T> clz);
}
可選用的序列化的協議很多比如：

jdk 的序列化方法。（不推薦，不利於之後的跨語言調用）
json 可讀性強，但是序列化速度慢，體積大。
protobuf，kyro，Hessian 等都是優秀的序列化框架，也可按需選擇。
為了簡單和便於調試，我們就選擇 json 作為序列化協議，使用jackson作為 json 解析框架。

/**

• @author Zhengxin
  */
  public class JsonSerialization implements Serialization {
  private ObjectMapper objectMapper;
  public JsonSerialization(){
  this.objectMapper = new ObjectMapper();}
  @Override
  br/>}
  @Override
  try {
  return objectMapper.writeValueAsBytes(obj);
  } catch (JsonProcessingException e) {
  e.printStackTrace();
  }
  return null;}
  @Override
  br/>}
  @Override
  try {
  return objectMapper.readValue(data,clz);
  } catch (IOException e) {
  e.printStackTrace();
  }
  return null;
  }
  }
  因為 netty 支持自定義 coder 。所以只需要實現 ByteToMessageDecoder 和 MessageToByteEncoder 兩個接口。就解決了序列化的問題:

public class RpcDecoder extends ByteToMessageDecoder {
private Class<?> clz;
private Serialization serialization;
public RpcDecoder(Class<?> clz,Serialization serialization){
this.clz = clz;
this.serialization = serialization;}
@Override
br/>}
@Override
if(in.readableBytes() < 4){
return;
}
in.markReaderIndex();
int dataLength = in.readInt();
if (in.readableBytes() < dataLength) {
in.resetReaderIndex();
return;
}
byte[] data = new byte[dataLength];
in.readBytes(data);
Object obj = serialization.deSerialize(data, clz);
out.add(obj);
}
}
public class RpcEncoder extends MessageToByteEncoder {
private Class<?> clz;
private Serialization serialization;
public RpcEncoder(Class<?> clz, Serialization serialization){
this.clz = clz;
this.serialization = serialization;}
@Override
br/>}
@Override
if(clz != null){
byte[] bytes = serialization.serialize(msg);
out.writeInt(bytes.length);
out.writeBytes(bytes);
}
}
}
至此，protocol 就實現了，我們就可以把方法的調用和結果的返回，轉換為一串可以在網絡中傳輸的 byte[] 數組了。

4. 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 的反射機制。

5. 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 就能獲取，相應的返回結果。

/**

• @author Zhengxin
  */
  public class ClientHandler extends ChannelDuplexHandler {
  // 使用 map 維護 id 和 Future 的映射關系，在多線程環境下需要使用線程安全的容器
  private final Map<String, DefaultFuture> futureMap = new ConcurrentHashMap<>();@Override
  br/>@Override
  if(msg instanceof RpcRequest){
  RpcRequest request = (RpcRequest) msg;
  // 寫數據的時候，增加映射
  futureMap.putIfAbsent(request.getRequestId(),new DefaultFuture());
  }
  super.write(ctx, msg, promise);}
  @Override
  br/>}
  @Override
  if(msg instanceof RpcResponse){
  RpcResponse response = (RpcResponse) msg;
  // 獲取數據的時候 將結果放入 future 中
  DefaultFuture defaultFuture = futureMap.get(response.getRequestId());
  defaultFuture.setResponse(response);
  }
  super.channelRead(ctx, msg);
  }
  public RpcResponse getRpcResponse(String requestId){
  try {
  // 從 future 中獲取真正的結果。
  DefaultFuture defaultFuture = futureMap.get(requestId);
  return defaultFuture.getResponse(10);
  }finally {
  // 完成後從 map 中移除。
  futureMap.remove(requestId);
  }
  }
  }
  這裏沒有繼承 server 中的 InboundHandler 而使用了 ChannelDuplexHandler。顧名思義就是在寫入和讀取數據的時候，都會觸發相應的方法。寫入的時候在 Map 中保存 ID 和 Future。讀到數據的時候從 Map 中取出 Future 並將結果放入 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 框架中動態代理的作用就是徹底替換原有方法，直接調用遠程方法。

代理工廠類：

public class ProxyFactory {@SuppressWarnings("unchecked")
br/>@SuppressWarnings("unchecked")
return (T) Proxy.newProxyInstance(
interfaceClass.getClassLoader(),
new Class<?>[]{interfaceClass},
new RpcInvoker<T>(interfaceClass)
);
}
}
當 proxyFactory 生成的類被調用的時候，就會執行 RpcInvoker 方法。

public class RpcInvoker<T> implements InvocationHandler {
private Class<T> clz;
public RpcInvoker(Class<T> clz){
this.clz = clz;}
@Override
br/>}
@Override
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);
return Transporters.send(request).getResult();
}
}
看到這個 invoke 方法，主要三個作用，

生成 RequestId。
拼裝 RpcRequest。
調用 Transports 發送請求，獲取結果。
至此終於，整個調用鏈完整了。我們終於完成了一次 RPC 調用。

與 Spring 集成

為了使我們的 client 能夠易於使用我們需要考慮，定義一個自定義註解 @RpcInterface 當我們的項目接入 Spring 以後，Spring 掃描到這個註解之後，自動的通過我們的 ProxyFactory 創建代理對象，並存放在 spring 的 applicationContext 中。這樣我們就可以通過 @Autowired 註解直接註入使用了。

@Target({ElementType.TYPE})@Retention(RetentionPolicy.RUNTIME)
br/>@Retention(RetentionPolicy.RUNTIME)
}

@Configuration@Slf4j
br/>@Slf4j
private ApplicationContext applicationContext;@Override
br/>@Override
this.applicationContext = applicationContext;}
@Override
br/>}
@Override
Reflections reflections = new Reflections("com.xilidou");
DefaultListableBeanFactory beanFactory = (DefaultListableBeanFactory) applicationContext.getAutowireCapableBeanFactory();
// 獲取 @RpcInterfac 標註的接口
Set<Class<?>> typesAnnotatedWith = reflections.getTypesAnnotatedWith(RpcInterface.class);
for (Class<?> aClass : typesAnnotatedWith) {
// 創建代理對象，並註冊到 spring 上下文。
beanFactory.registerSingleton(aClass.getSimpleName(),ProxyFactory.create(aClass));
}
log.info("afterPropertiesSet is {}",typesAnnotatedWith);
}
}
終於我們最簡單的 RPC 框架就開發完了。下面可以測試一下。

6. Demo
   api

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

IHelloSerivce 的實現：

@Service@Slf4j
br/>@Slf4j
br/>@Override
log.info(name);
return "Hello " + name;
}
}
啟動服務：

@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 遠程調用的模塊。

徒手擼框架--實現 RPC 遠程調用