diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 603b665..8c748b1 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -91,6 +91,46 @@ jobs: -Drpc.registry=local -Drpc.protocol=netty + # Performance data is observational only. Hosted Runner variance makes absolute + # latency/QPS unsuitable as a merge threshold, but every run publishes the raw + # samples and summary so regressions can be inspected over time. + - name: Run RPC performance snapshot + run: >- + mvn -B -ntp test + -pl rpc-consumer -am + -Dtest=RpcPerformanceSnapshotTest + -Dsurefire.failIfNoSpecifiedTests=false + -Drpc.registry=local + -Drpc.protocol=netty + -Drpc.serializer=kryo + -Drpc.perf.warmup=100 + -Drpc.perf.sequential-requests=200 + -Drpc.perf.concurrent-requests=1000 + -Drpc.perf.concurrency=16 + + - name: Publish RPC performance summary + if: always() + shell: bash + run: | + if [[ -f rpc-consumer/target/rpc-performance/summary.md ]]; then + cat rpc-consumer/target/rpc-performance/summary.md >> "$GITHUB_STEP_SUMMARY" + else + { + echo "# RPC CI Performance Snapshot" + echo + echo "No performance snapshot was generated. Check the test logs for the failure cause." + } >> "$GITHUB_STEP_SUMMARY" + fi + + - name: Upload RPC performance snapshot + if: always() + uses: actions/upload-artifact@v7 + with: + name: rpc-performance-snapshot + retention-days: 30 + if-no-files-found: warn + path: rpc-consumer/target/rpc-performance/ + - name: Upload integration test reports if: always() uses: actions/upload-artifact@v7 diff --git a/rpc-common/src/main/proto/rpc_meta.proto b/rpc-common/src/main/proto/rpc_meta.proto index c53164a..3ac21cb 100644 --- a/rpc-common/src/main/proto/rpc_meta.proto +++ b/rpc-common/src/main/proto/rpc_meta.proto @@ -1,42 +1,37 @@ syntax = "proto3"; -option java_package = "com.xiaoyu.rpc.common.vo"; // 生成 Java 类的包名 - -// 拆分生成多个 Java 文件，避免全部消息挤在一个类里 +option java_package = "com.xiaoyu.rpc.common.vo"; option java_multiple_files = true; -// 请求对象 +enum RpcStatusCode { + RPC_STATUS_UNSPECIFIED = 0; + SUCCESS = 1; + SERVICE_NOT_FOUND = 2; + METHOD_NOT_FOUND = 3; + INVALID_ARGUMENT = 4; + SERVER_BUSY = 5; + BUSINESS_ERROR = 6; + INTERNAL_ERROR = 7; + TIMEOUT = 8; + UNAVAILABLE = 9; + CLIENT_CLOSED = 10; +} + message RpcRequest { - // 对应 String interfaceName string interface_name = 1; - - // 对应 String methodName string method_name = 2; - - // 对应 Class[] paramTypes - // Proto存不了Class对象，只能存全类名(String)，比如 "java.lang.String" repeated string param_types = 3; - - // 对应 Object[] parameters - // Proto存不了Object，必须存成二进制(bytes)。 - // 这里用 repeated 表示数组 repeated bytes parameters = 4; - - // 请求ID，用于多路复用 string request_id = 5; + map metadata = 6; } -// 响应对象 message RpcResponse { - // 对应 Object data - // 同样无法存Object，只能存序列化后的二进制 bytes data = 1; - - // 对应 String message string message = 2; - - // 请求ID，用于多路复用 string request_id = 3; + RpcStatusCode status_code = 4; + string error_type = 5; } service GrpcService { diff --git a/rpc-consumer/src/test/java/com/xiaoyu/rpc/consumer/RpcPerformanceSnapshotTest.java b/rpc-consumer/src/test/java/com/xiaoyu/rpc/consumer/RpcPerformanceSnapshotTest.java new file mode 100644 index 0000000..c838576 --- /dev/null +++ b/rpc-consumer/src/test/java/com/xiaoyu/rpc/consumer/RpcPerformanceSnapshotTest.java @@ -0,0 +1,426 @@ +package com.xiaoyu.rpc.consumer; + +import com.google.protobuf.ByteString; +import com.xiaoyu.rpc.api.HelloService; +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.core.client.RpcClient; +import com.xiaoyu.rpc.core.config.RpcConfig; +import com.xiaoyu.rpc.core.observability.RpcMetricSide; +import com.xiaoyu.rpc.core.observability.RpcMetrics; +import com.xiaoyu.rpc.core.server.RpcServer; +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.time.Instant; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Locale; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assertions.fail; + +/** + * CI 中的轻量端到端性能快照。 + * + *

它的目标是生成可比较的观测数据，而不是用 GitHub Hosted Runner 的绝对性能值作为合并门槛。 + * JMH 仍然负责更严格的微基准；这里关注真实 RPC 主链路的延迟、吞吐和框架内置 metrics 是否一致。

+ */ +public class RpcPerformanceSnapshotTest { + + private static final int WARMUP_REQUESTS = Integer.getInteger("rpc.perf.warmup", 100); + private static final int SEQUENTIAL_REQUESTS = Integer.getInteger("rpc.perf.sequential-requests", 200); + private static final int CONCURRENT_REQUESTS = Integer.getInteger("rpc.perf.concurrent-requests", 1000); + private static final int CONCURRENCY = Integer.getInteger("rpc.perf.concurrency", 16); + private static final Path OUTPUT_DIR = Path.of("target", "rpc-performance"); + + public static class HelloServiceImpl implements HelloService { + @Override + public String sayHello(String name) { + return "Hello, " + name + "! (from Netty Server)"; + } + } + + @Test + void generatePerformanceSnapshot() throws Exception { + String registry = System.getProperty("rpc.registry", "local"); + String protocol = System.getProperty("rpc.protocol", "netty"); + String serializerName = System.getProperty("rpc.serializer", "kryo"); + int port = findFreePort(); + + System.setProperty("rpc.registry", registry); + System.setProperty("rpc.server-host", "127.0.0.1"); + System.setProperty("rpc.server-port", String.valueOf(port)); + System.setProperty("rpc.serializer", serializerName); + System.setProperty("rpc.transport", "netty"); + System.setProperty("rpc.protocol", protocol); + + RpcServer server = null; + try { + server = new RpcServer(); + server.register(HelloService.class, new HelloServiceImpl()); + server.start(); + + Serializer serializer = SerializerCode.getSerializerByCode(RpcConfig.getInstance().getSerializerCode()); + try (RpcClient client = new RpcClient()) { + warmUp(client, serializer); + + PhaseResult sequential = runSequential(client, serializer); + PhaseResult concurrent = runConcurrent(client, serializer); + + writeArtifacts(registry, protocol, serializerName, sequential, concurrent); + } + } finally { + if (server != null) { + server.close(); + } + RpcMetrics.getInstance().reset(); + System.clearProperty("rpc.server-host"); + System.clearProperty("rpc.server-port"); + System.clearProperty("rpc.transport"); + System.clearProperty("rpc.protocol"); + System.clearProperty("rpc.serializer"); + System.clearProperty("rpc.registry"); + } + } + + private static void warmUp(RpcClient client, Serializer serializer) throws Exception { + for (int i = 0; i < WARMUP_REQUESTS; i++) { + invokeOnce(client, serializer, "Warmup-" + i); + } + } + + private static PhaseResult runSequential(RpcClient client, Serializer serializer) throws Exception { + RpcMetrics.getInstance().reset(); + long[] samples = new long[SEQUENTIAL_REQUESTS]; + long phaseStart = System.nanoTime(); + + for (int i = 0; i < SEQUENTIAL_REQUESTS; i++) { + samples[i] = invokeOnce(client, serializer, "Sequential-" + i); + } + + long elapsedNanos = System.nanoTime() - phaseStart; + return createPhaseResult("sequential", SEQUENTIAL_REQUESTS, 1, elapsedNanos, samples); + } + + private static PhaseResult runConcurrent(RpcClient client, Serializer serializer) throws Exception { + RpcMetrics.getInstance().reset(); + long[] samples = new long[CONCURRENT_REQUESTS]; + ExecutorService executor = Executors.newFixedThreadPool(CONCURRENCY); + CountDownLatch startGate = new CountDownLatch(1); + CountDownLatch doneGate = new CountDownLatch(CONCURRENT_REQUESTS); + AtomicReference failure = new AtomicReference<>(); + + try { + for (int i = 0; i < CONCURRENT_REQUESTS; i++) { + final int requestIndex = i; + executor.execute(() -> { + try { + startGate.await(); + samples[requestIndex] = invokeOnce(client, serializer, "Concurrent-" + requestIndex); + } catch (Throwable t) { + failure.compareAndSet(null, t); + } finally { + doneGate.countDown(); + } + }); + } + + long phaseStart = System.nanoTime(); + startGate.countDown(); + boolean completed = doneGate.await(45, TimeUnit.SECONDS); + long elapsedNanos = System.nanoTime() - phaseStart; + + assertTrue(completed, "Concurrent performance phase should finish within 45 seconds"); + Throwable error = failure.get(); + if (error != null) { + fail("Concurrent RPC performance phase failed", error); + } + + return createPhaseResult("concurrent", CONCURRENT_REQUESTS, CONCURRENCY, elapsedNanos, samples); + } finally { + executor.shutdownNow(); + executor.awaitTermination(5, TimeUnit.SECONDS); + } + } + + private static long invokeOnce(RpcClient client, Serializer serializer, String name) throws Exception { + long startNanos = System.nanoTime(); + String result = (String) client + .sendRequest(buildRequest(name, serializer), String.class) + .get(5, TimeUnit.SECONDS); + long elapsedNanos = System.nanoTime() - startNanos; + + if (result == null || !result.contains(name)) { + throw new AssertionError("Unexpected RPC result for " + name + ": " + result); + } + return elapsedNanos; + } + + private static PhaseResult createPhaseResult(String name, int requests, int concurrency, + long elapsedNanos, long[] samples) { + RpcMetrics.Snapshot clientMetrics = RpcMetrics.getInstance().snapshot(RpcMetricSide.CLIENT); + RpcMetrics.Snapshot serverMetrics = RpcMetrics.getInstance().snapshot(RpcMetricSide.SERVER); + + assertMetrics("client", requests, clientMetrics); + assertMetrics("server", requests, serverMetrics); + + long[] sorted = samples.clone(); + Arrays.sort(sorted); + long sum = 0L; + for (long sample : samples) { + sum += sample; + } + + double elapsedSeconds = elapsedNanos / 1_000_000_000D; + double throughput = elapsedSeconds <= 0D ? 0D : requests / elapsedSeconds; + double averageMillis = requests == 0 ? 0D : nanosToMillis(sum / (double) requests); + + return new PhaseResult( + name, + requests, + concurrency, + elapsedNanos, + throughput, + nanosToMillis(sorted[0]), + averageMillis, + nanosToMillis(percentile(sorted, 0.50D)), + nanosToMillis(percentile(sorted, 0.95D)), + nanosToMillis(percentile(sorted, 0.99D)), + nanosToMillis(sorted[sorted.length - 1]), + clientMetrics, + serverMetrics, + samples.clone()); + } + + private static void assertMetrics(String side, int expectedRequests, RpcMetrics.Snapshot snapshot) { + assertEquals(expectedRequests, snapshot.totalRequests(), side + " total requests mismatch"); + assertEquals(expectedRequests, snapshot.successRequests(), side + " success requests mismatch"); + assertEquals(0, snapshot.failedRequests(), side + " failed requests should be zero"); + assertEquals(0, snapshot.timeoutRequests(), side + " timeout requests should be zero"); + assertEquals(0, snapshot.activeRequests(), side + " active requests should return to zero"); + } + + private static long percentile(long[] sorted, double percentile) { + int rank = (int) Math.ceil(percentile * sorted.length); + int index = Math.min(sorted.length - 1, Math.max(0, rank - 1)); + return sorted[index]; + } + + private static double nanosToMillis(double nanos) { + return nanos / 1_000_000D; + } + + private static RpcRequest buildRequest(String name, Serializer serializer) { + return RpcRequest.newBuilder() + .setInterfaceName(HelloService.class.getName()) + .setMethodName("sayHello") + .addParamTypes(String.class.getName()) + .addParameters(ByteString.copyFrom(serializer.serialize(name))) + .build(); + } + + private static int findFreePort() throws IOException { + try (java.net.ServerSocket socket = new java.net.ServerSocket(0, 1, + java.net.InetAddress.getByName("127.0.0.1"))) { + return socket.getLocalPort(); + } + } + + private static void writeArtifacts(String registry, String protocol, String serializer, + PhaseResult sequential, PhaseResult concurrent) throws IOException { + Files.createDirectories(OUTPUT_DIR); + Files.writeString(OUTPUT_DIR.resolve("performance.json"), + toJson(registry, protocol, serializer, sequential, concurrent), StandardCharsets.UTF_8); + Files.writeString(OUTPUT_DIR.resolve("summary.md"), + toMarkdown(registry, protocol, serializer, sequential, concurrent), StandardCharsets.UTF_8); + Files.writeString(OUTPUT_DIR.resolve("latency-samples.csv"), + toCsv(sequential, concurrent), StandardCharsets.UTF_8); + } + + private static String toJson(String registry, String protocol, String serializer, + PhaseResult sequential, PhaseResult concurrent) { + StringBuilder out = new StringBuilder(); + out.append("{\n"); + field(out, "generatedAt", Instant.now().toString(), true, 1); + field(out, "gitSha", envOrDefault("GITHUB_SHA", "local"), true, 1); + out.append(" \"environment\": {\n"); + field(out, "javaVersion", System.getProperty("java.version"), true, 2); + field(out, "os", System.getProperty("os.name") + " " + System.getProperty("os.arch"), true, 2); + numberField(out, "availableProcessors", Runtime.getRuntime().availableProcessors(), true, 2); + numberField(out, "maxHeapMb", Runtime.getRuntime().maxMemory() / (1024D * 1024D), false, 2); + out.append(" },\n"); + out.append(" \"config\": {\n"); + field(out, "registry", registry, true, 2); + field(out, "protocol", protocol, true, 2); + field(out, "serializer", serializer, true, 2); + numberField(out, "warmupRequests", WARMUP_REQUESTS, true, 2); + numberField(out, "sequentialRequests", SEQUENTIAL_REQUESTS, true, 2); + numberField(out, "concurrentRequests", CONCURRENT_REQUESTS, true, 2); + numberField(out, "concurrency", CONCURRENCY, false, 2); + out.append(" },\n"); + out.append(" \"phases\": [\n"); + appendPhaseJson(out, sequential, true); + appendPhaseJson(out, concurrent, false); + out.append(" ]\n"); + out.append("}\n"); + return out.toString(); + } + + private static void appendPhaseJson(StringBuilder out, PhaseResult phase, boolean comma) { + out.append(" {\n"); + field(out, "name", phase.name(), true, 3); + numberField(out, "requests", phase.requests(), true, 3); + numberField(out, "concurrency", phase.concurrency(), true, 3); + numberField(out, "durationMs", nanosToMillis(phase.elapsedNanos()), true, 3); + numberField(out, "throughputRps", phase.throughputRps(), true, 3); + out.append(" \"latencyMs\": {\n"); + numberField(out, "min", phase.minMillis(), true, 4); + numberField(out, "average", phase.averageMillis(), true, 4); + numberField(out, "p50", phase.p50Millis(), true, 4); + numberField(out, "p95", phase.p95Millis(), true, 4); + numberField(out, "p99", phase.p99Millis(), true, 4); + numberField(out, "max", phase.maxMillis(), false, 4); + out.append(" },\n"); + out.append(" \"clientMetrics\": "); + appendMetricsJson(out, phase.clientMetrics()); + out.append(",\n"); + out.append(" \"serverMetrics\": "); + appendMetricsJson(out, phase.serverMetrics()); + out.append("\n }"); + out.append(comma ? ",\n" : "\n"); + } + + private static void appendMetricsJson(StringBuilder out, RpcMetrics.Snapshot snapshot) { + out.append("{\"total\":").append(snapshot.totalRequests()) + .append(",\"success\":").append(snapshot.successRequests()) + .append(",\"failed\":").append(snapshot.failedRequests()) + .append(",\"timeout\":").append(snapshot.timeoutRequests()) + .append(",\"active\":").append(snapshot.activeRequests()) + .append(",\"averageLatencyMs\":").append(format(snapshot.averageLatencyMillis())) + .append(",\"maxLatencyMs\":").append(format(snapshot.maxLatencyMillis())) + .append('}'); + } + + private static String toMarkdown(String registry, String protocol, String serializer, + PhaseResult sequential, PhaseResult concurrent) { + StringBuilder out = new StringBuilder(); + out.append("# RPC CI Performance Snapshot\n\n"); + out.append("> Observational snapshot only. GitHub Hosted Runner performance varies; these values are not merge thresholds.\n\n"); + out.append("- Commit: `").append(envOrDefault("GITHUB_SHA", "local")).append("`\n"); + out.append("- Protocol: `").append(protocol).append("`\n"); + out.append("- Serializer: `").append(serializer).append("`\n"); + out.append("- Registry: `").append(registry).append("`\n"); + out.append("- Java: `").append(System.getProperty("java.version")).append("`\n"); + out.append("- CPU visible to JVM: `").append(Runtime.getRuntime().availableProcessors()).append("`\n\n"); + out.append("| Phase | Requests | Concurrency | Throughput req/s | Avg ms | P50 ms | P95 ms | P99 ms | Max ms |\n"); + out.append("| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |\n"); + appendPhaseMarkdown(out, sequential); + appendPhaseMarkdown(out, concurrent); + out.append("\n## Framework metrics\n\n"); + out.append("| Phase | Side | Total | Success | Failed | Timeout | Active | Metric avg ms | Metric max ms |\n"); + out.append("| --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: |\n"); + appendMetricsMarkdown(out, sequential, "CLIENT", sequential.clientMetrics()); + appendMetricsMarkdown(out, sequential, "SERVER", sequential.serverMetrics()); + appendMetricsMarkdown(out, concurrent, "CLIENT", concurrent.clientMetrics()); + appendMetricsMarkdown(out, concurrent, "SERVER", concurrent.serverMetrics()); + out.append("\nRaw per-request samples are available in `latency-samples.csv`; machine-readable totals are in `performance.json`.\n"); + return out.toString(); + } + + private static void appendPhaseMarkdown(StringBuilder out, PhaseResult phase) { + out.append("| ").append(phase.name()) + .append(" | ").append(phase.requests()) + .append(" | ").append(phase.concurrency()) + .append(" | ").append(format(phase.throughputRps())) + .append(" | ").append(format(phase.averageMillis())) + .append(" | ").append(format(phase.p50Millis())) + .append(" | ").append(format(phase.p95Millis())) + .append(" | ").append(format(phase.p99Millis())) + .append(" | ").append(format(phase.maxMillis())) + .append(" |\n"); + } + + private static void appendMetricsMarkdown(StringBuilder out, PhaseResult phase, String side, + RpcMetrics.Snapshot snapshot) { + out.append("| ").append(phase.name()) + .append(" | ").append(side) + .append(" | ").append(snapshot.totalRequests()) + .append(" | ").append(snapshot.successRequests()) + .append(" | ").append(snapshot.failedRequests()) + .append(" | ").append(snapshot.timeoutRequests()) + .append(" | ").append(snapshot.activeRequests()) + .append(" | ").append(format(snapshot.averageLatencyMillis())) + .append(" | ").append(format(snapshot.maxLatencyMillis())) + .append(" |\n"); + } + + private static String toCsv(PhaseResult... phases) { + StringBuilder out = new StringBuilder("phase,index,latency_ms\n"); + for (PhaseResult phase : phases) { + long[] samples = phase.samplesNanos(); + for (int i = 0; i < samples.length; i++) { + out.append(phase.name()).append(',') + .append(i).append(',') + .append(format(nanosToMillis(samples[i]))) + .append('\n'); + } + } + return out.toString(); + } + + private static void field(StringBuilder out, String name, String value, boolean comma, int indent) { + out.append(" ".repeat(indent)) + .append('\"').append(name).append("\": \"") + .append(escapeJson(value)).append('\"') + .append(comma ? ",\n" : "\n"); + } + + private static void numberField(StringBuilder out, String name, double value, boolean comma, int indent) { + out.append(" ".repeat(indent)) + .append('\"').append(name).append("\": ") + .append(format(value)) + .append(comma ? ",\n" : "\n"); + } + + private static String escapeJson(String value) { + return value.replace("\\", "\\\\").replace("\"", "\\\""); + } + + private static String format(double value) { + return String.format(Locale.ROOT, "%.3f", value); + } + + private static String envOrDefault(String name, String fallback) { + String value = System.getenv(name); + return value == null || value.isBlank() ? fallback : value; + } + + private record PhaseResult( + String name, + int requests, + int concurrency, + long elapsedNanos, + double throughputRps, + double minMillis, + double averageMillis, + double p50Millis, + double p95Millis, + double p99Millis, + double maxMillis, + RpcMetrics.Snapshot clientMetrics, + RpcMetrics.Snapshot serverMetrics, + long[] samplesNanos) { + } +} diff --git a/rpc-core/src/main/java/com/xiaoyu/rpc/core/client/RpcClient.java b/rpc-core/src/main/java/com/xiaoyu/rpc/core/client/RpcClient.java index deb0bea..b659ddf 100644 --- a/rpc-core/src/main/java/com/xiaoyu/rpc/core/client/RpcClient.java +++ b/rpc-core/src/main/java/com/xiaoyu/rpc/core/client/RpcClient.java @@ -5,7 +5,14 @@ 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.common.vo.RpcStatusCode; import com.xiaoyu.rpc.core.config.RpcConfig; +import com.xiaoyu.rpc.core.exception.RpcException; +import com.xiaoyu.rpc.core.interceptor.RpcInterceptorRegistry; +import com.xiaoyu.rpc.core.interceptor.RpcInvocationContext; +import com.xiaoyu.rpc.core.interceptor.RpcSide; +import com.xiaoyu.rpc.core.observability.RpcMetricSide; +import com.xiaoyu.rpc.core.observability.RpcMetrics; import com.xiaoyu.rpc.core.registry.ServiceDiscovery; import com.xiaoyu.rpc.core.transport.Transport; import com.xiaoyu.rpc.core.transport.TransportClient; @@ -13,6 +20,7 @@ import java.net.InetSocketAddress; import java.util.Objects; +import java.util.UUID; import java.util.concurrent.CompletableFuture; import java.util.concurrent.atomic.AtomicBoolean; @@ -37,41 +45,119 @@ public RpcClient() { } public CompletableFuture sendRequest(RpcRequest request, Class returnType) { + Objects.requireNonNull(request, "request"); + Objects.requireNonNull(returnType, "returnType"); + + RpcRequest preparedRequest = ensureRequestId(request); + RpcInvocationContext context = new RpcInvocationContext(RpcSide.CLIENT, preparedRequest); + RpcMetrics.CallTimer timer = RpcMetrics.getInstance().startCall(RpcMetricSide.CLIENT); + if (closed.get()) { - return CompletableFuture.failedFuture(new IllegalStateException("RpcClient 已关闭")); + RpcException error = new RpcException(RpcStatusCode.CLIENT_CLOSED, "RpcClient 已关闭"); + timer.failure(error.getStatusCode()); + RpcInterceptorRegistry.onError(context, error); + return CompletableFuture.failedFuture(error); } + final RpcRequest interceptedRequest; try { - InetSocketAddress address = serviceDiscovery.lookupService(request.getInterfaceName()); + interceptedRequest = RpcInterceptorRegistry.before(context, preparedRequest); + } catch (Exception e) { + RpcException error = toRpcException(e, RpcStatusCode.INTERNAL_ERROR, "客户端拦截器执行失败"); + timer.failure(error.getStatusCode()); + RpcInterceptorRegistry.onError(context, error); + return CompletableFuture.failedFuture(error); + } + final InetSocketAddress address; + try { + address = serviceDiscovery.lookupService(interceptedRequest.getInterfaceName()); if (address == null) { - return CompletableFuture.failedFuture( - new RuntimeException("未发现服务: " + request.getInterfaceName())); + throw new RpcException( + RpcStatusCode.UNAVAILABLE, + "未发现服务: " + interceptedRequest.getInterfaceName()); } + } catch (Exception e) { + RpcException error = toRpcException(e, RpcStatusCode.UNAVAILABLE, + "服务发现失败: " + interceptedRequest.getInterfaceName()); + timer.failure(error.getStatusCode()); + RpcInterceptorRegistry.onError(context, error); + return CompletableFuture.failedFuture(error); + } - CompletableFuture transportFuture = transportClient.sendRequest(request, address); + CompletableFuture resultFuture; + try { + CompletableFuture transportFuture = transportClient.sendRequest(interceptedRequest, address); + resultFuture = transportFuture.thenApply(result -> handleResponse(result, returnType, context)); + } catch (Exception e) { + resultFuture = CompletableFuture.failedFuture(e); + } - return transportFuture.thenApply(result -> { - if (!(result instanceof RpcResponse)) { - String actualType = result == null ? "null" : result.getClass().getName(); - throw new RuntimeException("Unexpected response type: " + actualType); - } + return resultFuture.whenComplete((result, throwable) -> { + if (throwable == null) { + timer.success(); + } else { + Throwable cause = RpcException.unwrap(throwable); + timer.failure(RpcException.statusOf(cause)); + RpcInterceptorRegistry.onError(context, cause); + } + }); + } - RpcResponse response = (RpcResponse) result; - if (returnType == void.class || returnType == Void.class) { - return null; - } + private Object handleResponse(Object result, Class returnType, RpcInvocationContext context) { + if (!(result instanceof RpcResponse response)) { + String actualType = result == null ? "null" : result.getClass().getName(); + throw new RpcException(RpcStatusCode.INTERNAL_ERROR, "Unexpected response type: " + actualType); + } - byte[] data = response.getData().toByteArray(); - Serializer serializer = SerializerCode - .getSerializerByCode(RpcConfig.getInstance().getSerializerCode()); - Class deserializeType = TypeUtils.wrapPrimitive(returnType); - return serializer.deserialize(data, deserializeType); - }); + RpcStatusCode statusCode = effectiveStatus(response); + if (statusCode != RpcStatusCode.SUCCESS) { + throw new RpcException( + statusCode, + response.getMessage().isEmpty() ? statusCode.name() : response.getMessage(), + response.getErrorType()); + } - } catch (Exception e) { - return CompletableFuture.failedFuture(e); + RpcInterceptorRegistry.after(context, response); + + if (returnType == void.class || returnType == Void.class) { + return null; + } + + byte[] data = response.getData().toByteArray(); + Serializer serializer = SerializerCode + .getSerializerByCode(RpcConfig.getInstance().getSerializerCode()); + Class deserializeType = TypeUtils.wrapPrimitive(returnType); + return serializer.deserialize(data, deserializeType); + } + + private static RpcStatusCode effectiveStatus(RpcResponse response) { + if (response.getStatusCode() != RpcStatusCode.RPC_STATUS_UNSPECIFIED) { + return response.getStatusCode(); + } + return "Success".equals(response.getMessage()) + ? RpcStatusCode.SUCCESS + : RpcStatusCode.INTERNAL_ERROR; + } + + private static RpcRequest ensureRequestId(RpcRequest request) { + if (!request.getRequestId().isEmpty()) { + return request; } + return request.toBuilder().setRequestId(UUID.randomUUID().toString()).build(); + } + + private static RpcException toRpcException(Throwable throwable, RpcStatusCode fallback, String message) { + Throwable cause = RpcException.unwrap(throwable); + if (cause instanceof RpcException rpcException) { + return rpcException; + } + return new RpcException(fallback, message + ": " + safeMessage(cause), cause); + } + + private static String safeMessage(Throwable throwable) { + String message = throwable.getMessage(); + return message == null || message.isEmpty() ? throwable.getClass().getSimpleName() : message; } @Override diff --git a/rpc-core/src/main/java/com/xiaoyu/rpc/core/exception/RpcException.java b/rpc-core/src/main/java/com/xiaoyu/rpc/core/exception/RpcException.java new file mode 100644 index 0000000..d924d01 --- /dev/null +++ b/rpc-core/src/main/java/com/xiaoyu/rpc/core/exception/RpcException.java @@ -0,0 +1,59 @@ +package com.xiaoyu.rpc.core.exception; + +import com.xiaoyu.rpc.common.vo.RpcStatusCode; + +import java.util.concurrent.CompletionException; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeoutException; + +public class RpcException extends RuntimeException { + + private final RpcStatusCode statusCode; + private final String remoteErrorType; + + public RpcException(RpcStatusCode statusCode, String message) { + this(statusCode, message, null, null); + } + + public RpcException(RpcStatusCode statusCode, String message, Throwable cause) { + this(statusCode, message, null, cause); + } + + public RpcException(RpcStatusCode statusCode, String message, String remoteErrorType) { + this(statusCode, message, remoteErrorType, null); + } + + public RpcException(RpcStatusCode statusCode, String message, String remoteErrorType, Throwable cause) { + super(message, cause); + this.statusCode = statusCode == null ? RpcStatusCode.INTERNAL_ERROR : statusCode; + this.remoteErrorType = remoteErrorType == null ? "" : remoteErrorType; + } + + public RpcStatusCode getStatusCode() { + return statusCode; + } + + public String getRemoteErrorType() { + return remoteErrorType; + } + + public static Throwable unwrap(Throwable throwable) { + Throwable current = throwable; + while ((current instanceof CompletionException || current instanceof ExecutionException) + && current.getCause() != null) { + current = current.getCause(); + } + return current; + } + + public static RpcStatusCode statusOf(Throwable throwable) { + Throwable cause = unwrap(throwable); + if (cause instanceof RpcException rpcException) { + return rpcException.getStatusCode(); + } + if (cause instanceof TimeoutException) { + return RpcStatusCode.TIMEOUT; + } + return RpcStatusCode.INTERNAL_ERROR; + } +} diff --git a/rpc-core/src/main/java/com/xiaoyu/rpc/core/interceptor/RpcInterceptor.java b/rpc-core/src/main/java/com/xiaoyu/rpc/core/interceptor/RpcInterceptor.java new file mode 100644 index 0000000..4a5ab36 --- /dev/null +++ b/rpc-core/src/main/java/com/xiaoyu/rpc/core/interceptor/RpcInterceptor.java @@ -0,0 +1,21 @@ +package com.xiaoyu.rpc.core.interceptor; + +import com.xiaoyu.rpc.common.vo.RpcRequest; +import com.xiaoyu.rpc.common.vo.RpcResponse; + +public interface RpcInterceptor { + + default int order() { + return 0; + } + + default RpcRequest before(RpcInvocationContext context, RpcRequest request) { + return request; + } + + default void after(RpcInvocationContext context, RpcResponse response) { + } + + default void onError(RpcInvocationContext context, Throwable error) { + } +} diff --git a/rpc-core/src/main/java/com/xiaoyu/rpc/core/interceptor/RpcInterceptorRegistry.java b/rpc-core/src/main/java/com/xiaoyu/rpc/core/interceptor/RpcInterceptorRegistry.java new file mode 100644 index 0000000..dcbf702 --- /dev/null +++ b/rpc-core/src/main/java/com/xiaoyu/rpc/core/interceptor/RpcInterceptorRegistry.java @@ -0,0 +1,69 @@ +package com.xiaoyu.rpc.core.interceptor; + +import com.xiaoyu.rpc.common.vo.RpcRequest; +import com.xiaoyu.rpc.common.vo.RpcResponse; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Comparator; +import java.util.List; +import java.util.Objects; +import java.util.concurrent.CopyOnWriteArrayList; + +public final class RpcInterceptorRegistry { + + private static final Logger log = LoggerFactory.getLogger(RpcInterceptorRegistry.class); + private static final CopyOnWriteArrayList INTERCEPTORS = new CopyOnWriteArrayList<>(); + + private RpcInterceptorRegistry() { + } + + public static void register(RpcInterceptor interceptor) { + Objects.requireNonNull(interceptor, "interceptor"); + if (INTERCEPTORS.addIfAbsent(interceptor)) { + INTERCEPTORS.sort(Comparator.comparingInt(RpcInterceptor::order)); + } + } + + public static void unregister(RpcInterceptor interceptor) { + INTERCEPTORS.remove(interceptor); + } + + public static List getInterceptors() { + return List.copyOf(INTERCEPTORS); + } + + public static void clear() { + INTERCEPTORS.clear(); + } + + public static RpcRequest before(RpcInvocationContext context, RpcRequest request) { + RpcRequest current = request; + for (RpcInterceptor interceptor : INTERCEPTORS) { + current = Objects.requireNonNull( + interceptor.before(context, current), + () -> "RpcInterceptor.before must not return null: " + interceptor.getClass().getName()); + } + return current; + } + + public static void after(RpcInvocationContext context, RpcResponse response) { + for (int i = INTERCEPTORS.size() - 1; i >= 0; i--) { + try { + INTERCEPTORS.get(i).after(context, response); + } catch (Exception e) { + log.warn("RPC interceptor after callback failed: {}", INTERCEPTORS.get(i).getClass().getName(), e); + } + } + } + + public static void onError(RpcInvocationContext context, Throwable error) { + for (int i = INTERCEPTORS.size() - 1; i >= 0; i--) { + try { + INTERCEPTORS.get(i).onError(context, error); + } catch (Exception e) { + log.warn("RPC interceptor error callback failed: {}", INTERCEPTORS.get(i).getClass().getName(), e); + } + } + } +} diff --git a/rpc-core/src/main/java/com/xiaoyu/rpc/core/interceptor/RpcInvocationContext.java b/rpc-core/src/main/java/com/xiaoyu/rpc/core/interceptor/RpcInvocationContext.java new file mode 100644 index 0000000..892c24a --- /dev/null +++ b/rpc-core/src/main/java/com/xiaoyu/rpc/core/interceptor/RpcInvocationContext.java @@ -0,0 +1,60 @@ +package com.xiaoyu.rpc.core.interceptor; + +import com.xiaoyu.rpc.common.vo.RpcRequest; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +public final class RpcInvocationContext { + + private final RpcSide side; + private final String requestId; + private final String interfaceName; + private final String methodName; + private final long startNanos; + private final Map attributes = new ConcurrentHashMap<>(); + + public RpcInvocationContext(RpcSide side, RpcRequest request) { + this.side = side; + this.requestId = request.getRequestId(); + this.interfaceName = request.getInterfaceName(); + this.methodName = request.getMethodName(); + this.startNanos = System.nanoTime(); + } + + public RpcSide getSide() { + return side; + } + + public String getRequestId() { + return requestId; + } + + public String getInterfaceName() { + return interfaceName; + } + + public String getMethodName() { + return methodName; + } + + public long getStartNanos() { + return startNanos; + } + + public void putAttribute(String key, Object value) { + if (value == null) { + attributes.remove(key); + } else { + attributes.put(key, value); + } + } + + public Object getAttribute(String key) { + return attributes.get(key); + } + + public Map getAttributes() { + return Map.copyOf(attributes); + } +} diff --git a/rpc-core/src/main/java/com/xiaoyu/rpc/core/interceptor/RpcSide.java b/rpc-core/src/main/java/com/xiaoyu/rpc/core/interceptor/RpcSide.java new file mode 100644 index 0000000..64b3ea2 --- /dev/null +++ b/rpc-core/src/main/java/com/xiaoyu/rpc/core/interceptor/RpcSide.java @@ -0,0 +1,6 @@ +package com.xiaoyu.rpc.core.interceptor; + +public enum RpcSide { + CLIENT, + SERVER +} diff --git a/rpc-core/src/main/java/com/xiaoyu/rpc/core/observability/RpcMetricSide.java b/rpc-core/src/main/java/com/xiaoyu/rpc/core/observability/RpcMetricSide.java new file mode 100644 index 0000000..f43b462 --- /dev/null +++ b/rpc-core/src/main/java/com/xiaoyu/rpc/core/observability/RpcMetricSide.java @@ -0,0 +1,6 @@ +package com.xiaoyu.rpc.core.observability; + +public enum RpcMetricSide { + CLIENT, + SERVER +} diff --git a/rpc-core/src/main/java/com/xiaoyu/rpc/core/observability/RpcMetrics.java b/rpc-core/src/main/java/com/xiaoyu/rpc/core/observability/RpcMetrics.java new file mode 100644 index 0000000..c4ea559 --- /dev/null +++ b/rpc-core/src/main/java/com/xiaoyu/rpc/core/observability/RpcMetrics.java @@ -0,0 +1,123 @@ +package com.xiaoyu.rpc.core.observability; + +import com.xiaoyu.rpc.common.vo.RpcStatusCode; + +import java.util.EnumMap; +import java.util.Map; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.LongAdder; + +public final class RpcMetrics { + + private static final RpcMetrics INSTANCE = new RpcMetrics(); + + private final Map scopes = new EnumMap<>(RpcMetricSide.class); + + private RpcMetrics() { + for (RpcMetricSide side : RpcMetricSide.values()) { + scopes.put(side, new ScopeMetrics()); + } + } + + public static RpcMetrics getInstance() { + return INSTANCE; + } + + public CallTimer startCall(RpcMetricSide side) { + ScopeMetrics scope = scopes.get(side); + scope.totalRequests.increment(); + scope.activeRequests.increment(); + return new CallTimer(scope); + } + + public Snapshot snapshot(RpcMetricSide side) { + ScopeMetrics scope = scopes.get(side); + long total = scope.totalRequests.sum(); + long totalLatency = scope.totalLatencyNanos.sum(); + double averageLatencyMillis = total == 0 ? 0D : (totalLatency / 1_000_000D) / total; + double maxLatencyMillis = scope.maxLatencyNanos.get() / 1_000_000D; + return new Snapshot( + total, + scope.successRequests.sum(), + scope.failedRequests.sum(), + scope.timeoutRequests.sum(), + scope.activeRequests.sum(), + averageLatencyMillis, + maxLatencyMillis); + } + + public void reset() { + for (ScopeMetrics scope : scopes.values()) { + scope.reset(); + } + } + + public static final class CallTimer { + private final ScopeMetrics scope; + private final long startNanos = System.nanoTime(); + private final AtomicBoolean completed = new AtomicBoolean(false); + + private CallTimer(ScopeMetrics scope) { + this.scope = scope; + } + + public void success() { + finish(true, RpcStatusCode.SUCCESS); + } + + public void failure(RpcStatusCode statusCode) { + finish(false, statusCode); + } + + private void finish(boolean success, RpcStatusCode statusCode) { + if (!completed.compareAndSet(false, true)) { + return; + } + + long elapsed = Math.max(0L, System.nanoTime() - startNanos); + scope.activeRequests.add(-1L); + scope.totalLatencyNanos.add(elapsed); + scope.maxLatencyNanos.accumulateAndGet(elapsed, Math::max); + + if (success) { + scope.successRequests.increment(); + } else { + scope.failedRequests.increment(); + if (statusCode == RpcStatusCode.TIMEOUT) { + scope.timeoutRequests.increment(); + } + } + } + } + + public record Snapshot( + long totalRequests, + long successRequests, + long failedRequests, + long timeoutRequests, + long activeRequests, + double averageLatencyMillis, + double maxLatencyMillis) { + } + + private static final class ScopeMetrics { + private final LongAdder totalRequests = new LongAdder(); + private final LongAdder successRequests = new LongAdder(); + private final LongAdder failedRequests = new LongAdder(); + private final LongAdder timeoutRequests = new LongAdder(); + private final LongAdder activeRequests = new LongAdder(); + private final LongAdder totalLatencyNanos = new LongAdder(); + private final AtomicLong maxLatencyNanos = new AtomicLong(); + + private void reset() { + totalRequests.reset(); + successRequests.reset(); + failedRequests.reset(); + timeoutRequests.reset(); + activeRequests.reset(); + totalLatencyNanos.reset(); + maxLatencyNanos.set(0L); + } + } +} diff --git a/rpc-core/src/test/java/com/xiaoyu/rpc/core/client/RpcClientTest.java b/rpc-core/src/test/java/com/xiaoyu/rpc/core/client/RpcClientTest.java index 5480d03..27cab6e 100644 --- a/rpc-core/src/test/java/com/xiaoyu/rpc/core/client/RpcClientTest.java +++ b/rpc-core/src/test/java/com/xiaoyu/rpc/core/client/RpcClientTest.java @@ -5,7 +5,9 @@ import com.xiaoyu.rpc.common.serialization.Serializer; import com.xiaoyu.rpc.common.vo.RpcRequest; import com.xiaoyu.rpc.common.vo.RpcResponse; +import com.xiaoyu.rpc.common.vo.RpcStatusCode; import com.xiaoyu.rpc.core.config.RpcConfig; +import com.xiaoyu.rpc.core.exception.RpcException; import com.xiaoyu.rpc.core.registry.ServiceDiscovery; import com.xiaoyu.rpc.core.transport.TransportClient; import org.junit.jupiter.api.AfterEach; @@ -124,7 +126,7 @@ void testVoidReturnType() throws Exception { } @Test - @DisplayName("close 幂等释放传输层且关闭后拒绝新请求") + @DisplayName("close 幂等释放传输层且关闭后返回 CLIENT_CLOSED") void testCloseIsIdempotentAndRejectsNewRequests() throws Exception { AtomicInteger closeCount = new AtomicInteger(); AtomicInteger discoveryCount = new AtomicInteger(); @@ -151,7 +153,8 @@ public void close() { assertEquals(1, closeCount.get()); CompletableFuture future = rpcClient.sendRequest(minimalRequest(), String.class); ExecutionException ex = assertThrows(ExecutionException.class, () -> future.get(1, TimeUnit.SECONDS)); - assertInstanceOf(IllegalStateException.class, ex.getCause()); + assertInstanceOf(RpcException.class, ex.getCause()); + assertEquals(RpcStatusCode.CLIENT_CLOSED, ((RpcException) ex.getCause()).getStatusCode()); assertEquals(0, discoveryCount.get(), "关闭后的请求不应继续访问服务发现"); } diff --git a/rpc-core/src/test/java/com/xiaoyu/rpc/core/client/RpcObservabilityTest.java b/rpc-core/src/test/java/com/xiaoyu/rpc/core/client/RpcObservabilityTest.java new file mode 100644 index 0000000..298ba7f --- /dev/null +++ b/rpc-core/src/test/java/com/xiaoyu/rpc/core/client/RpcObservabilityTest.java @@ -0,0 +1,186 @@ +package com.xiaoyu.rpc.core.client; + +import com.google.protobuf.ByteString; +import com.xiaoyu.rpc.common.extension.ExtensionLoader; +import com.xiaoyu.rpc.common.serialization.Serializer; +import com.xiaoyu.rpc.common.vo.RpcRequest; +import com.xiaoyu.rpc.common.vo.RpcResponse; +import com.xiaoyu.rpc.common.vo.RpcStatusCode; +import com.xiaoyu.rpc.core.config.RpcConfig; +import com.xiaoyu.rpc.core.exception.RpcException; +import com.xiaoyu.rpc.core.interceptor.RpcInterceptor; +import com.xiaoyu.rpc.core.interceptor.RpcInterceptorRegistry; +import com.xiaoyu.rpc.core.interceptor.RpcInvocationContext; +import com.xiaoyu.rpc.core.observability.RpcMetricSide; +import com.xiaoyu.rpc.core.observability.RpcMetrics; +import com.xiaoyu.rpc.core.registry.ServiceDiscovery; +import com.xiaoyu.rpc.core.transport.TransportClient; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; + +import java.lang.reflect.Field; +import java.net.InetSocketAddress; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import static org.junit.jupiter.api.Assertions.*; + +@DisplayName("RPC 可观测性与拦截器测试") +class RpcObservabilityTest { + + private Serializer serializer; + + @BeforeEach + void setUp() throws Exception { + System.setProperty("rpc.registry", "local"); + System.setProperty("rpc.serializer", "java"); + resetRpcConfigSingleton(); + serializer = ExtensionLoader.getExtensionLoader(Serializer.class).getExtension("java"); + RpcInterceptorRegistry.clear(); + RpcMetrics.getInstance().reset(); + } + + @AfterEach + void tearDown() throws Exception { + RpcInterceptorRegistry.clear(); + RpcMetrics.getInstance().reset(); + System.clearProperty("rpc.registry"); + System.clearProperty("rpc.serializer"); + resetRpcConfigSingleton(); + } + + @Test + @DisplayName("客户端拦截器按顺序执行并可注入 metadata") + void testClientInterceptorOrderingAndMetadata() throws Exception { + List callbacks = new ArrayList<>(); + AtomicReference sentRequest = new AtomicReference<>(); + + RpcInterceptorRegistry.register(interceptor("late", 20, callbacks)); + RpcInterceptorRegistry.register(interceptor("early", 10, callbacks)); + + TransportClient transportClient = (request, address) -> { + sentRequest.set(request); + return CompletableFuture.completedFuture(successResponse(request, "ok")); + }; + ServiceDiscovery discovery = serviceName -> new InetSocketAddress("127.0.0.1", 8080); + + try (RpcClient client = new RpcClient(transportClient, discovery)) { + Object result = client.sendRequest(request(), String.class).get(1, TimeUnit.SECONDS); + assertEquals("ok", result); + } + + assertEquals(List.of("before-early", "before-late", "after-late", "after-early"), callbacks); + assertEquals("early", sentRequest.get().getMetadataOrThrow("early")); + assertEquals("late", sentRequest.get().getMetadataOrThrow("late")); + assertFalse(sentRequest.get().getRequestId().isEmpty()); + + RpcMetrics.Snapshot snapshot = RpcMetrics.getInstance().snapshot(RpcMetricSide.CLIENT); + assertEquals(1, snapshot.totalRequests()); + assertEquals(1, snapshot.successRequests()); + assertEquals(0, snapshot.failedRequests()); + assertEquals(0, snapshot.activeRequests()); + } + + @Test + @DisplayName("结构化错误转换为 RpcException 并计入失败指标") + void testStructuredErrorAndFailureMetrics() throws Exception { + AtomicReference interceptedError = new AtomicReference<>(); + RpcInterceptorRegistry.register(new RpcInterceptor() { + @Override + public void onError(RpcInvocationContext context, Throwable error) { + interceptedError.set(error); + } + }); + + TransportClient transportClient = (request, address) -> CompletableFuture.completedFuture( + RpcResponse.newBuilder() + .setRequestId(request.getRequestId()) + .setStatusCode(RpcStatusCode.BUSINESS_ERROR) + .setMessage("Error: boom") + .setErrorType(IllegalStateException.class.getName()) + .build()); + ServiceDiscovery discovery = serviceName -> new InetSocketAddress("127.0.0.1", 8080); + + try (RpcClient client = new RpcClient(transportClient, discovery)) { + ExecutionException exception = assertThrows( + ExecutionException.class, + () -> client.sendRequest(request(), String.class).get(1, TimeUnit.SECONDS)); + assertInstanceOf(RpcException.class, exception.getCause()); + RpcException rpcException = (RpcException) exception.getCause(); + assertEquals(RpcStatusCode.BUSINESS_ERROR, rpcException.getStatusCode()); + assertEquals(IllegalStateException.class.getName(), rpcException.getRemoteErrorType()); + } + + assertInstanceOf(RpcException.class, interceptedError.get()); + RpcMetrics.Snapshot snapshot = RpcMetrics.getInstance().snapshot(RpcMetricSide.CLIENT); + assertEquals(1, snapshot.totalRequests()); + assertEquals(0, snapshot.successRequests()); + assertEquals(1, snapshot.failedRequests()); + } + + @Test + @DisplayName("超时错误单独计入 timeout 指标") + void testTimeoutMetrics() throws Exception { + TransportClient transportClient = (request, address) -> CompletableFuture.failedFuture( + new RpcException(RpcStatusCode.TIMEOUT, "timeout")); + ServiceDiscovery discovery = serviceName -> new InetSocketAddress("127.0.0.1", 8080); + + try (RpcClient client = new RpcClient(transportClient, discovery)) { + assertThrows( + ExecutionException.class, + () -> client.sendRequest(request(), String.class).get(1, TimeUnit.SECONDS)); + } + + RpcMetrics.Snapshot snapshot = RpcMetrics.getInstance().snapshot(RpcMetricSide.CLIENT); + assertEquals(1, snapshot.failedRequests()); + assertEquals(1, snapshot.timeoutRequests()); + } + + private RpcInterceptor interceptor(String name, int order, List callbacks) { + return new RpcInterceptor() { + @Override + public int order() { + return order; + } + + @Override + public RpcRequest before(RpcInvocationContext context, RpcRequest request) { + callbacks.add("before-" + name); + return request.toBuilder().putMetadata(name, name).build(); + } + + @Override + public void after(RpcInvocationContext context, RpcResponse response) { + callbacks.add("after-" + name); + } + }; + } + + private RpcResponse successResponse(RpcRequest request, String value) { + return RpcResponse.newBuilder() + .setRequestId(request.getRequestId()) + .setStatusCode(RpcStatusCode.SUCCESS) + .setMessage("Success") + .setData(ByteString.copyFrom(serializer.serialize(value))) + .build(); + } + + private static RpcRequest request() { + return RpcRequest.newBuilder() + .setInterfaceName("com.example.DemoService") + .setMethodName("ping") + .build(); + } + + private static void resetRpcConfigSingleton() throws Exception { + Field field = RpcConfig.class.getDeclaredField("instance"); + field.setAccessible(true); + field.set(null, null); + } +} diff --git a/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/client/NettyRpcClientHandler.java b/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/client/NettyRpcClientHandler.java index 69f61d5..22ead0c 100644 --- a/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/client/NettyRpcClientHandler.java +++ b/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/client/NettyRpcClientHandler.java @@ -51,7 +51,8 @@ protected void channelRead0(ChannelHandlerContext ctx, RpcResponse response) { CompletableFuture future = pendingRequests.remove(requestId); if (future != null) { - log.info("Client received response for requestId: {}, status: {}", requestId, response.getMessage()); + // 成功响应属于高频路径，避免 INFO 级逐请求日志影响吞吐和延迟观测。 + log.debug("Client received response for requestId: {}, status: {}", requestId, response.getMessage()); future.complete(response); } else { log.warn("Client received response for unknown or timed-out requestId: {}", requestId); diff --git a/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/server/NettyRpcHandler.java b/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/server/NettyRpcHandler.java index 6b00266..ea68728 100644 --- a/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/server/NettyRpcHandler.java +++ b/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/server/NettyRpcHandler.java @@ -5,7 +5,14 @@ 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.common.vo.RpcStatusCode; import com.xiaoyu.rpc.core.config.RpcConfig; +import com.xiaoyu.rpc.core.exception.RpcException; +import com.xiaoyu.rpc.core.interceptor.RpcInterceptorRegistry; +import com.xiaoyu.rpc.core.interceptor.RpcInvocationContext; +import com.xiaoyu.rpc.core.interceptor.RpcSide; +import com.xiaoyu.rpc.core.observability.RpcMetricSide; +import com.xiaoyu.rpc.core.observability.RpcMetrics; import com.xiaoyu.rpc.core.util.TypeUtils; import io.netty.channel.ChannelHandler; import io.netty.channel.ChannelHandlerContext; @@ -28,9 +35,6 @@ public class NettyRpcHandler extends SimpleChannelInboundHandler { private final Executor businessExecutor; - /** - * 兼容直接构造场景。NettyTransportServer 会注入独立的有界业务线程池。 - */ public NettyRpcHandler() { this(ForkJoinPool.commonPool()); } @@ -41,30 +45,37 @@ public NettyRpcHandler(Executor businessExecutor) { @Override protected void channelRead0(ChannelHandlerContext ctx, RpcRequest request) { + RpcMetrics.CallTimer timer = RpcMetrics.getInstance().startCall(RpcMetricSide.SERVER); + RpcInvocationContext context = new RpcInvocationContext(RpcSide.SERVER, request); try { - // 反序列化、反射调用以及用户业务逻辑都可能阻塞，不能占用 Netty EventLoop。 - businessExecutor.execute(() -> processRequest(ctx, request)); + businessExecutor.execute(() -> processRequest(ctx, request, context, timer)); } catch (RejectedExecutionException e) { log.warn("RPC业务线程池已满，拒绝请求: interface={}, method={}, requestId={}", request.getInterfaceName(), request.getMethodName(), request.getRequestId()); - writeErrorResponse(ctx, request, "服务器繁忙，请稍后重试"); + RpcException error = new RpcException(RpcStatusCode.SERVER_BUSY, "服务器繁忙，请稍后重试", e); + timer.failure(error.getStatusCode()); + RpcInterceptorRegistry.onError(context, error); + writeErrorResponse(ctx, request, error.getStatusCode(), safeMessage(error), e.getClass().getName()); } } - private void processRequest(ChannelHandlerContext ctx, RpcRequest request) { - RpcResponse.Builder responseBuilder = RpcResponse.newBuilder() - .setRequestId(request.getRequestId()); - + private void processRequest(ChannelHandlerContext ctx, RpcRequest originalRequest, + RpcInvocationContext context, RpcMetrics.CallTimer timer) { + RpcRequest request = originalRequest; try { + request = RpcInterceptorRegistry.before(context, request); + Object serviceBean = ServiceRepository.getService(request.getInterfaceName()); if (serviceBean == null) { - throw new RuntimeException("未找到服务实现: " + request.getInterfaceName()); + throw new RpcException( + RpcStatusCode.SERVICE_NOT_FOUND, + "未找到服务实现: " + request.getInterfaceName()); } List paramTypeNames = request.getParamTypesList(); List paramByteList = request.getParametersList(); if (paramTypeNames.size() != paramByteList.size()) { - throw new IllegalArgumentException("参数类型数量与参数数量不一致"); + throw new RpcException(RpcStatusCode.INVALID_ARGUMENT, "参数类型数量与参数数量不一致"); } Class[] parameterTypes = new Class[paramTypeNames.size()]; @@ -74,49 +85,96 @@ private void processRequest(ChannelHandlerContext ctx, RpcRequest request) { for (int i = 0; i < paramTypeNames.size(); i++) { Class parameterType = TypeUtils.resolveClass(paramTypeNames.get(i)); parameterTypes[i] = parameterType; - byte[] bytes = paramByteList.get(i).toByteArray(); Class deserializeType = TypeUtils.wrapPrimitive(parameterType); parameters[i] = serializer.deserialize(bytes, deserializeType); } - Method method = serviceBean.getClass().getMethod(request.getMethodName(), parameterTypes); - Object result = method.invoke(serviceBean, parameters); + Method method; + try { + method = serviceBean.getClass().getMethod(request.getMethodName(), parameterTypes); + } catch (NoSuchMethodException e) { + throw new RpcException( + RpcStatusCode.METHOD_NOT_FOUND, + "未找到服务方法: " + request.getMethodName(), + e); + } + + Object result; + try { + result = method.invoke(serviceBean, parameters); + } catch (InvocationTargetException e) { + Throwable target = e.getTargetException() == null ? e : e.getTargetException(); + if (target instanceof RpcException rpcException) { + throw rpcException; + } + throw new RpcException( + RpcStatusCode.BUSINESS_ERROR, + safeMessage(target), + target.getClass().getName(), + target); + } byte[] resultBytes = result == null ? new byte[0] : serializer.serialize(result); - responseBuilder.setData(ByteString.copyFrom(resultBytes)); - responseBuilder.setMessage("Success"); + RpcResponse response = RpcResponse.newBuilder() + .setRequestId(request.getRequestId()) + .setData(ByteString.copyFrom(resultBytes)) + .setStatusCode(RpcStatusCode.SUCCESS) + .setMessage("Success") + .build(); + + RpcInterceptorRegistry.after(context, response); + timer.success(); + ctx.writeAndFlush(response); } catch (Exception e) { - Throwable cause = unwrapInvocationException(e); - log.error("Failed to process RPC request: interface={}, method={}, requestId={}", - request.getInterfaceName(), request.getMethodName(), request.getRequestId(), cause); - responseBuilder.setMessage("Error: " + safeMessage(cause)); - responseBuilder.setData(ByteString.EMPTY); + RpcException error = normalizeError(e); + log.error("Failed to process RPC request: interface={}, method={}, requestId={}, status={}", + request.getInterfaceName(), request.getMethodName(), request.getRequestId(), + error.getStatusCode(), error); + timer.failure(error.getStatusCode()); + RpcInterceptorRegistry.onError(context, error); + writeErrorResponse( + ctx, + request, + error.getStatusCode(), + error.getMessage(), + error.getRemoteErrorType().isEmpty() + ? RpcException.unwrap(error).getClass().getName() + : error.getRemoteErrorType()); } + } - ctx.writeAndFlush(responseBuilder.build()); + private RpcException normalizeError(Exception error) { + if (error instanceof RpcException rpcException) { + return rpcException; + } + if (error instanceof IllegalArgumentException) { + return new RpcException( + RpcStatusCode.INVALID_ARGUMENT, + safeMessage(error), + error.getClass().getName(), + error); + } + return new RpcException( + RpcStatusCode.INTERNAL_ERROR, + safeMessage(error), + error.getClass().getName(), + error); } - private void writeErrorResponse(ChannelHandlerContext ctx, RpcRequest request, String message) { + private void writeErrorResponse(ChannelHandlerContext ctx, RpcRequest request, + RpcStatusCode statusCode, String message, String errorType) { RpcResponse response = RpcResponse.newBuilder() .setRequestId(request.getRequestId()) + .setStatusCode(statusCode) .setMessage("Error: " + message) + .setErrorType(errorType == null ? "" : errorType) .setData(ByteString.EMPTY) .build(); ctx.writeAndFlush(response); } - private Throwable unwrapInvocationException(Exception e) { - if (e instanceof InvocationTargetException) { - Throwable target = ((InvocationTargetException) e).getTargetException(); - if (target != null) { - return target; - } - } - return e; - } - - private String safeMessage(Throwable throwable) { + private static String safeMessage(Throwable throwable) { String message = throwable.getMessage(); return message == null || message.isEmpty() ? throwable.getClass().getSimpleName() : message; } diff --git a/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/transport/netty/NettyTransportClient.java b/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/transport/netty/NettyTransportClient.java index 8a46dbf..8d071d7 100644 --- a/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/transport/netty/NettyTransportClient.java +++ b/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/transport/netty/NettyTransportClient.java @@ -1,10 +1,11 @@ package com.xiaoyu.rpc.core.transport.netty; import com.xiaoyu.rpc.common.vo.RpcRequest; -import com.xiaoyu.rpc.common.vo.RpcResponse; +import com.xiaoyu.rpc.common.vo.RpcStatusCode; import com.xiaoyu.rpc.core.client.ChannelProvider; import com.xiaoyu.rpc.core.client.NettyRpcClientHandler; import com.xiaoyu.rpc.core.config.RpcConfig; +import com.xiaoyu.rpc.core.exception.RpcException; import com.xiaoyu.rpc.core.protocol.Protocol; import com.xiaoyu.rpc.core.protocol.ProtocolFactory; import com.xiaoyu.rpc.core.transport.TransportClient; @@ -25,7 +26,6 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; @Slf4j @@ -58,7 +58,8 @@ protected void initChannel(SocketChannel ch) { @Override public CompletableFuture sendRequest(RpcRequest request, InetSocketAddress address) { if (closed.get()) { - return CompletableFuture.failedFuture(new IllegalStateException("NettyTransportClient 已关闭")); + return CompletableFuture.failedFuture( + new RpcException(RpcStatusCode.CLIENT_CLOSED, "NettyTransportClient 已关闭")); } RpcConfig config = RpcConfig.getInstance(); @@ -74,17 +75,20 @@ public CompletableFuture sendRequest(RpcRequest request, InetSocketAddre }); } catch (Exception e) { log.error("RPC请求发起失败", e); - return CompletableFuture.failedFuture(e); + return CompletableFuture.failedFuture( + new RpcException(RpcStatusCode.UNAVAILABLE, "RPC请求发起失败: " + address, e)); } } private CompletableFuture sendOnChannel(RpcRequest request, Channel channel, RpcConfig config, String protocolName) { if (closed.get()) { - return CompletableFuture.failedFuture(new IllegalStateException("NettyTransportClient 已关闭")); + return CompletableFuture.failedFuture( + new RpcException(RpcStatusCode.CLIENT_CLOSED, "NettyTransportClient 已关闭")); } if (channel == null || !channel.isActive()) { - return CompletableFuture.failedFuture(new RuntimeException("无法连接到服务器: " + channel)); + return CompletableFuture.failedFuture( + new RpcException(RpcStatusCode.UNAVAILABLE, "无法连接到服务器: " + channel)); } NettyRpcClientHandler handler = channel.pipeline().get(NettyRpcClientHandler.class); @@ -99,10 +103,12 @@ private CompletableFuture sendOnChannel(RpcRequest request, Channel chan } final NettyRpcClientHandler clientHandler = handler; - String requestId = UUID.randomUUID().toString(); - RpcRequest newRequest = request.toBuilder() - .setRequestId(requestId) - .build(); + String requestId = request.getRequestId().isEmpty() + ? UUID.randomUUID().toString() + : request.getRequestId(); + RpcRequest newRequest = request.getRequestId().isEmpty() + ? request.toBuilder().setRequestId(requestId).build() + : request; CompletableFuture resultFuture = new CompletableFuture<>(); clientHandler.addFuture(requestId, resultFuture); @@ -112,8 +118,9 @@ private CompletableFuture sendOnChannel(RpcRequest request, Channel chan try { timeoutTask = channel.eventLoop().schedule( () -> clientHandler.failRequest(requestId, - new TimeoutException("RPC请求超时: requestId=" + requestId - + ", timeoutMs=" + timeoutMillis)), + new RpcException( + RpcStatusCode.TIMEOUT, + "RPC请求超时: requestId=" + requestId + ", timeoutMs=" + timeoutMillis)), timeoutMillis, TimeUnit.MILLISECONDS); } catch (Exception e) { @@ -130,19 +137,11 @@ private CompletableFuture sendOnChannel(RpcRequest request, Channel chan try { protocol.sendRequest(channel, newRequest, clientHandler); } catch (Exception e) { - clientHandler.failRequest(requestId, e); + clientHandler.failRequest(requestId, + new RpcException(RpcStatusCode.UNAVAILABLE, "发送 RPC 请求失败", e)); } - return resultFuture.thenApply(result -> { - if (result instanceof RpcResponse) { - RpcResponse rpcResponse = (RpcResponse) result; - if (!"Success".equals(rpcResponse.getMessage())) { - throw new RuntimeException("服务端报错: " + rpcResponse.getMessage()); - } - return rpcResponse; - } - throw new RuntimeException("服务端返回的不是 RpcResponse 类型"); - }); + return resultFuture; } @Override diff --git a/rpc-transport-netty/src/test/java/com/xiaoyu/rpc/core/server/NettyRpcObservabilityTest.java b/rpc-transport-netty/src/test/java/com/xiaoyu/rpc/core/server/NettyRpcObservabilityTest.java new file mode 100644 index 0000000..244b18b --- /dev/null +++ b/rpc-transport-netty/src/test/java/com/xiaoyu/rpc/core/server/NettyRpcObservabilityTest.java @@ -0,0 +1,176 @@ +package com.xiaoyu.rpc.core.server; + +import com.google.protobuf.ByteString; +import com.xiaoyu.rpc.common.extension.ExtensionLoader; +import com.xiaoyu.rpc.common.serialization.Serializer; +import com.xiaoyu.rpc.common.vo.RpcRequest; +import com.xiaoyu.rpc.common.vo.RpcResponse; +import com.xiaoyu.rpc.common.vo.RpcStatusCode; +import com.xiaoyu.rpc.core.config.RpcConfig; +import com.xiaoyu.rpc.core.interceptor.RpcInterceptor; +import com.xiaoyu.rpc.core.interceptor.RpcInterceptorRegistry; +import com.xiaoyu.rpc.core.interceptor.RpcInvocationContext; +import com.xiaoyu.rpc.core.observability.RpcMetricSide; +import com.xiaoyu.rpc.core.observability.RpcMetrics; +import io.netty.channel.embedded.EmbeddedChannel; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; + +import java.lang.reflect.Field; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.*; + +@DisplayName("RPC 服务端错误模型与可观测性测试") +class NettyRpcObservabilityTest { + + private Serializer serializer; + + @BeforeEach + void setUp() throws Exception { + System.setProperty("rpc.registry", "local"); + System.setProperty("rpc.serializer", "java"); + resetRpcConfigSingleton(); + serializer = ExtensionLoader.getExtensionLoader(Serializer.class).getExtension("java"); + RpcInterceptorRegistry.clear(); + RpcMetrics.getInstance().reset(); + } + + @AfterEach + void tearDown() throws Exception { + RpcInterceptorRegistry.clear(); + RpcMetrics.getInstance().reset(); + System.clearProperty("rpc.registry"); + System.clearProperty("rpc.serializer"); + resetRpcConfigSingleton(); + } + + @Test + @DisplayName("服务不存在返回结构化 SERVICE_NOT_FOUND") + void testServiceNotFoundStatus() { + EmbeddedChannel channel = new EmbeddedChannel(new NettyRpcHandler(Runnable::run)); + channel.writeInbound(RpcRequest.newBuilder() + .setRequestId("missing-1") + .setInterfaceName("com.example.MissingService") + .setMethodName("call") + .build()); + + RpcResponse response = channel.readOutbound(); + assertEquals(RpcStatusCode.SERVICE_NOT_FOUND, response.getStatusCode()); + assertTrue(response.getMessage().startsWith("Error:")); + assertFalse(response.getErrorType().isEmpty()); + + RpcMetrics.Snapshot snapshot = RpcMetrics.getInstance().snapshot(RpcMetricSide.SERVER); + assertEquals(1, snapshot.totalRequests()); + assertEquals(1, snapshot.failedRequests()); + channel.finishAndReleaseAll(); + } + + @Test + @DisplayName("业务异常返回 BUSINESS_ERROR 和远端异常类型") + void testBusinessErrorStatus() { + ServiceRepository.registerService(FailingService.class.getName(), new FailingServiceImpl()); + EmbeddedChannel channel = new EmbeddedChannel(new NettyRpcHandler(Runnable::run)); + channel.writeInbound(RpcRequest.newBuilder() + .setRequestId("business-1") + .setInterfaceName(FailingService.class.getName()) + .setMethodName("fail") + .build()); + + RpcResponse response = channel.readOutbound(); + assertEquals(RpcStatusCode.BUSINESS_ERROR, response.getStatusCode()); + assertEquals(IllegalStateException.class.getName(), response.getErrorType()); + assertTrue(response.getMessage().contains("boom")); + channel.finishAndReleaseAll(); + } + + @Test + @DisplayName("服务端拦截器收到 before/after 回调并记录成功指标") + void testServerInterceptorAndMetrics() { + ServiceRepository.registerService(EchoService.class.getName(), new EchoServiceImpl()); + AtomicInteger beforeCount = new AtomicInteger(); + AtomicInteger afterCount = new AtomicInteger(); + RpcInterceptorRegistry.register(new RpcInterceptor() { + @Override + public RpcRequest before(RpcInvocationContext context, RpcRequest request) { + assertEquals("trace-123", request.getMetadataOrThrow("trace-id")); + beforeCount.incrementAndGet(); + return request; + } + + @Override + public void after(RpcInvocationContext context, RpcResponse response) { + afterCount.incrementAndGet(); + } + }); + + EmbeddedChannel channel = new EmbeddedChannel(new NettyRpcHandler(Runnable::run)); + channel.writeInbound(RpcRequest.newBuilder() + .setRequestId("echo-1") + .setInterfaceName(EchoService.class.getName()) + .setMethodName("echo") + .addParamTypes(String.class.getName()) + .addParameters(ByteString.copyFrom(serializer.serialize("hello"))) + .putMetadata("trace-id", "trace-123") + .build()); + + RpcResponse response = channel.readOutbound(); + assertEquals(RpcStatusCode.SUCCESS, response.getStatusCode()); + assertEquals(1, beforeCount.get()); + assertEquals(1, afterCount.get()); + + RpcMetrics.Snapshot snapshot = RpcMetrics.getInstance().snapshot(RpcMetricSide.SERVER); + assertEquals(1, snapshot.successRequests()); + assertEquals(0, snapshot.activeRequests()); + channel.finishAndReleaseAll(); + } + + @Test + @DisplayName("业务线程池拒绝请求返回 SERVER_BUSY") + void testServerBusyStatus() { + EmbeddedChannel channel = new EmbeddedChannel(new NettyRpcHandler(command -> { + throw new RejectedExecutionException("full"); + })); + channel.writeInbound(RpcRequest.newBuilder() + .setRequestId("busy-1") + .setInterfaceName("com.example.AnyService") + .setMethodName("call") + .build()); + + RpcResponse response = channel.readOutbound(); + assertEquals(RpcStatusCode.SERVER_BUSY, response.getStatusCode()); + assertEquals(1, RpcMetrics.getInstance().snapshot(RpcMetricSide.SERVER).failedRequests()); + channel.finishAndReleaseAll(); + } + + public interface FailingService { + String fail(); + } + + public static class FailingServiceImpl implements FailingService { + @Override + public String fail() { + throw new IllegalStateException("boom"); + } + } + + public interface EchoService { + String echo(String value); + } + + public static class EchoServiceImpl implements EchoService { + @Override + public String echo(String value) { + return value; + } + } + + private static void resetRpcConfigSingleton() throws Exception { + Field field = RpcConfig.class.getDeclaredField("instance"); + field.setAccessible(true); + field.set(null, null); + } +}