LengthFieldBasedFrameDecoder使用

LengthFieldBasedFrameDecoder解码器，会读取byteBuf中的前几个字节，获取数据长度。

作用是防止粘包。

如果客户端也是netty，那么可以配合LengthFieldPrepender使用。LengthFieldPrepender会自动在原始数据前面加上数据长度。

socketChannel.pipeline().addLast(new LengthFieldBasedFrameDecoder(65535,0,2,0,2));

第二个参数指定lengthFieldOffset，也就是长度字段的偏移距，一般设置为0 第三个参数指定lengthFieldLength，也就是几个字节，2个字节的话，一个frame最多65535个字节，short类型的最大值。 lengthAdjustment, initialBytesToStrip，后面2个参数，lengthAdjustment一般为0，initialBytesToStrip和lengthFieldLength保持一致。

使用示例 服务端

public class EchoServer { public static void main(String[] args) { EventLoopGroup bossGroup = new NioEventLoopGroup(); EventLoopGroup workerGroup = new NioEventLoopGroup(); try { ServerBootstrap serverBootstrap = new ServerBootstrap(); serverBootstrap.group(bossGroup,workerGroup) .channel(NioServerSocketChannel.class) .option(ChannelOption.SO_BACKLOG,100) .option(ChannelOption.CONNECT_TIMEOUT_MILLIS,3000) .childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel socketChannel) throws Exception { // ByteBuf delimiter = Unpooled.copiedBuffer("$_".getBytes()); // ByteBuf delimiter0 = Unpooled.copiedBuffer("A".getBytes()); // ByteBuf delimiter1 = Unpooled.copiedBuffer("B".getBytes()); // socketChannel.pipeline().addLast(new DelimiterBasedFrameDecoder(1024,delimiter,delimiter0,delimiter1)); // socketChannel.pipeline().addLast(new FixedLengthFrameDecoder(10)); // socketChannel.pipeline().addLast(new StringDecoder()); socketChannel.pipeline().addLast(new LengthFieldBasedFrameDecoder(65535,0,2,0,2)); socketChannel.pipeline().addLast(new MsgpackDecoder()); socketChannel.pipeline().addLast(new LengthFieldPrepender(2)); socketChannel.pipeline().addLast(new MsgpackEncoder()); socketChannel.pipeline().addLast(new EchoServerHandler()); } }); ChannelFuture channelFuture =serverBootstrap.bind(7799).sync(); channelFuture.channel().closeFuture().sync(); } catch (InterruptedException e) { e.printStackTrace(); } finally { bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } } } public class EchoServerHandler extends SimpleChannelInboundHandler<Object> { @Override protected void channelRead0(ChannelHandlerContext channelHandlerContext, Object o) throws Exception { System.out.println(o); // o +="$_"; // ByteBuf echo = Unpooled.copiedBuffer(o.getBytes()); channelHandlerContext.writeAndFlush(o); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { cause.printStackTrace(); ctx.close(); }

客户端

public class EchoClient { public void connect(int port,String host){ EventLoopGroup group = new NioEventLoopGroup(); try { Bootstrap b = new Bootstrap(); b.group(group).channel(NioSocketChannel.class) .option(ChannelOption.TCP_NODELAY,true) .handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel socketChannel) throws Exception { // ByteBuf delimiter = Unpooled.copiedBuffer("$_".getBytes()); // socketChannel.pipeline().addLast(new DelimiterBasedFrameDecoder(1024,delimiter)); // socketChannel.pipeline().addLast(new StringDecoder()); socketChannel.pipeline().addLast(new LengthFieldBasedFrameDecoder(65535,0,2,0,2)); socketChannel.pipeline().addLast(new MsgpackDecoder()); socketChannel.pipeline().addLast(new LengthFieldPrepender(2)); socketChannel.pipeline().addLast(new MsgpackEncoder()); socketChannel.pipeline().addLast(new EchoClientHandler()); } }); ChannelFuture channelFuture=b.connect(host,port).sync(); channelFuture.channel().closeFuture().sync(); } catch (InterruptedException e) { e.printStackTrace(); }finally { group.shutdownGracefully(); } } public static void main(String[] args) { EchoClient echoClient = new EchoClient(); echoClient.connect(7799,"127.0.0.1"); } } public class EchoClientHandler extends SimpleChannelInboundHandler<Object> { public static final String STRINGTAG="HELLO,这是什"; @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { List<UserInfo> userInfos = new ArrayList<>(); for (int i =0;i<2;i++){ UserInfo userInfo = new UserInfo(); userInfo.setName("张三"+i); userInfo.setAge(i); userInfos.add(userInfo); ctx.write(userInfo); } ctx.flush(); // ctx.writeAndFlush(userInfos); // for (int i=0;i<10;i++){ // ctx.writeAndFlush(Unpooled.copiedBuffer(STRINGTAG.getBytes())); // Thread.sleep(3000); // ctx.writeAndFlush(Unpooled.copiedBuffer("么你A好吗$_不是B".getBytes())); // ctx.writeAndFlush(Unpooled.copiedBuffer(new byte[1024])); // } } @Override protected void channelRead0(ChannelHandlerContext channelHandlerContext, Object o) throws Exception { System.out.println(o); } }