從零開始實現簡單 RPC 框架 8：網絡通信之 Request-Response 模型

Netty 在服務端與客戶端的網絡通信中，使用的是異步雙向通信(雙工)的方式，即客戶端和服務端可以相互主動發請求給對方，發消息后不會同步等響應。這樣就會有一下問題：

1. 如何識別消息是請求還是響應？
2. 請求如何正確對應到響應？

1. 如何識別消息是請求還是響應

為了識別消息類型是請求或者響應，我們在消息中加入了 messageType 的屬性，在上文我們也提到，這個消息類型在自定義協議的頭部，他有幾種類型：請求、響應、心跳，我們先來說說請求、響應。

public enum MessageType {
    /**
     * 普通請求
     */
    REQUEST((byte) 1),

    /**
     * 普通響應
     */
    RESPONSE((byte) 2),

    /**
     * 心跳
     */
    HEARTBEAT((byte) 3),
    ;
    private final byte value;
}

請求(Request)的核心字段如下：

public class RpcRequest {
    /**
     * 接口名
     */
    private String interfaceName;
    /**
     * 方法名
     */
    private String methodName;
    /**
     * 參數列表
     */
    private Object[] params;
    /**
     * 參數類型列表
     */
    private Class<?>[] paramTypes;
    /**
     * 接口版本
     */
    private String version;
}

響應(Response)的核心字段如下：

public class RpcResponse<T> {
    /**
     * 請求id
     */
    private long requestId;
    /**
     * 響應碼
     */
    private Integer code;
    /**
     * 提示消息
     */
    private String message;
    /**
     * 響應數據
     */
    private T data;
}

發送消息的時候，按照消息類型和結構體，將數據組裝好，寫到 channel 即可。接收消息則要先解碼，從消息頭拿到消息類型，根據消息類型來反序列化到對應的結構體。

2. 請求如何正確對應到響應

流程圖如下：

有幾個關鍵點：

1. 客戶端請求之后拿到 Future
2. 有一個 Map 存放未響應的請求，Key: RequestId，Value: Future
3. 服務端響應的數據中，包含了客戶端的 RequestId，這是對應的關鍵
4. 響應的結果會被 NettyClientHandler.channelRead0 監聽到，根據響應的 RequestId 取出對應的 Future
5. 將結果寫到對應的 Future 中
6. 客戶端通過 future.get() 獲取到數據

1) 客戶端發請求

代碼如下：

public class NettyInvoker extends AbstractInvoker {

    private final NettyClient nettyClient = NettyClient.getInstance();

    @Override
    protected RpcResult doInvoke(RpcRequest request, URL selected) throws RpcException {
        // 獲取 Channel
        Channel channel = nettyClient.getChannel(socketAddress);
        // 構造一個空 Future
        CompletableFuture<RpcResponse<?>> resultFuture = new CompletableFuture<>();
        // 構建 RPC 消息，此處會構建 requestId
        RpcMessage rpcMessage = buildRpcMessage(request);
        // 將 request 和 Future 對應放到 Map 中
        UnprocessedRequests.put(rpcMessage.getRequestId(), resultFuture);
        // 發出請求
        channel.writeAndFlush(rpcMessage);
        // 返回結果
        return new AsyncResult(resultFuture);
    }
    // ...
}

返回的 AsyncResult 只是 future 的包裝。

public class AsyncResult implements RpcResult {

    private final CompletableFuture<?> future;

    public AsyncResult(CompletableFuture<?> future) {
        this.future = future;
    }
}

2) 請求暫存

這個存儲未響應的請求在 ccx-rpc 中是 UnprocessedRequests 類在管理：

public class UnprocessedRequests {
    private static final Map<Long, CompletableFuture<RpcResponse<?>>> FUTURE_MAP = new ConcurrentHashMap<>();

    public static void put(long requestId, CompletableFuture<RpcResponse<?>> future) {
        FUTURE_MAP.put(requestId, future);
    }
}

3) 服務端響應數據監聽

使用 Netty 的 Handler 監聽服務端響應的數據，當有數據響應，則調用 UnprocessedRequests.complete 寫入。

public class NettyClientHandler extends SimpleChannelInboundHandler<RpcMessage> {
    @Override
    protected void channelRead0(ChannelHandlerContext context, RpcMessage requestMsg) {
        RpcResponse<?> response = (RpcResponse<?>) requestMsg.getData();
        UnprocessedRequests.complete(response);
    }
}

UnprocessedRequests.complete 實際上就是找出並刪除對應的請求，然后將數據寫入：future.complete(rpcResponse)

public class UnprocessedRequests {
    private static final Map<Long, CompletableFuture<RpcResponse<?>>> FUTURE_MAP = new ConcurrentHashMap<>();

    /**
     * 完成響應
     *
     * @param rpcResponse 響應內容
     */
    public static void complete(RpcResponse<?> rpcResponse) {
        CompletableFuture<RpcResponse<?>> future = FUTURE_MAP.remove(rpcResponse.getRequestId());
        if (future != null) {
            future.complete(rpcResponse);
        } else {
            throw new IllegalStateException("future is null. rpcResponse=" + JSONUtil.toJsonStr(rpcResponse));
        }
    }
}

最后通過 AsyncResult.getData 可以獲取到數據。

public class AsyncResult implements RpcResult {

    private final CompletableFuture<?> future;

    public AsyncResult(CompletableFuture<?> future) {
        this.future = future;
    }

    @Override
    public Object getData() {
        try {
            return future.get();
        } catch (InterruptedException | ExecutionException e) {
            log.error("getData error.", e);
        }
        return null;
    }
}

總結

Netty 網絡通信是異步雙工的，我們需要用正確 Request-Response 模型讓客戶端和服務端正確交互。

1. 如何區分請求或響應？
   在消息中，可以加入 messageType 字段用來區分是請求或者響應。
2. 如何把請求和響應對應？
   發出的請求需要用 RequestId 標記並用 Map 存起來。服務端收到請求之后，將 RequestId 原封不動寫到響應結果中。客戶端收到響應結果后，拿出 RequestId 找到對應的 Future 並寫入結果。

ccx-rpc 代碼已經開源
Github：https://github.com/chenchuxin/ccx-rpc
Gitee：https://gitee.com/imccx/ccx-rpc