feat: 将 RPC 请求处理改为异步 CompletableFuture 模式

Author: xiaoyumuxi

rpc-core/src/main/java/com/xiaoyu/rpc/core/client/ByteBuddyProxyFactory.java

@@ -50,6 +50,8 @@ public Object invoke(Object proxy, Method method, Object[] args) throws Throwabl
                             }
 
                             RpcRequest request = builder.build();
+                            // 注意：这里直接返回 Future。
+                            // 此时要求业务接口 Method 的返回类型必须是 CompletableFuture，否则会发生类型转换异常。
                             return rpcClient.sendRequest(request, method.getReturnType());
                         }
                     })).make().load(clazz.getClassLoader()).getLoaded().getConstructor().newInstance();

rpc-core/src/main/java/com/xiaoyu/rpc/core/client/RpcClient.java

@@ -25,40 +25,39 @@ public RpcClient() {
         this.transportClient = transport.createClient();
     }
 
-    public Object sendRequest(RpcRequest request, Class<?> returnType) {
+    public java.util.concurrent.CompletableFuture<Object> sendRequest(RpcRequest request, Class<?> returnType) {
         try {
-            // 1. 服务发现
+            // 1. 服务发现 (同步查找，通常有本地缓存)
             InetSocketAddress address = serviceDiscovery.lookupService(request.getInterfaceName());
 
             if (address == null) {
-                throw new RuntimeException("未发现服务: " + request.getInterfaceName());
+                java.util.concurrent.CompletableFuture<Object> future = new java.util.concurrent.CompletableFuture<>();
+                future.completeExceptionally(new RuntimeException("未发现服务: " + request.getInterfaceName()));
+                return future;
             }
 
-            // 2. 使用传输层发送请求
-            // 注意: TransportClient 返回的是反序列化后的结果(RpcResponse)或者已经提取的数据
-            // 在 NettyTransportClient 实现中，我们返回了 RpcResponse 对象
-            Object result = transportClient.sendRequest(request, address);
+            // 2. 使用传输层发送请求 (返回的是异步 Future)
+            java.util.concurrent.CompletableFuture<Object> transportFuture = transportClient.sendRequest(request,
+                    address);
 
-            // 3. 处理结果 (这一步逻辑如果 NettyTransportClient 已经做了反序列化，这里可能有点冗余，但保持检查是好的)
-            if (result instanceof com.xiaoyu.rpc.common.vo.RpcResponse) {
-                com.xiaoyu.rpc.common.vo.RpcResponse response = (com.xiaoyu.rpc.common.vo.RpcResponse) result;
+            // 3. 异步处理结果 (链式调用 thenApply)
+            return transportFuture.thenApply(result -> {
+                if (result instanceof com.xiaoyu.rpc.common.vo.RpcResponse) {
+                    com.xiaoyu.rpc.common.vo.RpcResponse response = (com.xiaoyu.rpc.common.vo.RpcResponse) result;
 
-                // 数据已经在 TransportClient 中反序列化了吗？
-                // 查看 NettyTransportClient 代码:
-                // return rpcResponse; -> 它并没有反序列化 data 字段成 returnType
-                // 所以这里需要反序列化
+                    byte[] data = response.getData().toByteArray();
 
-                byte[] data = response.getData().toByteArray();
-
-                com.xiaoyu.rpc.common.serialization.Serializer serializer = com.xiaoyu.rpc.common.serialization.SerializerCode
-                        .getSerializerByCode(RpcConfig.getInstance().getSerializerCode());
-                return serializer.deserialize(data, returnType);
-            }
-
-            throw new RuntimeException("Unexpected response type: " + result.getClass());
+                    com.xiaoyu.rpc.common.serialization.Serializer serializer = com.xiaoyu.rpc.common.serialization.SerializerCode
+                            .getSerializerByCode(RpcConfig.getInstance().getSerializerCode());
+                    return serializer.deserialize(data, returnType);
+                }
+                throw new RuntimeException("Unexpected response type: " + result.getClass());
+            });
 
         } catch (Exception e) {
-            throw new RuntimeException("RPC请求发送失败", e);
+            java.util.concurrent.CompletableFuture<Object> future = new java.util.concurrent.CompletableFuture<>();
+            future.completeExceptionally(e);
+            return future;
         }
     }
 }

rpc-core/src/main/java/com/xiaoyu/rpc/core/transport/TransportClient.java

@@ -14,7 +14,7 @@ public interface TransportClient {
      *
      * @param request 请求对象
      * @param address 目标地址
-     * @return 响应结果 (CompletableFuture 或 直接结果)
+     * @return 响应结果 (CompletableFuture)
      */
-    Object sendRequest(RpcRequest request, InetSocketAddress address);
+    CompletableFuture<Object> sendRequest(RpcRequest request, InetSocketAddress address);
 }

rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/transport/netty/NettyTransportClient.java

@@ -53,7 +53,7 @@ protected void initChannel(SocketChannel ch) {
     }
 
     @Override
-    public Object sendRequest(RpcRequest request, InetSocketAddress address) {
+    public CompletableFuture<Object> sendRequest(RpcRequest request, InetSocketAddress address) {
         String protocolName = RpcConfig.getInstance().getProtocol();
 
         try {
@@ -73,13 +73,6 @@ public Object sendRequest(RpcRequest request, InetSocketAddress address) {
 
             // Generate ID and set to request
             String requestId = java.util.UUID.randomUUID().toString();
-            // Use reflection or builder if setter not available, but since we regenerated
-            // proto, we should use builder properly
-            // However, RpcRequest is immutable if generated by Proto?
-            // Wait, RpcRequest.java is generated. It has a Builder.
-            // But the signature of sendRequest takes RpcRequest, which is already built.
-            // We need to rebuild it with the ID.
-
             RpcRequest.Builder builder = request.toBuilder();
             builder.setRequestId(requestId);
             RpcRequest newRequest = builder.build();
@@ -88,25 +81,25 @@ public Object sendRequest(RpcRequest request, InetSocketAddress address) {
             clientHandler.addFuture(requestId, resultFuture);
 
             Protocol protocol = ProtocolFactory.getProtocol(protocolName);
-            // Pass the handler so protocol can ensure it's in pipeline if needed (double
-            // check)
             protocol.sendRequest(channel, newRequest, clientHandler);
 
-            Object result = resultFuture.get(5, TimeUnit.SECONDS);
-
-            if (result instanceof RpcResponse) {
-                RpcResponse rpcResponse = (RpcResponse) result;
-
-                if (!"Success".equals(rpcResponse.getMessage())) {
-                    throw new RuntimeException("服务端报错: " + rpcResponse.getMessage());
+            // 彻底移除 resultFuture.get()，直接返回异步 Future
+            return resultFuture.thenApply(result -> {
+                if (result instanceof RpcResponse) {
+                    RpcResponse rpcResponse = (RpcResponse) result;
+                    if (!"Success".equals(rpcResponse.getMessage())) {
+                        throw new RuntimeException("服务端报错: " + rpcResponse.getMessage());
+                    }
+                    return rpcResponse;
+                } else {
+                    throw new RuntimeException("服务端返回的不是 RpcResponse 类型");
                 }
-                return rpcResponse;
-
-            } else {
-                throw new RuntimeException("服务端返回的不是 RpcResponse 类型");
-            }
+            });
         } catch (Exception e) {
-            throw new RuntimeException("RPC请求发送失败", e);
+            log.error("RPC请求发起失败", e);
+            CompletableFuture<Object> future = new CompletableFuture<>();
+            future.completeExceptionally(e);
+            return future;
         }
     }
 }