使用netty4.x客户端接收较大数据量报文时发生的读取不完整bug修复记录

 1 ClientHandler clientHandler = new ClientHandler(this);
 2             bootstrap.group(eventLoop)
 3                     .channel(NioSocketChannel.class)
 4                     .option(ChannelOption.TCP_NODELAY, true)
 5                     .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000)
 6                     .option(ChannelOption.MAX_MESSAGES_PER_READ, Integer.MAX_VALUE)
 7                     .handler(new ClientChannelInitializer(clientHandler));
 8             
 9             ChannelFuture f = bootstrap.connect(host, port).sync();
10 
11             f.channel().closeFuture().sync();

 1 private class ClientChannelInitializer extends ChannelInitializer<SocketChannel> {
 2         private ClientHandler clientHandler;
 3 
 4         public ClientChannelInitializer(ClientHandler clientHandler) {
 5             this.clientHandler = clientHandler;
 6         }
 7 
 8         @Override
 9         protected void initChannel(SocketChannel socketChannel) throws Exception {
10 
11             socketChannel.pipeline().addLast(new SplDecoder());
12             socketChannel.pipeline().addLast(clientHandler);
13             channel = socketChannel;
14         }
15     }

// 解决问题前 initChannel的实现是这样的，使用了netty内部的长度字段解码器
@Override
protected void initChannel(SocketChannel ch) throws Exception {
    ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE,4,4,-8,0));
    ch.pipeline().addLast(clientHandler);
}

public class ClientHandler extends ChannelInboundHandlerAdapter {

 1 @Override
 2     public void channelActive(ChannelHandlerContext context) throws Exception {
 3         logger.info("Ready to send request...");
 4         ByteBuffer result = getByteBuffer();
 5         ByteBuf buf = Unpooled.buffer(result.remaining());
 6         buf.writeBytes(result);
 7 
 8         context.writeAndFlush(buf);
 9     }
10 
11     @Override
12     public void channelRead(ChannelHandlerContext context, Object msg) throws Exception {
13         logger.info("Get server response...");
14 
15         String[] result = (String[]) msg;
16 
17         logger.debug("response xml is : " + result[1]);
18         client.setResponse(result);
19 
20         ok = true;
21     }
22 
23     @Override
24     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
25         ctx.flush();
26         if (ok) {
27             ctx.close();
28         } else {
29             ctx.read();
30         }
31     }