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

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

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

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

1. RPC 的調用過程

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

當一次調用開始：

1. client 會調用本地動態代理 proxy
2. 這個代理會將調用通過協議轉序列化字節流
3. 通過 netty 網絡框架，將字節流發送到服務端
4. 服務端在受到這個字節流後，會根據協議，反序列化為原始的調用，利用反射原理調用服務方提供的方法
5. 如果請求有返回值，又需要把結果根據協議序列化後，再通過 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 {
  byte[] serialize(T obj);
  T deSerialize(byte[] data,Class 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
 public  byte[] serialize(T obj) {
 try {
 return objectMapper.writeValueAsBytes(obj);
 } catch (JsonProcessingException e) {
 e.printStackTrace();
 }
 return null;
 }
 @Override
 public  T deSerialize(byte[] data, Class clz) {
 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
 protected void decode(ChannelHandlerContext ctx, ByteBuf in, List

至此，protocol 就實現了，我們就可以把方法的調用和結果的返回，轉換為一串可以在網絡中傳輸的 byte[] 數組了。

4. server

server 是負責處理客戶端請求的組件。在互聯網高併發的環境下，使用 Nio 非阻塞的方式可以相對輕鬆的應付高併發的場景。netty 是一個優秀的 Nio 處理框架。Server 的關鍵代碼如下：

1. 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, Object> tcpChannelOptions = tcpChannelOptions();
 Set> keySet = tcpChannelOptions.keySet();
 for (@SuppressWarnings("rawtypes") ChannelOption option : keySet) {
 serverBootstrap.option(option, tcpChannelOptions.get(option));
 }
 return serverBootstrap;
 }

1. 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);

1. 實現具體的 ServerHandler 用於處理真正的調用。

ServerHandler 繼承 SimpleChannelInboundHandler。簡單來說這個 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 的吞吐量，可提供的思路有以下幾種：

1. 使用對象池：建立多個 client 以後保存在對象池中。但是代碼的複雜度和維護 client 的成本會很高。
2. 儘可能的複用 netty 中的 channel。
3. 之前你可能注意到，為什麼要在 RpcRequest 和 RpcResponse 中增加一個 ID。因為 netty 中的 channel 是會被多個線程使用的。當一個結果異步的返回後，你並不知道是哪個線程返回的。這個時候就可以考慮利用一個 Map，建立一個 ID 和 Future 映射。這樣請求的線程只要使用對應的 ID 就能獲取，相應的返回結果。

/**
 * @author Zhengxin
 */
public class ClientHandler extends ChannelDuplexHandler {
 // 使用 map 維護 id 和 Future 的映射關係，在多線程環境下需要使用線程安全的容器
 private final Map futureMap = new ConcurrentHashMap<>();
 @Override
 public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
 if(msg instanceof RpcRequest){
 RpcRequest request = (RpcRequest) msg;
 // 寫數據的時候，增加映射
 futureMap.putIfAbsent(request.getRequestId(),new DefaultFuture()); 

 }
 super.write(ctx, msg, promise);
 }
 @Override
 public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
 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")
 public static  T create(Class interfaceClass){
 return (T) Proxy.newProxyInstance(
 interfaceClass.getClassLoader(),
 new Class>[]{interfaceClass},
 new RpcInvoker(interfaceClass)
 );
 }
}

當 proxyFactory 生成的類被調用的時候，就會執行 RpcInvoker 方法。

public class RpcInvoker implements InvocationHandler {
 private Class clz;
 public RpcInvoker(Class clz){ 

 this.clz = clz;
 }
 @Override
 public 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);
 return Transporters.send(request).getResult();
 }
}

看到這個 invoke 方法，主要三個作用，

1. 生成 RequestId。
2. 拼裝 RpcRequest。
3. 調用 Transports 發送請求，獲取結果。

至此終於，整個調用鏈完整了。我們終於完成了一次 RPC 調用。

與 Spring 集成

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

@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
public @interface RpcInterface {
}

@Configuration
@Slf4j
public class RpcConfig implements ApplicationContextAware,InitializingBean {
	private ApplicationContext applicationContext;
	@Override
	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
		this.applicationContext = applicationContext;
	}
	@Override
	public void afterPropertiesSet() throws Exception {
		Reflections reflections = new Reflections("com.xilidou");
		DefaultListableBeanFactory beanFactory = (DefaultListableBeanFactory) applicationContext.getAutowireCapableBeanFactory();
 // 獲取 @RpcInterfac 標註的接口
		Set> 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
public class TestServiceImpl implements IHelloService {
 @Override
 public String sayHi(String name) {
 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 遠程調用的模塊。

閱讀更多 JAVA柯尼塞克丶 的文章

關鍵字: Google Java 遠程