fix: harden RPC timeout, threading and configuration (#6)

* fix: support primitive RPC parameter and return types

* fix: deserialize primitive RPC return types

* fix: reuse RpcClient in JDK proxies

* fix: unify RPC configuration overrides

* config: add timeout and business executor settings

* fix: expose core reliability settings in Spring Boot starter

* fix: propagate Spring Boot RPC settings to core

* fix: add RPC request timeout and pending cleanup

* fix: offload RPC business calls from Netty event loop

* fix: reuse server handler in auto protocol mode

* fix: add bounded business executor to Netty server

* test: cover RPC configuration overrides

* test: cover primitive RPC return values

* test: verify JDK proxy reuses RpcClient

* test: cover primitive parameters and business executor

* ci: avoid host cgroup namespace for Nacos

* ci: decouple tests from external Nacos service

* ci: include reactor dependencies in unit tests

Author: xiaoyumuxi

.github/workflows/ci.yml

@@ -10,23 +10,6 @@ jobs:
   build-and-test:
     runs-on: ubuntu-24.04
 
-    services:
-      nacos:
-        image: nacos/nacos-server:v2.4.3-slim
-        env:
-          MODE: standalone
-          JAVA_OPT: "-Djdk.attach.allowAttachSelf=true"
-        ports:
-          - 8848:8848
-          - 9848:9848
-        options: >-
-          --health-cmd="curl -f http://localhost:8848/nacos/v1/console/health/readiness"
-          --health-interval=10s
-          --health-timeout=5s
-          --health-retries=15
-          --health-start-period=60s
-          --cgroupns=host
-
     steps:
       - name: Checkout code
         uses: actions/checkout@v4
@@ -42,12 +25,10 @@ jobs:
         run: mvn clean package -DskipTests
 
       - name: Run unit tests
-        run: mvn test -pl rpc-core,rpc-transport-netty
+        run: mvn test -pl rpc-core,rpc-transport-netty -am
 
       - name: Run integration tests
         run: mvn test -pl rpc-consumer -am -Dtest=FullIntegrationTest
-        env:
-          RPC_REGISTRY: local
 
       - name: Check test coverage
         run: mvn jacoco:report

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

@@ -1,18 +1,29 @@
 package com.xiaoyu.rpc.core.client;
 
+import com.google.protobuf.ByteString;
 import com.xiaoyu.rpc.common.serialization.Serializer;
 import com.xiaoyu.rpc.common.serialization.SerializerCode;
 import com.xiaoyu.rpc.common.vo.RpcRequest;
-import com.google.protobuf.ByteString;
 import com.xiaoyu.rpc.core.config.RpcConfig;
 
 import java.lang.reflect.InvocationHandler;
 import java.lang.reflect.Method;
 import java.lang.reflect.Proxy;
+import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
 
 public class JdkProxyFactory implements ProxyFactory {
 
+    private final RpcClient rpcClient;
+
+    public JdkProxyFactory() {
+        this(new RpcClient());
+    }
+
+    JdkProxyFactory(RpcClient rpcClient) {
+        this.rpcClient = Objects.requireNonNull(rpcClient, "rpcClient");
+    }
+
     @Override
     @SuppressWarnings("unchecked")
     public <T> T getProxy(Class<T> clazz) {
@@ -34,7 +45,6 @@ public Object invoke(Object proxy, Method method, Object[] args) throws Throwabl
                         }
 
                         if (args != null) {
-                            // 获取配置的序列化器
                             Serializer serializer = SerializerCode
                                     .getSerializerByCode(RpcConfig.getInstance().getSerializerCode());
                             for (Object arg : args) {
@@ -44,7 +54,7 @@ public Object invoke(Object proxy, Method method, Object[] args) throws Throwabl
                         }
 
                         RpcRequest request = builder.build();
-                        CompletableFuture<Object> future = new RpcClient().sendRequest(request, method.getReturnType());
+                        CompletableFuture<Object> future = rpcClient.sendRequest(request, method.getReturnType());
                         // 如果业务接口声明的返回类型是异步的，直接返回 Future；否则阻塞等待结果
                         if (CompletableFuture.class.isAssignableFrom(method.getReturnType())) {
                             return future;

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

@@ -1,14 +1,19 @@
 package com.xiaoyu.rpc.core.client;
 
 import com.xiaoyu.rpc.common.extension.ExtensionLoader;
+import com.xiaoyu.rpc.common.serialization.Serializer;
+import com.xiaoyu.rpc.common.serialization.SerializerCode;
 import com.xiaoyu.rpc.common.vo.RpcRequest;
+import com.xiaoyu.rpc.common.vo.RpcResponse;
 import com.xiaoyu.rpc.core.config.RpcConfig;
 import com.xiaoyu.rpc.core.registry.ServiceDiscovery;
 import com.xiaoyu.rpc.core.transport.Transport;
 import com.xiaoyu.rpc.core.transport.TransportClient;
+import com.xiaoyu.rpc.core.util.TypeUtils;
 
 import java.net.InetSocketAddress;
 import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
 
 public class RpcClient {
 
@@ -31,37 +36,40 @@ public RpcClient() {
         this.serviceDiscovery = Objects.requireNonNull(serviceDiscovery, "serviceDiscovery");
     }
 
-    public java.util.concurrent.CompletableFuture<Object> sendRequest(RpcRequest request, Class<?> returnType) {
+    public CompletableFuture<Object> sendRequest(RpcRequest request, Class<?> returnType) {
         try {
             // 先做一次服务发现（同步查找，通常会命中本地缓存）
             InetSocketAddress address = serviceDiscovery.lookupService(request.getInterfaceName());
 
             if (address == null) {
-                java.util.concurrent.CompletableFuture<Object> future = new java.util.concurrent.CompletableFuture<>();
+                CompletableFuture<Object> future = new CompletableFuture<>();
                 future.completeExceptionally(new RuntimeException("未发现服务: " + request.getInterfaceName()));
                 return future;
             }
 
             // 交给传输层发送，返回异步 Future
-            java.util.concurrent.CompletableFuture<Object> transportFuture = transportClient.sendRequest(request,
-                    address);
+            CompletableFuture<Object> transportFuture = transportClient.sendRequest(request, address);
 
             // 在回调里把响应体反序列化成目标返回类型
             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;
-
-                    byte[] data = response.getData().toByteArray();
+                if (!(result instanceof RpcResponse)) {
+                    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);
+                RpcResponse response = (RpcResponse) result;
+                if (returnType == void.class || returnType == Void.class) {
+                    return null;
                 }
-                throw new RuntimeException("Unexpected response type: " + result.getClass());
+
+                byte[] data = response.getData().toByteArray();
+                Serializer serializer = SerializerCode
+                        .getSerializerByCode(RpcConfig.getInstance().getSerializerCode());
+                Class<?> deserializeType = TypeUtils.wrapPrimitive(returnType);
+                return serializer.deserialize(data, deserializeType);
             });
 
         } catch (Exception e) {
-            java.util.concurrent.CompletableFuture<Object> future = new java.util.concurrent.CompletableFuture<>();
+            CompletableFuture<Object> future = new CompletableFuture<>();
             future.completeExceptionally(e);
             return future;
         }