美文网首页netty
Netty4实战 - TCP粘包&拆包解决方案

Netty4实战 - TCP粘包&拆包解决方案

作者: FX_SKY | 来源:发表于2017-05-19 18:06 被阅读2106次

    Netty是目前业界最流行的NIO框架之一,它的健壮性、高性能、可定制和可扩展性在同类框架中都是首屈一指。它已经得到了成百上千的商业项目的验证,例如Hadoop的RPC框架Avro就使用了Netty作为底层通信框架,其他的业界主流RPC框架,例如:Dubbo、Google 开源的gRPC、新浪微博开源的Motan、Twitter 开源的 finagle也使用Netty来构建高性能的异步通信能力。另外,阿里巴巴开源的消息中间件RocketMQ也使用Netty作为底层通信框架。

    TCP黏包/拆包

    TCP是一个“流”协议,所谓流,就是没有界限的一长串二进制数据。TCP作为传输层协议并不不了解上层业务数据的具体含义,它会根据TCP缓冲区的实际情况进行数据包的划分,所以在业务上认为是一个完整的包,可能会被TCP拆分成多个包进行发送,也有可能把多个小的包封装成一个大的数据包发送,这就是所谓的TCP粘包和拆包问题。

    粘包问题的解决策略

    由于底层的TCP无法理解上层的业务数据,所以在底层是无法保证数据包不被拆分和重组的,这个问题只能通过上层的应用协议栈设计来解决。业界的主流协议的解决方案,可以归纳如下:

    1. 消息定长,报文大小固定长度,例如每个报文的长度固定为200字节,如果不够空位补空格;
    2. 包尾添加特殊分隔符,例如每条报文结束都添加回车换行符(例如FTP协议)或者指定特殊字符作为报文分隔符,接收方通过特殊分隔符切分报文区分;
    3. 将消息分为消息头和消息体,消息头中包含表示信息的总长度(或者消息体长度)的字段;
    4. 更复杂的自定义应用层协议。

    Netty粘包和拆包解决方案

    Netty提供了多个解码器,可以进行分包的操作,分别是:

    • LineBasedFrameDecoder
    • DelimiterBasedFrameDecoder(添加特殊分隔符报文来分包)
    • FixedLengthFrameDecoder(使用定长的报文来分包)
    • LengthFieldBasedFrameDecoder

    LineBasedFrameDecoder解码器

    LineBasedFrameDecoder是回车换行解码器,如果用户发送的消息以回车换行符作为消息结束的标识,则可以直接使用Netty的LineBasedFrameDecoder对消息进行解码,只需要在初始化Netty服务端或者客户端时将LineBasedFrameDecoder正确的添加到ChannelPipeline中即可,不需要自己重新实现一套换行解码器。

    Netty依赖

            <dependency>
                <groupId>io.netty</groupId>
                <artifactId>netty-all</artifactId>
                <version>4.1.9.Final</version>
            </dependency>
    

    1.1 Server端

    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.*;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    import io.netty.handler.codec.LineBasedFrameDecoder;
    import io.netty.handler.codec.string.StringDecoder;
    import io.netty.handler.codec.string.StringEncoder;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    /**
     *
     * @author Ricky
     *
     */
    public class LineBasedServer {
        private Logger logger = LoggerFactory.getLogger(getClass());
    
        public void bind(int port) throws Exception {
    
            EventLoopGroup bossGroup = new NioEventLoopGroup();
            EventLoopGroup workerGroup = new NioEventLoopGroup();
            try {
                ServerBootstrap b = new ServerBootstrap();
                b.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .option(ChannelOption.SO_BACKLOG, 1024)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch) throws Exception {
    
                        ChannelPipeline p = ch.pipeline();
                        p.addLast(new LineBasedFrameDecoder(1024));
                        p.addLast(new StringDecoder());
                        p.addLast(new StringEncoder());
    
                        p.addLast(new LineServerHandler());
                    }
                });
    
                // Bind and start to accept incoming connections.
                ChannelFuture f = b.bind(port).sync(); // (7)
    
                logger.info("server bind port:{}", port);
    
                // Wait until the server socket is closed.
                f.channel().closeFuture().sync();
    
            } finally {
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    
        public static void main(String[] args) throws Exception {
    
            new LineBasedServer().bind(Constants.PORT);
        }
    }
    
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    public class LineServerHandler extends ChannelInboundHandlerAdapter {
        private Logger logger = LoggerFactory.getLogger(getClass());
    
        private int count = 0;
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg)
                throws Exception {
    
            count++;
            String body = (String) msg;
            logger.info("server read msg:{}, count:{}", body, count);
    
            String response = "hello from server"+System.getProperty("line.separator");
            ctx.writeAndFlush(response);
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
                throws Exception {
            logger.error("server caught exception", cause);
            ctx.close();
        }
    
    }
    

    1.2 Client

    import io.netty.bootstrap.Bootstrap;
    import io.netty.channel.*;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioSocketChannel;
    import io.netty.handler.codec.LineBasedFrameDecoder;
    import io.netty.handler.codec.string.StringDecoder;
    import io.netty.handler.codec.string.StringEncoder;
    
    public class LineBasedClient {
    
        public void connect(String host, int port) throws InterruptedException {
    
            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 ch) throws Exception {
    
                        ChannelPipeline p = ch.pipeline();
                        p.addLast(new LineBasedFrameDecoder(1024));
                        p.addLast(new StringDecoder());
                        p.addLast(new StringEncoder());
    
                        p.addLast(new LineClientHandler());
                    }
                });
    
                ChannelFuture future = b.connect(Constants.HOST, Constants.PORT).sync();
    
                future.channel().closeFuture().sync();
            } finally {
                group.shutdownGracefully();
            }
        }
    
        public static void main(String[] args) throws Exception {
    
            new LineBasedClient().connect(Constants.HOST, Constants.PORT);
        }
    }
    
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    public class LineClientHandler extends ChannelInboundHandlerAdapter {
        private Logger logger = LoggerFactory.getLogger(getClass());
    
        private int count =0;
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            // Send the message to Server
           for(int i=0; i<100; i++){
    
               String msg = "hello from client "+i;
               logger.info("client send message:{}", msg);
    
               ctx.writeAndFlush(msg+System.getProperty("line.separator"));
           }
    
        }
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg)
                throws Exception {
            String body = (String) msg;
            count++;
            logger.info("client read msg:{}, count:{}", body, count);
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
                throws Exception {
            logger.error("client caught exception", cause);
            ctx.close();
        }
    }
    

    DelimiterBasedFrameDecoder解码器

    DelimiterBasedFrameDecoder是分隔符解码器,用户可以指定消息结束的分隔符,它可以自动完成以分隔符作为码流结束标识的消息的解码。回车换行解码器实际上是一种特殊的DelimiterBasedFrameDecoder解码器。

    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.*;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    import io.netty.handler.codec.DelimiterBasedFrameDecoder;
    import io.netty.handler.codec.string.StringDecoder;
    import io.netty.handler.codec.string.StringEncoder;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    /**
     * 
     * @author Ricky
     *
     */
    public class DelimiterServer {
        private Logger logger = LoggerFactory.getLogger(getClass());
    
    
        public void bind(int port) throws Exception {
    
            EventLoopGroup bossGroup = new NioEventLoopGroup(1);
            EventLoopGroup workerGroup = new NioEventLoopGroup();
            try {
                ServerBootstrap b = new ServerBootstrap();
                b.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .option(ChannelOption.SO_BACKLOG, 1024)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch) throws Exception {
    
                        ChannelPipeline p = ch.pipeline();
                        p.addLast(new DelimiterBasedFrameDecoder(1024, Unpooled.copiedBuffer(Constants.DELIMITER.getBytes())));
                        p.addLast(new StringDecoder());
                        p.addLast(new StringEncoder());
    
                        p.addLast(new DelimiterServerHandler());
                    }
                });
    
                // Bind and start to accept incoming connections.
                ChannelFuture f = b.bind(port).sync(); // (7)
    
                logger.info("server bind port:{}", port);
    
                // Wait until the server socket is closed.
                f.channel().closeFuture().sync();
    
            } finally {
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    
        public static void main(String[] args) throws Exception {
    
            new DelimiterServer().bind(Constants.PORT);
        }
    }
    

    Client:

    import io.netty.bootstrap.Bootstrap;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.*;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioSocketChannel;
    import io.netty.handler.codec.DelimiterBasedFrameDecoder;
    import io.netty.handler.codec.string.StringDecoder;
    import io.netty.handler.codec.string.StringEncoder;
    
    public class DelimiterClient {
        
        public void connect(String host, int port) throws InterruptedException {
    
            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 ch) throws Exception {
    
                        ChannelPipeline p = ch.pipeline();
                        p.addLast(new DelimiterBasedFrameDecoder(1024, Unpooled.copiedBuffer(Constants.DELIMITER.getBytes())));
                        p.addLast(new StringDecoder());
                        p.addLast(new StringEncoder());
    
                        p.addLast(new DelimiterClientHandler());
                    }
                });
    
                ChannelFuture future = b.connect(Constants.HOST, Constants.PORT).sync();
    
                future.channel().closeFuture().sync();
            } finally {
                group.shutdownGracefully();
            }
        }
    
        public static void main(String[] args) throws Exception {
    
            new DelimiterClient().connect(Constants.HOST, Constants.PORT);
        }
    }
    

    FixedLengthFrameDecoder解码器

    FixedLengthFrameDecoder是固定长度解码器,它能够按照指定的长度对消息进行自动解码,开发者不需要考虑TCP的粘包/拆包等问题,非常实用。

    对于定长消息,如果消息实际长度小于定长,则往往会进行补位操作,它在一定程度上导致了空间和资源的浪费。但是它的优点也是非常明显的,编解码比较简单,因此在实际项目中仍然有一定的应用场景。

    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.*;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    import io.netty.handler.codec.FixedLengthFrameDecoder;
    import io.netty.handler.codec.string.StringDecoder;
    import io.netty.handler.codec.string.StringEncoder;
    import io.netty.handler.logging.LogLevel;
    import io.netty.handler.logging.LoggingHandler;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    /**
     * @author Ricky Fung
     */
    public class NettyServer {
        private Logger logger = LoggerFactory.getLogger(getClass());
    
        public void bind(int port) throws InterruptedException {
    
            EventLoopGroup bossGroup = new NioEventLoopGroup(1);
            EventLoopGroup workerGroup = new NioEventLoopGroup();
            try {
                //配置服务器启动类
                ServerBootstrap b = new ServerBootstrap();
                b.group(bossGroup, workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .option(ChannelOption.SO_BACKLOG, 100)
                        .handler(new LoggingHandler(LogLevel.INFO))//配置日志输出
                        .childHandler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel ch)
                                    throws Exception {
                                ch.pipeline().addLast(new FixedLengthFrameDecoder(1<<5));
                                ch.pipeline().addLast(new StringDecoder());
                                ch.pipeline().addLast(new StringEncoder());
    
                                ch.pipeline().addLast(new ServerHandler());
                            }
                        });
    
                ChannelFuture f = b.bind(port).sync();
                //等待服务器退出
                f.channel().closeFuture().sync();
            } finally {
                //释放线程资源
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    
        private class ServerHandler extends ChannelInboundHandlerAdapter {
            private int counter = 0;
    
            @Override
            public void channelActive(ChannelHandlerContext ctx) throws Exception {
                super.channelActive(ctx);
            }
    
            @Override
            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    
                logger.info("接收客户端msg:{}", msg);
    
                ByteBuf echo = Unpooled.copiedBuffer(String.format("Hello from server:", counter).getBytes());
                ctx.writeAndFlush(echo);
            }
        }
    
        public static void main(String[] args) throws InterruptedException {
    
            new NettyServer().bind(Constants.PORT);
        }
    }
    
    import io.netty.bootstrap.Bootstrap;
    import io.netty.channel.*;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioSocketChannel;
    import io.netty.handler.codec.FixedLengthFrameDecoder;
    import io.netty.handler.codec.string.StringDecoder;
    import io.netty.handler.codec.string.StringEncoder;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    /**
     * @author Ricky Fung
     */
    public class NettyClient {
    
        private Logger logger = LoggerFactory.getLogger(getClass());
    
        public void connect(String host, int port) throws InterruptedException {
            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 ch) throws Exception {
    
                                ch.pipeline().addLast(new FixedLengthFrameDecoder(1<<5));
                                ch.pipeline().addLast(new StringDecoder());
                                ch.pipeline().addLast(new StringEncoder());
    
                                ch.pipeline().addLast(new ClientHandler());
                            }
                        });
    
                ChannelFuture future = b.connect(host, port).sync();
    
                future.channel().closeFuture().sync();
            } finally {
                group.shutdownGracefully();
            }
        }
    
        private class ClientHandler extends ChannelInboundHandlerAdapter {
    
            @Override
            public void channelActive(ChannelHandlerContext ctx) throws Exception {
    
                for(int i=0; i<100; i++){
    
                    String msg = "hello from client "+i;
                    logger.info("client send message:{}", msg);
    
                    ctx.writeAndFlush(msg);
                }
            }
    
            @Override
            public void channelRead(ChannelHandlerContext ctx, Object msg)
                    throws Exception {
    
                logger.info("接收服务端msg:{}", msg);
            }
    
            @Override
            public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
                    throws Exception {
                cause.printStackTrace();
                ctx.close();
            }
    
        }
    
        public static void main(String[] args) throws InterruptedException {
    
            new NettyClient().connect(Constants.HOST, Constants.PORT);
        }
    }
    

    LengthFieldBasedFrameDecoder解码器

    大多数的协议(私有或者公有),协议头中会携带长度字段,用于标识消息体或者整包消息的长度,例如SMPP、HTTP协议等。由于基于长度解码需求的通用性,以及为了降低用户的协议开发难度,Netty提供了LengthFieldBasedFrameDecoder,自动屏蔽TCP底层的拆包和粘包问题,只需要传入正确的参数,即可轻松解决“读半包“问题。

    Message.java

    import java.nio.charset.Charset;
    
    /**
     * @author Ricky Fung
     */
    public class Message {
    
        private final Charset charset = Charset.forName("utf-8");
    
        private byte magicType;
        private byte type;//消息类型  0xAF 表示心跳包    0xBF 表示超时包  0xCF 业务信息包
        private long requestId; //请求id
        private int length;
        private String body;
    
        public Message(){
    
        }
    
        public Message(byte magicType, byte type, long requestId, byte[] data) {
            this.magicType = magicType;
            this.type = type;
            this.requestId = requestId;
            this.length = data.length;
            this.body = new String(data, charset);
        }
    
        public Message(byte magicType, byte type, long requestId, String body) {
            this.magicType = magicType;
            this.type = type;
            this.requestId = requestId;
            this.length = body.getBytes(charset).length;
            this.body = body;
        }
        ...setter/getter
    }
    

    MessageDecoder.java

    import com.mindflow.netty4.unpack.model.Message;
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    /**
     * @author Ricky Fung
     */
    public class MessageDecoder extends LengthFieldBasedFrameDecoder {
        private Logger logger = LoggerFactory.getLogger(getClass());
    
        //头部信息的大小应该是 byte+byte+int = 1+1+8+4 = 14
        private static final int HEADER_SIZE = 14;
    
        public MessageDecoder(int maxFrameLength, int lengthFieldOffset, int lengthFieldLength) {
            super(maxFrameLength, lengthFieldOffset, lengthFieldLength);
        }
    
        @Override
        protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
            if (in == null) {
                return null;
            }
    
            if (in.readableBytes() <= HEADER_SIZE) {
                return null;
            }
    
            in.markReaderIndex();
    
            byte magic = in.readByte();
            byte type = in.readByte();
            long requestId = in.readLong();
            int dataLength = in.readInt();
    
            // FIXME 如果dataLength过大,可能导致问题
            if (in.readableBytes() < dataLength) {
                in.resetReaderIndex();
                return null;
            }
    
            byte[] data = new byte[dataLength];
            in.readBytes(data);
    
            String body = new String(data, "UTF-8");
            Message msg = new Message(magic, type, requestId, body);
            return msg;
        }
    }
    

    MessageEncoder.java

    import com.mindflow.netty4.unpack.model.Message;
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.handler.codec.MessageToByteEncoder;
    import java.nio.charset.Charset;
    
    /**
     * @author Ricky Fung
     */
    public class MessageEncoder extends MessageToByteEncoder<Message> {
        private final Charset charset = Charset.forName("utf-8");
    
        @Override
        protected void encode(ChannelHandlerContext ctx, Message msg, ByteBuf out) throws Exception {
    
            //
            out.writeByte(msg.getMagicType());
            out.writeByte(msg.getType());
            out.writeLong(msg.getRequestId());
    
            byte[] data = msg.getBody().getBytes(charset);
            out.writeInt(data.length);
            out.writeBytes(data);
        }
    }
    
    

    服务端:

    import com.mindflow.netty4.unpack.model.Message;
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.*;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    /**
     * @author Ricky Fung
     */
    public class NettyServer {
        private Logger logger = LoggerFactory.getLogger(this.getClass());
    
        public void bind(int port) throws InterruptedException {
    
            EventLoopGroup bossGroup = new NioEventLoopGroup(1);
            EventLoopGroup workerGroup = new NioEventLoopGroup();
            try {
                ServerBootstrap b = new ServerBootstrap();
                b.group(bossGroup, workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .option(ChannelOption.SO_BACKLOG, 128)
                        .childHandler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            public void initChannel(SocketChannel ch) throws Exception {
    
                                ChannelPipeline p = ch.pipeline();
                                p.addLast(new MessageDecoder(1<<20, 10, 4));
                                p.addLast(new MessageEncoder());
                                p.addLast(new ServerHandler());
                            }
                        });
    
                // Bind and start to accept incoming connections.
                ChannelFuture future = b.bind(port).sync(); // (7)
    
                logger.info("server bind port:{}", port);
    
                // Wait until the server socket is closed.
                future.channel().closeFuture().sync();
    
            } finally {
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    
        private class ServerHandler extends SimpleChannelInboundHandler<Message> {
    
            @Override
            protected void channelRead0(ChannelHandlerContext ctx, Message msg) throws Exception {
    
                logger.info("server read msg:{}", msg);
    
                Message resp = new Message(msg.getMagicType(), msg.getType(), msg.getRequestId(), "Hello world from server");
                ctx.writeAndFlush(resp);
            }
        }
    
        public static void main(String[] args) throws Exception {
    
            new NettyServer().bind(Constants.PORT);
        }
    }
    

    客户端:

    import com.mindflow.netty4.unpack.model.Message;
    import io.netty.bootstrap.Bootstrap;
    import io.netty.channel.*;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioSocketChannel;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import java.util.concurrent.TimeUnit;
    
    /**
     * @author Ricky Fung
     */
    public class NettyClient {
    
        public void connect(String host, int port) throws InterruptedException {
            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 ch) throws Exception {
    
                                ChannelPipeline p = ch.pipeline();
                                p.addLast(new MessageDecoder(1<<20, 10, 4));
                                p.addLast(new MessageEncoder());
    
                                p.addLast(new ClientHandler());
                            }
                        });
    
                ChannelFuture future = b.connect(host, port).sync();
    
                future.awaitUninterruptibly(2000, TimeUnit.MILLISECONDS);
                if(future.channel().isActive()){
    
                    for(int i=0; i<100; i++) {
    
                        String body = "Hello world from client:"+ i;
                        Message msg = new Message((byte) 0XAF, (byte) 0XBF, i, body);
    
                        future.channel().writeAndFlush(msg);
                    }
                }
    
                future.channel().closeFuture().sync();
            } finally {
                group.shutdownGracefully();
            }
        }
    
        private class ClientHandler extends ChannelInboundHandlerAdapter {
            private Logger logger = LoggerFactory.getLogger(getClass());
    
            @Override
            public void channelRead(ChannelHandlerContext ctx, Object msg)
                    throws Exception {
    
                logger.info("client read msg:{}, ", msg);
    
            }
    
            @Override
            public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
                    throws Exception {
                logger.error("client caught exception", cause);
                ctx.close();
            }
        }
    
        public static void main(String[] args) throws Exception {
    
            new NettyClient().connect(Constants.HOST, Constants.PORT);
        }
    }
    

    参考资料

    Netty API Reference

    Netty系列之Netty编解码框架分析

    源码下载

    https://github.com/TiFG/netty4-in-action/tree/master/netty4-unpack-demo

    相关文章

      网友评论

        本文标题:Netty4实战 - TCP粘包&拆包解决方案

        本文链接:https://www.haomeiwen.com/subject/iynyxxtx.html