Skip to content

Commit 0fc42e6

Browse files
xiaoyumuxixiaoyumuxi
committed
给Java的客户端也添加了gRPC模式，删除了之前的gRPC仅仅支持跨语言客户端的设计
1 parent b78d811 commit 0fc42e6

6 files changed

Lines changed: 220 additions & 7 deletions

File tree

‎README.md‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -156,8 +156,8 @@ rpc:
156156
```
157157
158158
> [!IMPORTANT]
159-
> Use `netty`/`http`/`http2` with `rpc-consumer` (`RpcClientProxy`).
160-
> Switch to `grpc` only when interoperating with standard grpc clients (Python/Go).
159+
> Default is `netty` for a faster local Java-to-Java path.
160+
> `grpc` now also works with `rpc-consumer` (`RpcClientProxy`) and can be used for both Java and Python/Go interoperability.
161161

162162
## 🔌 SPI Design & Ecosystem
163163

‎README_ZH.md‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -193,8 +193,8 @@ rpc:
193193
```
194194
195195
> [!IMPORTANT]
196-
> 若通过 `rpc-consumer` 的 `RpcClientProxy` 进行调用，请使用 `netty`/`http`/`http2` 协议。
197-
> `grpc` 仅建议用于与标准 grpc 客户端（Python/Go）互操作场景。
196+
> 默认使用 `netty`，适合本地 Java-to-Java 的高性能链路。
197+
> 现在 `grpc` 也支持 `rpc-consumer`（`RpcClientProxy`）调用，可同时用于 Java 侧和 Python/Go 互操作。
198198

199199
---
200200

‎rpc-consumer/src/test/java/com/xiaoyu/rpc/consumer/FullIntegrationTest.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -41,7 +41,7 @@ public void testFullIntegration() throws InterruptedException {
4141
// 沙箱或受限环境无法监听端口时，跳过此集成测试，避免把环境问题算成代码失败
4242
Assumptions.assumeTrue(canBindLocalPort(9090), "No permission to bind local test port 9090");
4343

44-
// 当前默认配置可能是 grpc，但客户端泛化 grpc 尚未支持，测试里强制切到 netty
44+
// 集成测试固定走 netty，减少跨协议变量，确保该用例只验证端到端调用主链路
4545
forceConfig("protocol", "netty");
4646

4747
// Start Server in a thread

‎rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/protocol/grpc/GrpcClientResponseHandler.java‎

Lines changed: 77 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,77 @@
1+
package com.xiaoyu.rpc.core.protocol.grpc;
2+
3+
import com.xiaoyu.rpc.common.vo.RpcResponse;
4+
import com.xiaoyu.rpc.core.client.NettyRpcClientHandler;
5+
import io.netty.buffer.ByteBuf;
6+
import io.netty.channel.ChannelHandlerContext;
7+
import io.netty.channel.SimpleChannelInboundHandler;
8+
import io.netty.handler.codec.http2.Http2DataFrame;
9+
import io.netty.handler.codec.http2.Http2Frame;
10+
import io.netty.handler.codec.http2.Http2Headers;
11+
import io.netty.handler.codec.http2.Http2HeadersFrame;
12+
13+
/**
14+
* 将 gRPC/HTTP2 帧转换为内部 RpcResponse，并交给通用客户端处理器完成 requestId 关联。
15+
*/
16+
class GrpcClientResponseHandler extends SimpleChannelInboundHandler<Http2Frame> {
17+
18+
private final NettyRpcClientHandler clientHandler;
19+
private final String requestId;
20+
21+
GrpcClientResponseHandler(NettyRpcClientHandler clientHandler, String requestId) {
22+
this.clientHandler = clientHandler;
23+
this.requestId = requestId;
24+
}
25+
26+
@Override
27+
protected void channelRead0(ChannelHandlerContext ctx, Http2Frame frame) throws Exception {
28+
if (frame instanceof Http2DataFrame) {
29+
Http2DataFrame dataFrame = (Http2DataFrame) frame;
30+
ByteBuf content = dataFrame.content();
31+
if (content.readableBytes() < 5) {
32+
clientHandler.failRequest(requestId, new IllegalStateException("Invalid gRPC frame: missing 5-byte prefix"));
33+
return;
34+
}
35+
36+
byte compressedFlag = content.readByte();
37+
if (compressedFlag != 0) {
38+
clientHandler.failRequest(requestId, new UnsupportedOperationException("Compressed gRPC payload is not supported"));
39+
return;
40+
}
41+
42+
int length = content.readInt();
43+
if (content.readableBytes() < length) {
44+
clientHandler.failRequest(requestId, new IllegalStateException("Invalid gRPC frame: payload length mismatch"));
45+
return;
46+
}
47+
48+
ByteBuf slice = content.readSlice(length);
49+
RpcResponse response;
50+
if (slice.nioBufferCount() > 0) {
51+
response = RpcResponse.parseFrom(slice.nioBuffer());
52+
} else {
53+
byte[] bytes = new byte[length];
54+
slice.readBytes(bytes);
55+
response = RpcResponse.parseFrom(bytes);
56+
}
57+
ctx.fireChannelRead(response);
58+
return;
59+
}
60+
61+
if (frame instanceof Http2HeadersFrame) {
62+
Http2Headers headers = ((Http2HeadersFrame) frame).headers();
63+
CharSequence grpcStatus = headers.get("grpc-status");
64+
if (grpcStatus != null && !"0".contentEquals(grpcStatus)) {
65+
CharSequence grpcMessage = headers.get("grpc-message");
66+
String message = grpcMessage == null ? "unknown grpc error" : grpcMessage.toString();
67+
clientHandler.failRequest(requestId, new RuntimeException("gRPC error status=" + grpcStatus + ", message=" + message));
68+
}
69+
}
70+
}
71+
72+
@Override
73+
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
74+
clientHandler.failRequest(requestId, cause);
75+
ctx.close();
76+
}
77+
}

‎rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/protocol/grpc/GrpcProtocol.java‎

Lines changed: 61 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -5,10 +5,23 @@
55
import com.xiaoyu.rpc.core.protocol.Protocol;
66
import io.netty.channel.Channel;
77
import io.netty.channel.ChannelHandler;
8+
import io.netty.channel.ChannelHandlerContext;
9+
import io.netty.channel.ChannelInboundHandlerAdapter;
810
import io.netty.channel.ChannelInitializer;
911
import io.netty.channel.ChannelPipeline;
12+
import io.netty.handler.codec.http.HttpHeaderNames;
13+
import io.netty.handler.codec.http2.DefaultHttp2DataFrame;
14+
import io.netty.handler.codec.http2.DefaultHttp2Headers;
15+
import io.netty.handler.codec.http2.DefaultHttp2HeadersFrame;
1016
import io.netty.handler.codec.http2.Http2FrameCodecBuilder;
17+
import io.netty.handler.codec.http2.Http2Headers;
18+
import io.netty.handler.codec.http2.Http2Settings;
19+
import io.netty.handler.codec.http2.Http2StreamChannel;
20+
import io.netty.handler.codec.http2.Http2StreamChannelBootstrap;
1121
import io.netty.handler.codec.http2.Http2MultiplexHandler;
22+
import io.netty.util.ReferenceCountUtil;
23+
24+
import java.net.InetSocketAddress;
1225

1326
public class GrpcProtocol implements Protocol {
1427

@@ -35,12 +48,58 @@ protected void initChannel(Channel ch) throws Exception {
3548
}
3649
}));
3750
} else {
38-
throw new UnsupportedOperationException("Client side grpc not supported yet");
51+
pipeline.addLast(Http2FrameCodecBuilder.forClient()
52+
.autoAckSettingsFrame(true)
53+
.autoAckPingFrame(true)
54+
.initialSettings(Http2Settings.defaultSettings().maxHeaderListSize(8192))
55+
.build());
56+
pipeline.addLast(new Http2MultiplexHandler(new ChannelInboundHandlerAdapter() {
57+
@Override
58+
public void channelRead(ChannelHandlerContext ctx, Object msg) {
59+
// 连接级残留帧统一释放，避免引用计数对象泄漏
60+
ReferenceCountUtil.release(msg);
61+
}
62+
}));
3963
}
4064
}
4165

4266
@Override
4367
public void sendRequest(Channel channel, RpcRequest request, NettyRpcClientHandler clientHandler) throws Exception {
44-
throw new UnsupportedOperationException("Client side generic grpc not supported yet");
68+
Http2StreamChannelBootstrap streamBootstrap = new Http2StreamChannelBootstrap(channel);
69+
streamBootstrap.open().addListener(openFuture -> {
70+
if (!openFuture.isSuccess()) {
71+
clientHandler.failRequest(request.getRequestId(), openFuture.cause());
72+
return;
73+
}
74+
75+
Http2StreamChannel streamChannel = (Http2StreamChannel) openFuture.getNow();
76+
streamChannel.pipeline().addLast(new GrpcClientResponseHandler(clientHandler, request.getRequestId()));
77+
streamChannel.pipeline().addLast(clientHandler);
78+
79+
byte[] payload = request.toByteArray();
80+
io.netty.buffer.ByteBuf body = streamChannel.alloc().buffer(payload.length + 5);
81+
body.writeByte(0); // compressed-flag
82+
body.writeInt(payload.length);
83+
body.writeBytes(payload);
84+
85+
Http2Headers headers = new DefaultHttp2Headers()
86+
.method("POST")
87+
.path("/GrpcService/handle")
88+
.scheme("http")
89+
.set(HttpHeaderNames.CONTENT_TYPE, "application/grpc")
90+
.set(HttpHeaderNames.TE, "trailers");
91+
92+
if (channel.remoteAddress() instanceof InetSocketAddress) {
93+
InetSocketAddress remote = (InetSocketAddress) channel.remoteAddress();
94+
headers.authority(remote.getHostString() + ":" + remote.getPort());
95+
}
96+
97+
streamChannel.write(new DefaultHttp2HeadersFrame(headers, false));
98+
streamChannel.writeAndFlush(new DefaultHttp2DataFrame(body, true)).addListener(writeFuture -> {
99+
if (!writeFuture.isSuccess()) {
100+
clientHandler.failRequest(request.getRequestId(), writeFuture.cause());
101+
}
102+
});
103+
});
45104
}
46105
}

‎rpc-transport-netty/src/test/java/com/xiaoyu/rpc/core/protocol/grpc/GrpcClientResponseHandlerTest.java‎

Lines changed: 77 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,77 @@
1+
package com.xiaoyu.rpc.core.protocol.grpc;
2+
3+
import com.google.protobuf.ByteString;
4+
import com.xiaoyu.rpc.common.vo.RpcResponse;
5+
import com.xiaoyu.rpc.core.client.NettyRpcClientHandler;
6+
import io.netty.buffer.ByteBuf;
7+
import io.netty.buffer.Unpooled;
8+
import io.netty.channel.embedded.EmbeddedChannel;
9+
import io.netty.handler.codec.http2.DefaultHttp2DataFrame;
10+
import io.netty.handler.codec.http2.DefaultHttp2HeadersFrame;
11+
import io.netty.handler.codec.http2.Http2Headers;
12+
import org.junit.jupiter.api.DisplayName;
13+
import org.junit.jupiter.api.Test;
14+
15+
import java.util.concurrent.CompletableFuture;
16+
import java.util.concurrent.ExecutionException;
17+
18+
import static org.junit.jupiter.api.Assertions.*;
19+
20+
@DisplayName("gRPC 客户端响应处理器测试")
21+
class GrpcClientResponseHandlerTest {
22+
23+
@Test
24+
@DisplayName("应把 DataFrame 解码为 RpcResponse 并完成 future")
25+
void testDecodeDataFrameToRpcResponse() throws Exception {
26+
NettyRpcClientHandler clientHandler = new NettyRpcClientHandler();
27+
EmbeddedChannel channel = new EmbeddedChannel(
28+
new GrpcClientResponseHandler(clientHandler, "req-1"),
29+
clientHandler);
30+
31+
CompletableFuture<Object> future = new CompletableFuture<>();
32+
clientHandler.addFuture("req-1", future);
33+
34+
RpcResponse response = RpcResponse.newBuilder()
35+
.setRequestId("req-1")
36+
.setMessage("Success")
37+
.setData(ByteString.copyFromUtf8("ok"))
38+
.build();
39+
40+
byte[] payload = response.toByteArray();
41+
ByteBuf buf = Unpooled.buffer();
42+
buf.writeByte(0);
43+
buf.writeInt(payload.length);
44+
buf.writeBytes(payload);
45+
46+
channel.writeInbound(new DefaultHttp2DataFrame(buf, true));
47+
48+
assertTrue(future.isDone(), "Future should be completed");
49+
Object result = future.get();
50+
assertInstanceOf(RpcResponse.class, result);
51+
assertEquals("req-1", ((RpcResponse) result).getRequestId());
52+
assertEquals("Success", ((RpcResponse) result).getMessage());
53+
}
54+
55+
@Test
56+
@DisplayName("收到 grpc-status 非 0 时应异常完成 future")
57+
void testFailFutureOnGrpcErrorStatus() {
58+
NettyRpcClientHandler clientHandler = new NettyRpcClientHandler();
59+
EmbeddedChannel channel = new EmbeddedChannel(
60+
new GrpcClientResponseHandler(clientHandler, "req-2"),
61+
clientHandler);
62+
63+
CompletableFuture<Object> future = new CompletableFuture<>();
64+
clientHandler.addFuture("req-2", future);
65+
66+
Http2Headers trailers = new io.netty.handler.codec.http2.DefaultHttp2Headers()
67+
.set("grpc-status", "13")
68+
.set("grpc-message", "internal");
69+
channel.writeInbound(new DefaultHttp2HeadersFrame(trailers, true));
70+
71+
assertTrue(future.isCompletedExceptionally(), "Future should be completed exceptionally");
72+
ExecutionException ex = assertThrows(ExecutionException.class, future::get);
73+
String message = ex.getCause().getMessage();
74+
assertNotNull(message, "Error message should not be null");
75+
assertTrue(message.toLowerCase().contains("grpc"), "Error message should include grpc details");
76+
}
77+
}

0 commit comments

Comments
 (0)