Skip to content

Commit 3fa0d0f

Browse files
xiaoyumuxixiaoyumuxi
committed
fix:多路复用重构和异步优化
1 parent 29f89ad commit 3fa0d0f

8 files changed

Lines changed: 94 additions & 45 deletions

File tree

‎rpc-benchmark/src/main/java/com/xiaoyu/rpc/benchmark/ProtocolBenchmark.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -25,8 +25,9 @@
2525
@OutputTimeUnit(TimeUnit.MILLISECONDS)
2626
@State(Scope.Benchmark)
2727
@Fork(value = 1, warmups = 0)
28-
@Warmup(iterations = 1, time = 1)
29-
@Measurement(iterations = 2, time = 1)
28+
@Warmup(iterations = 3, time = 1)
29+
@Measurement(iterations = 5, time = 1)
30+
@Threads(8)
3031
public class ProtocolBenchmark {
3132

3233
@Param({ "netty", "http", "http2" })

‎rpc-common/src/main/proto/rpc_meta.proto‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,9 @@ message RpcRequest {
2121
// Proto存不了Object，必须存成二进制(bytes)。
2222
// 这里用 repeated 表示数组
2323
repeated bytes parameters = 4;
24+
25+
// 请求ID，用于多路复用
26+
string request_id = 5;
2427
}
2528

2629
// 4. 对应你的 RpcResponse
@@ -32,6 +35,8 @@ message RpcResponse {
3235
// 对应 String message
3336
string message = 2;
3437

38+
// 请求ID，用于多路复用
39+
string request_id = 3;
3540
}
3641

3742
service GrpcService {

‎rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/client/NettyRpcClientHandler.java‎

Lines changed: 34 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -11,32 +11,54 @@
1111
import org.slf4j.Logger;
1212
import org.slf4j.LoggerFactory;
1313

14+
import io.netty.channel.ChannelHandler;
15+
1416
// 这是一个 Netty 的 Handler，专门负责“收信”
17+
@ChannelHandler.Sharable
1518
public class NettyRpcClientHandler extends SimpleChannelInboundHandler<RpcResponse> {
1619
private static final Logger log = LoggerFactory.getLogger(NettyRpcClientHandler.class);
1720

18-
private CompletableFuture<Object> future;
21+
// Key: RequestId, Value: Future
22+
private final java.util.Map<String, CompletableFuture<Object>> pendingRequests = new java.util.concurrent.ConcurrentHashMap<>();
23+
24+
public void addFuture(String requestId, CompletableFuture<Object> future) {
25+
pendingRequests.put(requestId, future);
26+
}
1927

20-
public void setFuture(CompletableFuture<Object> future) {
21-
this.future = future;
28+
public void removeFuture(String requestId) {
29+
pendingRequests.remove(requestId);
30+
}
31+
32+
public void failRequest(String requestId, Throwable cause) {
33+
CompletableFuture<Object> future = pendingRequests.remove(requestId);
34+
if (future != null) {
35+
future.completeExceptionally(cause);
36+
}
2237
}
2338

2439
@Override
2540
protected void channelRead0(ChannelHandlerContext ctx, RpcResponse response) {
26-
// 【关键修复点】
27-
// 之前你写的是 future.complete(response.getData()); 导致传回去的是 ByteString
28-
// 现在我们把整个 response 对象传回去，让 Proxy 去判断状态和拆包
29-
log.info("客户端收到响应状态: {}", response.getMessage());
30-
future.complete(response);
41+
String requestId = response.getRequestId();
42+
CompletableFuture<Object> future = pendingRequests.remove(requestId);
43+
44+
if (future != null) {
45+
log.info("Client received response for requestId: {}, status: {}", requestId, response.getMessage());
46+
future.complete(response);
47+
} else {
48+
log.warn("Client received response for unknown or timed-out requestId: {}", requestId);
49+
}
3150
}
3251

3352
@Override
3453
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
35-
cause.printStackTrace();
54+
log.error("Client caught exception", cause);
55+
// Fail all pending requests
56+
for (CompletableFuture<Object> future : pendingRequests.values()) {
57+
future.completeExceptionally(cause);
58+
}
59+
pendingRequests.clear();
3660
ctx.close();
3761
}
3862

39-
public CompletableFuture<Object> getFuture() {
40-
return future;
41-
}
63+
// public CompletableFuture<Object> getFuture() { ... } // Removed single getter
4264
}

‎rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/protocol/http/HttpProtocol.java‎

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -52,14 +52,13 @@ public void config(ChannelPipeline pipeline, boolean isServer, io.netty.channel.
5252
public void sendRequest(io.netty.channel.Channel channel, RpcRequest request,
5353
com.xiaoyu.rpc.core.client.NettyRpcClientHandler clientHandler) throws Exception {
5454
// HTTP 1.1 协议也复用主通道
55-
if (channel.pipeline().get(com.xiaoyu.rpc.core.client.NettyRpcClientHandler.class) != null) {
56-
channel.pipeline().remove(com.xiaoyu.rpc.core.client.NettyRpcClientHandler.class);
55+
if (channel.pipeline().get(com.xiaoyu.rpc.core.client.NettyRpcClientHandler.class) == null) {
56+
channel.pipeline().addLast(clientHandler);
5757
}
58-
channel.pipeline().addLast(clientHandler);
5958

6059
channel.writeAndFlush(request).addListener(future -> {
6160
if (!future.isSuccess()) {
62-
clientHandler.getFuture().completeExceptionally(future.cause());
61+
clientHandler.failRequest(request.getRequestId(), future.cause());
6362
}
6463
});
6564
}

‎rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/protocol/http2/Http2Protocol.java‎

Lines changed: 20 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -86,22 +86,29 @@ public void sendRequest(io.netty.channel.Channel channel, RpcRequest request,
8686
RpcConfig rpcConfig = RpcConfig.getInstance();
8787
Serializer serializer = SerializerCode.getSerializerByCode(rpcConfig.getSerializerCode());
8888

89-
// 使用 bootstrap 创建新流
89+
// 使用 bootstrap 异步创建新流
9090
io.netty.handler.codec.http2.Http2StreamChannelBootstrap streamBootstrap = new io.netty.handler.codec.http2.Http2StreamChannelBootstrap(
9191
channel);
92-
Http2StreamChannel streamChannel = streamBootstrap.open().get(5, java.util.concurrent.TimeUnit.SECONDS);
93-
94-
// 在流通道中构建完整的处理链
95-
streamChannel.pipeline().addLast(new Http2StreamFrameToHttpObjectCodec(false));
96-
streamChannel.pipeline().addLast(new io.netty.handler.codec.http.HttpObjectAggregator(512 * 1024));
97-
streamChannel.pipeline().addLast(new HttpRpcEncoder(serializer));
98-
streamChannel.pipeline().addLast(new HttpRpcDecoder(serializer, RpcResponse.class));
99-
streamChannel.pipeline().addLast(clientHandler);
100-
101-
streamChannel.writeAndFlush(request).addListener(future -> {
102-
if (!future.isSuccess()) {
103-
clientHandler.getFuture().completeExceptionally(future.cause());
92+
93+
streamBootstrap.open().addListener(f -> {
94+
if (!f.isSuccess()) {
95+
clientHandler.failRequest(request.getRequestId(), f.cause());
96+
return;
10497
}
98+
99+
Http2StreamChannel streamChannel = (Http2StreamChannel) f.getNow();
100+
// 在流通道中构建完整的处理链
101+
streamChannel.pipeline().addLast(new Http2StreamFrameToHttpObjectCodec(false));
102+
streamChannel.pipeline().addLast(new io.netty.handler.codec.http.HttpObjectAggregator(512 * 1024));
103+
streamChannel.pipeline().addLast(new HttpRpcEncoder(serializer));
104+
streamChannel.pipeline().addLast(new HttpRpcDecoder(serializer, RpcResponse.class));
105+
streamChannel.pipeline().addLast(clientHandler);
106+
107+
streamChannel.writeAndFlush(request).addListener(writeFuture -> {
108+
if (!writeFuture.isSuccess()) {
109+
clientHandler.failRequest(request.getRequestId(), writeFuture.cause());
110+
}
111+
});
105112
});
106113
}
107114
}

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

Lines changed: 3 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -43,16 +43,13 @@ public void config(ChannelPipeline pipeline, boolean isServer, io.netty.channel.
4343
public void sendRequest(io.netty.channel.Channel channel, RpcRequest request,
4444
com.xiaoyu.rpc.core.client.NettyRpcClientHandler clientHandler) throws Exception {
4545
// Netty 协议直接复用主通道
46-
// 如果 pipeline 里还没有 handler (第一次)，加上它
47-
if (channel.pipeline().get(com.xiaoyu.rpc.core.client.NettyRpcClientHandler.class) != null) {
48-
channel.pipeline().replace(com.xiaoyu.rpc.core.client.NettyRpcClientHandler.class, "handler", clientHandler);
49-
} else {
50-
channel.pipeline().addLast("handler", clientHandler);
46+
if (channel.pipeline().get(com.xiaoyu.rpc.core.client.NettyRpcClientHandler.class) == null) {
47+
channel.pipeline().addLast(clientHandler);
5148
}
5249

5350
channel.writeAndFlush(request).addListener(future -> {
5451
if (!future.isSuccess()) {
55-
clientHandler.getFuture().completeExceptionally(future.cause());
52+
clientHandler.failRequest(request.getRequestId(), future.cause());
5653
}
5754
});
5855
}

‎rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/server/NettyRpcHandler.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@ public class NettyRpcHandler extends SimpleChannelInboundHandler<RpcRequest> {
2525
@Override
2626
protected void channelRead0(ChannelHandlerContext ctx, RpcRequest request) throws Exception {
2727
RpcResponse.Builder responseBuilder = RpcResponse.newBuilder();
28+
responseBuilder.setRequestId(request.getRequestId());
2829

2930
try {
3031
// 1. 获取实现类 (从 ServiceRepository 获取)

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

Lines changed: 25 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -54,7 +54,6 @@ protected void initChannel(SocketChannel ch) {
5454

5555
@Override
5656
public Object sendRequest(RpcRequest request, InetSocketAddress address) {
57-
NettyRpcClientHandler clientHandler = new NettyRpcClientHandler();
5857
String protocolName = RpcConfig.getInstance().getProtocol();
5958

6059
try {
@@ -64,11 +63,34 @@ public Object sendRequest(RpcRequest request, InetSocketAddress address) {
6463
throw new RuntimeException("无法连接到服务器: " + address);
6564
}
6665

66+
// Reuse handler from pipeline
67+
NettyRpcClientHandler clientHandler = channel.pipeline().get(NettyRpcClientHandler.class);
68+
if (clientHandler == null) {
69+
// Should be added by initChannel, but for safety in some custom protocols:
70+
clientHandler = new NettyRpcClientHandler();
71+
channel.pipeline().addLast(clientHandler);
72+
}
73+
74+
// Generate ID and set to request
75+
String requestId = java.util.UUID.randomUUID().toString();
76+
// Use reflection or builder if setter not available, but since we regenerated
77+
// proto, we should use builder properly
78+
// However, RpcRequest is immutable if generated by Proto?
79+
// Wait, RpcRequest.java is generated. It has a Builder.
80+
// But the signature of sendRequest takes RpcRequest, which is already built.
81+
// We need to rebuild it with the ID.
82+
83+
RpcRequest.Builder builder = request.toBuilder();
84+
builder.setRequestId(requestId);
85+
RpcRequest newRequest = builder.build();
86+
6787
CompletableFuture<Object> resultFuture = new CompletableFuture<>();
68-
clientHandler.setFuture(resultFuture);
88+
clientHandler.addFuture(requestId, resultFuture);
6989

7090
Protocol protocol = ProtocolFactory.getProtocol(protocolName);
71-
protocol.sendRequest(channel, request, clientHandler);
91+
// Pass the handler so protocol can ensure it's in pipeline if needed (double
92+
// check)
93+
protocol.sendRequest(channel, newRequest, clientHandler);
7294

7395
Object result = resultFuture.get(5, TimeUnit.SECONDS);
7496

@@ -78,11 +100,6 @@ public Object sendRequest(RpcRequest request, InetSocketAddress address) {
78100
if (!"Success".equals(rpcResponse.getMessage())) {
79101
throw new RuntimeException("服务端报错: " + rpcResponse.getMessage());
80102
}
81-
82-
// 返回 Data 的 bytes，由上层可以通过配置的 Serializer 进行反序列化
83-
// 这里为了保持兼容性，还是直接返回 Response 对象或者 Data？
84-
// 根据 RpcClient 的逻辑，它是在这里反序列化的。
85-
// 此时我们其实只负责传输，最好返回 RpcResponse 对象。
86103
return rpcResponse;
87104

88105
} else {

0 commit comments

Comments
 (0)