java netty客户端只能从服务器接收一个响应
最近,我使用Netty4编写了一个客户端和服务器,客户端向服务器发送五条消息,但客户端只能从服务器收到一条响应,下面是我的代码
DTO:RpcRequest
package com.huawei.simple.rpc.base.request.dto;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;
import lombok.ToString;
import java.lang.reflect.Parameter;
@Setter
@Getter
@NoArgsConstructor
@AllArgsConstructor
@ToString
public class RpcRequest {
private String requestId;
private String interfaceName;
private String methodName;
}
DTO:Rp应答
package com.huawei.simple.rpc.base.response.dto;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;
import lombok.ToString;
@Setter
@Getter
@NoArgsConstructor
@AllArgsConstructor
@ToString
public class RpcResponse {
private String requestId;
private String responseMessage;
private int responseCode;
private Object responseBody;
}
编解码器
public class ResponseDecoder extends ByteToMessageDecoder {
private static ObjectMapper mapper = new ObjectMapper();
@Override
protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception {
if (byteBuf.readableBytes() < 4) {
return;
}
byteBuf.markReaderIndex();
int dataLength = byteBuf.readInt();
if (byteBuf.readableBytes() < dataLength) {
byteBuf.resetReaderIndex();
return;
}
byte[] data = new byte[dataLength];
byteBuf.readBytes(data);
RpcResponse response = mapper.readValue(data, RpcResponse.class);
list.add(response);
}
}
public class ResponseEncoder extends MessageToByteEncoder<RpcResponse> {
private static ObjectMapper mapper = new ObjectMapper();
@Override
protected void encode(ChannelHandlerContext channelHandlerContext, RpcResponse rpcResponse, ByteBuf byteBuf) throws Exception {
byte[] bytes = mapper.writeValueAsBytes(rpcResponse);
int dataLength = bytes.length;
byteBuf.writeInt(dataLength);
byteBuf.writeBytes(bytes);
byteBuf.writeBytes(SerializationUtil.serialize(rpcResponse));
}
}
public class RequestDecoder extends ByteToMessageDecoder {
private static ObjectMapper mapper = new ObjectMapper();
@Override
protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception {
if (byteBuf.readableBytes() < 4) {
return;
}
byteBuf.markReaderIndex();
int dataLength = byteBuf.readInt();
if (byteBuf.readableBytes() < dataLength) {
byteBuf.resetReaderIndex();
return;
}
byte[] data = new byte[dataLength];
byteBuf.readBytes(data);
RpcRequest request = mapper.readValue(data, RpcRequest.class);
list.add(request);
}
}
public class RequestEncoder extends MessageToByteEncoder<RpcRequest> {
private static ObjectMapper mapper = new ObjectMapper();
@Override
protected void encode(ChannelHandlerContext channelHandlerContext, RpcRequest rpcRequest, ByteBuf byteBuf) throws Exception {
byte[] bytes = mapper.writeValueAsBytes(rpcRequest);
int dataLength = bytes.length;
byteBuf.writeInt(dataLength);
byteBuf.writeBytes(bytes);
}
}
客户
public class RpcClient {
public static void main(String[] args) throws Exception {
RpcClient server = new RpcClient();
server.build();
}
public void build() {
NioEventLoopGroup boss = new NioEventLoopGroup();
Bootstrap s = new Bootstrap();
s.group(boss).channel(NioSocketChannel.class)
.option(ChannelOption.TCP_NODELAY,true)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
socketChannel.pipeline().addLast(new RequestEncoder());
socketChannel.pipeline().addLast(new ResponseDecoder());
socketChannel.pipeline().addLast(new RpcClientHandler());
}
});
try {
ChannelFuture future = s.connect(new InetSocketAddress("localhost",8080)).sync();
future.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
boss.shutdownGracefully();
}
}
}
public class RpcClientHandler extends ChannelInboundHandlerAdapter {
private static AtomicInteger id = new AtomicInteger(1);
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
for (int i = 0; i < 5; i++) {
RpcRequest request = new RpcRequest();
request.setRequestId(id.getAndIncrement() + "");
request.setMethodName("helloWorld");
request.setInterfaceName("interface");
ChannelFuture future = ctx.writeAndFlush(request);
future.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture channelFuture) throws Exception {
if (channelFuture.isSuccess()) {
System.out.println("Success");
}
}
});
TimeUnit.SECONDS.sleep(5);
}
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
super.channelInactive(ctx);
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
RpcResponse response = (RpcResponse) msg;
System.out.println("get reponse from server " + response.toString());
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
super.channelReadComplete(ctx);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
super.exceptionCaught(ctx, cause);
}
}
服务器
public class RpcServer {
public static void main(String[] args) throws Exception {
RpcServer server = new RpcServer();
server.build();
}
public void build() {
NioEventLoopGroup boss = new NioEventLoopGroup();
NioEventLoopGroup worker = new NioEventLoopGroup();
ServerBootstrap s = new ServerBootstrap();
s.group(boss, worker).channel(NioServerSocketChannel.class)
.option(ChannelOption.TCP_NODELAY, true)
.option(ChannelOption.SO_BACKLOG, 10)
.handler(new LoggingHandler(LogLevel.DEBUG))
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
socketChannel.pipeline().addLast(new RequestDecoder());
socketChannel.pipeline().addLast(new ResponseEncoder());
socketChannel.pipeline().addLast(new RpcServerHandler());
}
});
try {
ChannelFuture future = s.bind(new InetSocketAddress("localhost", 8080)).sync();
future.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
boss.shutdownGracefully();
worker.shutdownGracefully();
}
}
}
public class RpcServerHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
RpcRequest request = (RpcRequest) msg;
System.out.println("get request from client " + request.toString());
RpcResponse response = new RpcResponse();
response.setRequestId(request.getRequestId());
ChannelFuture future = ctx.writeAndFlush(response);
future.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture channelFuture) throws Exception {
if (channelFuture.isDone()) {
System.out.println("Done");
} else if (channelFuture.isSuccess()) {
System.out.println("Success");
}
}
});
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
super.channelReadComplete(ctx);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
super.exceptionCaught(ctx, cause);
}
}
输出是 服务器:
- 从客户端RpcRequest获取请求(requestId=1,interfaceName=interface,methodName=helloWorld) 完成
- 从客户端RpcRequest获取请求(requestId=2,interfaceName=interface,methodName=helloWorld) 完成
- 从客户端RpcRequest获取请求(requestId=3,interfaceName=interface,methodName=helloWorld) 完成
- 从客户端RpcRequest获取请求(requestId=4,interfaceName=interface,methodName=helloWorld) 完成
- 从客户端RpcRequest获取请求(requestId=5,interfaceName=interface,methodName=helloWorld) 完成
客户
- 成功
- 成功
- 成功
- 成功
- 成功
- 从服务器RPResponse获取响应(requestId=1, responseMessage=null,responseCode=0,ResponseBy=null)
为什么我的客户端只从服务器得到一个响应
# 1 楼答案
您要么在
ResponseEncoder
内出错,要么在ResponseDecoder
内出错(我猜是第一个编码器将响应写入为:
但这些在解码器中被读取为:
您要么忘记处理解码器中的额外字节,要么发送大量数据
这导致程序看起来不读取,因为客户端实际上正在等待更多数据,因为我假设“serialize”的数据不是以零开始的,这导致解码器内的长度值异常高