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

時間  2019-11-22
標籤 徒手 框架 實現 rpc 遠程 調用 欄目 微服務 简体版
原文   原文鏈接

原文連接java

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

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

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

相信，讀完這篇文章你也必定能夠本身實現一個能夠提供 RPC 調用的框架。編程

1. RPC 的調用過程

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

當一次調用開始：設計模式

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

2. 框架概覽和技術選型

看一看框架的組件:api

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

技術選型項目使用 maven 做爲包管理工具，json 做爲序列化協議，使用spring boot管理對象的生命週期，netty 做爲 nio 的網路組件。因此要閱讀這篇文章，你須要對spring boot和netty有基本的瞭解。promise

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

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
    public <T> byte[] serialize(T obj) {
        try {
            return objectMapper.writeValueAsBytes(obj);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
        return null;
    }

    @Override
    public <T> T deSerialize(byte[] data, Class<T> 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<Object> out) throws Exception {
        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
    protected void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) throws Exception {
        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 進行開發。關鍵代碼以下：

  1. netty 是基於 Reacotr 模型的。因此須要初始化兩組線程 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;
    }
複製代碼
  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<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 的吞吐量，可提供的思路有如下幾種：

  1. 使用對象池：創建多個 client 之後保存在對象池中。可是代碼的複雜度和維護 client 的成本會很高。

  2. 儘量的複用 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
    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> T create(Class<T> interfaceClass){
        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
    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<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
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 遠程調用的模塊。只是包含最最基礎的遠程調用功能。

若是你對這個項目感興趣，歡迎你與我聯繫，爲這個框架貢獻代碼。

老規矩 Github 地址：DouPpc

徒手擼框架系列文章地址：

徒手擼框架--實現IoC

徒手擼框架--實現Aop

徒手擼框架--高併發環境下的請求合併

歡迎關注個人微信公衆號：

二維碼
相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程