The whole process of solving the problem of sticky packet unpacking during TCP communication

[[359421]]

When using TCP protocol for communication, the most common problem is the problem of packet sticking and unpacking. This article will take a look at how to solve the problem of packet sticking and unpacking.

01-TCP packet sticking/unpacking problem and solution

Before solving TCP packet sticking and unpacking, let's take a look at an idea. Let's take a look at a demo of reading an Int data to experience this idea.

1.1 ReplayingDecoder

1. Custom decoder, read an Int from ByteBuf. (Key point, you must understand this code)

 public class IntegerHeaderFrameDecoder extends ByteToMessageDecoder {
 @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf buf, List<Object> out ) throws Exception {
      if (buf.readableBytes() < 4) {
 return ;
 }
 buf.markReaderIndex(); //Mark the current read pointer.
 int length = buf.readInt(); //Read an int from ByteBuf  
      if (buf.readableBytes() < length) {
 buf.resetReaderIndex(); //Restore to the read pointer just marked
 return ;
 }
 out .add (buf.readBytes(length)) ;
 }
 }

2. Optimize using ReplayingDecoder()

 public class IntegerHeaderFrameDecoder extends ReplayingDecoder<Void> {
    protected void decode(ChannelHandlerContext ctx, ByteBuf buf, List<Object> out ) throws Exception {
 out . add (buf.readBytes(buf.readInt()));
 }
 }

3. Instructions for using ReplayingDecoder (key points to understand)

• a uses a special ByteBuf called ReplayingDecoderByteBuf, which extends ByteBuf
• b Rewrote the readXxx() and other methods of ByteBuf, which will first check the readable byte length. Once it is detected that the requirement is not met, REPLAY will be thrown directly (REPLAY inherits ERROR)
• c ReplayingDecoder overrides the callDecode() method of ByteToMessageDecoder, captures Signal and resets the readerIndex of ByteBuf in the catch block.
• d Continue to wait for data until the data is available and continue reading, so as to ensure that the data that needs to be read is read.
• The generic type S in the e class definition is a state machine enumeration class used to record the decoding state, which is used in methods such as state(S s) and checkpoint(S s). You can also use java.lang.Void as a placeholder for simple decoding.

Summarize:

ReplayingDecoder is a subclass of ByteToMessageDecoder, which extends ByteBuf. Since methods such as readXxx() are written, if the current data in ByteBuf is less than the substitute data, it will wait for the data to be sufficient before retrieving the data. This means that the manual implementation of this code can be omitted.

4. Note

 1 Some operations on buffer (readBytes(ByteBuffer dst), retain(), release() and other methods will directly throw exceptions)
 2 In some cases, it may affect performance (such as decoding the same message multiple times)

Inherit ReplayingDecoder, error examples and modifications

 //This is an incorrect example:
 //The message contains 2 integers . The decode method in the code will be called twice. At this time, the queue size is not equal to 2. This code cannot achieve the expected result.
 public class MyDecoder extends ReplayingDecoder<Void> {
    private final Queue< Integer > values ​​= new LinkedList< Integer >();
 @Override
 public void decode(ByteBuf buf, List<Object> out ) throws Exception {
        // A message contains 2 integers.
 values ​​.offer(buf.readInt());
 values ​​.offer(buf.readInt());
        assert values ​​. size () == 2;
 out . add ( values ​​.poll() + values ​​.poll());
 }
 } 

 //Correct approach:
 public class MyDecoder extends ReplayingDecoder<Void> {
    private final Queue< Integer > values ​​= new LinkedList< Integer >();
 @Override
 public void decode(ByteBuf buf, List<Object> out ) throws Exception {
        // Revert the state of the variable that might have been changed
 // since the last   partial decode.
 values ​​.clear();
        // A message contains 2 integers.
 values ​​.offer(buf.readInt());
 values ​​.offer(buf.readInt());
        // Now we know this assertion will never fail.
        assert values ​​. size () == 2;
 out . add ( values ​​.poll() + values ​​.poll());
 }
 }

Implementation of ByteToIntegerDecoder2

 public class ByteToIntegerDecoder2 extends ReplayingDecoder<Void> {
 /**
 * @param ctx context
 * @param in Input ByteBuf message data
 * @param out The output container after conversion
 * @throws Exception
 */
 @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in , List<Object> out ) throws Exception {
 out . add ( in .readInt()); //Read int type data, put it into the output, and complete the data type conversion
 }
 }

1.2 Re-appearance of the unpacking and sticking problem (the client sends ten pieces of data to the server)

1. Client startup class

 public class NettyClient {
 public   static void main(String[] args) throws Exception{
        EventLoopGroup worker = new NioEventLoopGroup();
 try {
 //Server startup class
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group (worker);
            bootstrap.channel(NioSocketChannel.class);
            bootstrap.handler(new ChannelInitializer<SocketChannel>() {
 @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                     ch.pipeline().addLast(new ClientHandler());
 }
 });
            ChannelFuture future = bootstrap.connect ( "127.0.0.1" , 5566).sync();
            future.channel().closeFuture().sync();
 finally
         worker.shutdownGracefully();
 }
 }
 }

2. ClientHandler

 public class ClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
 private int   count ;
 @Override
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
 System.out.println ( "Received message from server: " +
      msg.toString(CharsetUtil.UTF_8));
 System. out .println( "Number of messages received from the server: " + (++ count ));
 }
 @Override
 public void channelActive(ChannelHandlerContext ctx) throws Exception {
 for ( int i = 0; i < 10; i++) {
      ctx.writeAndFlush(Unpooled.copiedBuffer( "from client a message!" ,
 CharsetUtil.UTF_8));
 }
 }
 @Override
 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
      cause.printStackTrace();
 ctx.close ();
 }
 }

3. NettyServer on the server side

 public class NettyServer {
 public   static void main(String[] args) throws Exception {
 // Main thread, does not process any business logic, just receives client connection requests
      EventLoopGroup boss = new NioEventLoopGroup(1);
 //Working thread, the default number of threads is: cpu*2
      EventLoopGroup worker = new NioEventLoopGroup();
 try {
 //Server startup class
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        serverBootstrap.group (boss, worker) ;
 //Configure server channel
        serverBootstrap.channel(NioServerSocketChannel.class);
        serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
 @Override
          protected void initChannel(SocketChannel ch) throws Exception {
 ch.pipeline()
            .addLast(new ServerHandler());
 }
 }); //Worker thread processor
        ChannelFuture future = serverBootstrap.bind(5566).sync();
 System. out .println( "Server startup completed..." );
 //Wait for the server listening port to close
        future.channel().closeFuture().sync();
 finally
 //Elegant shutdown
        boss.shutdownGracefully();
        worker.shutdownGracefully();
 }
 }
 }

4. ServerHandler

 public class ServerHandler extends SimpleChannelInboundHandler<ByteBuf> {
 private int   count ;
 @Override
      protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
 System.out.println ( "The server received the message: " +
          msg.toString(CharsetUtil.UTF_8));
 System. out .println( "Number of messages received by the server: " + (++ count ));
          ctx.writeAndFlush(Unpooled.copiedBuffer( "ok" , CharsetUtil.UTF_8));
 }
 }

1.3 What is TCP packet sticking and unpacking?

TCP is a stream transmission, so-called stream means data without boundaries. When the server receives the client data, it does not know whether it is one or more. When the server reads the data, there will be a packet sticking problem.

Therefore, when the server and client transmit data, they must formulate unpacking rules. The client sticks the packet according to the rules, and the server unpacks the packet according to the rules. If there is any violation of the rules, the server will not be able to get the expected data.

1. Solutions (three types)

• 1. Add a header to the data packet being sent and store the size of the data in the header. The server can then read the data according to this size and know where the boundary is.
• 2. Send data with a fixed length. If the length exceeds the limit, send it in multiple times. If the length is insufficient, fill it with 0. The receiving end can receive the data with the fixed length.
• 3. Set boundaries between data packets, such as adding special symbols, so that the receiving end can separate different data packets through this boundary.

1.4 Practice: Solving TCP's Packet Sticking/Unpacking Problems

1. Custom protocol

 public class MyProtocol {
 private Integer length; //Data header: length
 private byte[] body; //data body
 public   Integer getLength() {
 return length;
 }
 public void setLength( Integer length) {
 this.length = length;
 }
 public byte[] getBody() {
 return body;
 }
 public void setBody(byte[] body) {
 this.body = body;
 }

2. Encoder

 public class MyEncoder extends MessageToByteEncoder<MyProtocol> {
 @Override
  protected void encode(ChannelHandlerContext ctx, MyProtocol msg, ByteBuf out ) throws Exception {
 out .writeInt(msg.getLength());
 out .writeBytes(msg.getBody());
 }
 }

3. Decoder

 public class MyDecoder extends ReplayingDecoder<Void> {
 @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in , List<Object> out ) throws Exception {
 int length = in .readInt(); //Get the length
 byte[] data = new byte[length]; //Define byte array based on length
 in .readBytes(data); //Read data
        MyProtocol myProtocol = new MyProtocol();
        myProtocol.setLength(length);
        myProtocol.setBody(data);
 out . add ( myProtocol );
 }
 }

4. ClientHandler

 public class ClientHandler extends SimpleChannelInboundHandler<MyProtocol> {
 private int   count ;
 @Override
    protected void channelRead0(ChannelHandlerContext ctx, MyProtocol msg) throws Exception {
 System.out.println ( "Received message from server: " + new String(msg.getBody(),
 CharsetUtil.UTF_8));
 System. out .println( "Number of messages received from the server: " + (++ count ));
 }
 @Override
 public void channelActive(ChannelHandlerContext ctx) throws Exception {
 for ( int i = 0; i < 10; i++) {
            byte[] data = "from client a message!" .getBytes(CharsetUtil.UTF_8);
            MyProtocol myProtocol = new MyProtocol();
            myProtocol.setLength(data.length);
            myProtocol.setBody(data);
            ctx.writeAndFlush(myProtocol);
 }
 }
 @Override
 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)throws Exception {
      cause.printStackTrace();
 ctx.close ();
 }
 }

5. NettyClient

 public class NettyClient {
 public   static void main(String[] args) throws Exception{
        EventLoopGroup worker = new NioEventLoopGroup();
 try {
 //Server startup class
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group (worker);
            bootstrap.channel(NioSocketChannel.class);
            bootstrap.handler(new ChannelInitializer<SocketChannel>() {
 @Override
              protected void initChannel(SocketChannel ch) throws Exception {
                ch.pipeline().addLast(new MyEncoder());
                ch.pipeline().addLast(new MyDecoder());
                ch.pipeline().addLast(new ClientHandler());
 }
 });
            ChannelFuture future = bootstrap.connect ( "127.0.0.1" , 5566).sync();
            future.channel().closeFuture().sync();
 finally
         worker.shutdownGracefully();
 }
 }
 }

6. ServerHandler

 public class ServerHandler extends SimpleChannelInboundHandler<MyProtocol> {
 private int   count ;
 @Override
    protected void channelRead0(ChannelHandlerContext ctx, MyProtocol msg) throws Exception {
 System.out.println ( "The server received the message: " + new String(msg.getBody(),
 CharsetUtil.UTF_8));
 System. out .println( "Number of messages received by the server: " + (++ count ));
        byte[] data = "ok" .getBytes(CharsetUtil.UTF_8);
        MyProtocol myProtocol = new MyProtocol();
        myProtocol.setLength(data.length);
        myProtocol.setBody(data);
        ctx.writeAndFlush(myProtocol);
 }
 }

7. NettyServer

 public class NettyServer {
 public   static void main(String[] args) throws Exception {
 // Main thread, does not process any business logic, just receives client connection requests
        EventLoopGroup boss = new NioEventLoopGroup(1);
 //Working thread, the default number of threads is: cpu*2
        EventLoopGroup worker = new NioEventLoopGroup();
 try {
 //Server startup class
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group (boss, worker) ;
 //Configure server channel
            serverBootstrap.channel(NioServerSocketChannel.class);
            serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
 @Override
              protected void initChannel(SocketChannel ch) throws Exception {
                  ch.pipeline()
                  .addLast(new MyDecoder())
                  .addLast(new MyEncoder())
                  .addLast(new ServerHandler());
 }
 }); //Worker thread processor
            ChannelFuture future = serverBootstrap.bind(5566).sync();
 System. out .println( "Server startup completed..." );
 //Wait for the server listening port to close
            future.channel().closeFuture().sync();
 finally
 //Elegant shutdown
            boss.shutdownGracefully();
            worker.shutdownGracefully();
 }
 }
 }

8. Testing

022 Netty core source code analysis

2.1 Analysis of the server startup process

1. Create a server channel

 1 The bind() method of the ServerBootstrap object is also the entry method
 2 initAndRegister() in AbstractBootstrap to create Channel
 The work of creating a Channel is completed by the newChannel() method in the ReflectiveChannelFactory reflection class.
 3 In the construction method of NioServerSocketChannel, ServerSocketChannel is opened through the SelectorProvider of the underlying jdk nio.
 4 In the constructor of AbstractNioChannel, set the channel to non-blocking: ch.configureBlocking( false );
 5 Through the construction method of AbstractChannel, the id, unsafe, and pipeline contents are created.
 6 Get some parameters of TCP underlying layer through NioServerSocketChannelConfig

2. Initialize the server channel

 1 initAndRegister() in AbstractBootstrap initializes channel, code: init(channel); 
 
 2 Set channelOptions and Attributes in the init() method in ServerBootstrap. 
 
 3 Next, save the user-defined parameters and attributes to the local variables currentChildOptions and currentChildAttrs.
 For later use 
 
 4 If serverBootstrap.handler() is set, it will be added to the pipeline. 
 
 5 Add the connector ServerBootstrapAcceptor. When a new connection is added, add the custom childHandler to the connection.
 In the pipeline:

 ch.eventLoop(). execute (new Runnable() {
 @Override
 public void run() {
 pipeline.addLast(
        new ServerBootstrapAcceptor(ch, currentChildGroup,currentChildHandler, currentChildOptions, currentChildAttrs));
 }
 });

 @Override
 @SuppressWarnings( "unchecked" )
 public void channelRead(ChannelHandlerContext ctx, Object msg) {
 //Only when the client is connected will it be executed
    final Channel child = (Channel) msg;
 //Add the custom childHandler to the connected pipeline
    child.pipeline().addLast(childHandler);
    setChannelOptions(child, childOptions, logger);
    setAttributes(child, childAttrs);
 try {
        childGroup.register(child).addListener(new ChannelFutureListener(){
 @Override
 public void operationComplete(ChannelFuture future) throws Exception {
                if (!future.isSuccess()) {
                 forceClose(child, future.cause());
 }
 }
 });
 } catch (Throwable t) {
 forceClose(child, t);
 }
 }

3. Register the selector

 //Register
 1 ChannelFuture in initAndRegister() method regFuture = config(). group ().register(channel); 
 
 2. The actual registration is done in io.netty.channel.AbstractChannel.AbstractUnsafe#register()
 2.1 AbstractChannel.this.eventLoop = eventLoop; Assign eventLoop and subsequent IO events
 The work will be executed by this eventLoop.
 2.2 Call doRegister() in register0(promise) to perform actual registration 
     
 3 io.netty.channel.nio.AbstractNioChannel#doRegister implements the method

 //Register the multiplexer through the JDK underlying layer
 //javaChannel() -- the channel created previously  
 //eventLoop().unwrappedSelector() -- Get the selector  
 //Registering events of interest is 0, indicating that there are no events of interest, and events will be re-registered later
 //Register this object as an attachment to the selector, so that you can get the content of the current object later
 selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);

4. Bind port

 1 Enter io.netty.bootstrap.AbstractBootstrap#doBind0() and start a thread to perform the port binding operation 
 
 2 Call io.netty.channel.AbstractChannelHandlerContext#bind(java.net.SocketAddress,
 io.netty.channel.ChannelPromise) method to start the thread execution again 
   
 3 Finally, call io.netty.channel.socket.nio.NioServerSocketChannel#doBind() method to perform binding operation

 //Bind through the underlying channel of JDK
 @SuppressJava6Requirement(reason = "Usage guarded by java version check" )
 @Override
    protected void doBind(SocketAddress localAddress) throws Exception {
      if (PlatformDependent.javaVersion() >= 7) {
         javaChannel().bind(localAddress, config.getBacklog());
 } else {
          javaChannel().socket().bind(localAddress,
          config.getBacklog());
 }
 }

When will the master-slave event of the selector be updated? This is ultimately done in the io.netty.channel.nio.AbstractNioChannel#doBeginRead() method.

 protected void doBeginRead() throws Exception {
    // Channel. read () or ChannelHandlerContext. read () was called
    final SelectionKey selectionKey = this.selectionKey;
    if (!selectionKey.isValid()) {
 return ;
 }
 readPending = true ;
    final int interestOps = selectionKey.interestOps();
    if ((interestOps & readInterestOp) == 0) {
        selectionKey.interestOps(interestOps | readInterestOp); //Set
 The event of interest is OP_ACCEPT
 }
 }
 //Assign values ​​in the NioServerSocketChannel constructor
 public NioServerSocketChannel(ServerSocketChannel channel) {
      super( null , channel, SelectionKey.OP_ACCEPT);
      config = new NioServerSocketChannelConfig(this,
      javaChannel().socket());
 }

2.2 Source code analysis of the connection request process

1. New connection access

 Entrance at
 io.netty.channel.nio.NioEventLoop#processSelectedKey(java.nio.channels.SelectionKey,
 io.netty.channel.nio.AbstractNioChannel)
 Enter the read () method of NioMessageUnsafe 
     
 Call io.netty.channel.socket.nio.NioServerSocketChannel#doReadMessages() to create
 The underlying channel of jdk is encapsulated as NioSocketChannel and added to the List container

 @Override
 protected int doReadMessages(List<Object> buf) throws Exception {
    SocketChannel ch = SocketUtils.accept(javaChannel());
 try {
 if (ch != null ) {
          buf.add (new NioSocketChannel(this, ch));
 return 1;
 }
 } catch (Throwable t) {
        logger.warn("Failed to   create a new channel from an
        accepted socket.", t);
 try {
 ch.close ();
        } catch (Throwable t2) {
            logger.warn( "Failed to close a socket." , t2);
 }
 }
 return 0;
 }

 Creating a NioSocketChannel Object
 new NioSocketChannel(this, ch), created by new method
 Calling super's constructor
 Pass in the SelectionKey.OP_READ event flag
 Create id, unsafe, pipeline objects
 Set non-blocking ch.configureBlocking( false );
 Creating a NioSocketChannelConfig Object

2. Register read events

 In io.netty.channel.nio.AbstractNioMessageChannel.NioMessageUnsafe:
 for ( int i = 0; i < size ; i ++) {
 readPending = false ;
 pipeline.fireChannelRead(readBuf.get(i)); //Propagate read events
 }

 In io.netty.channel.AbstractChannelHandlerContext#invokeChannelRead(java.lang.Object)
 French-Chinese
 private void invokeChannelRead(Object msg) {
 if (invokeHandler()) {
 try {
 //Execute channelRead. Note that the first execution is HeadHandler and the second is
          ServerBootstrapAcceptor
 //Enter through ServerBootstrapAcceptor and register the same selector as the new connection
 Logical registration and event binding
          ((ChannelInboundHandler) handler()).channelRead(this, msg);
 } catch (Throwable t) {
         invokeExceptionCaught(t);
 }
 } else {
 fireChannelRead(msg);
 }
 }

03 Three Netty optimization points

Some simple tips for using Netty. Worth a look.

3.1 Zero Copy

• 1 Bytebuf uses the pooled DirectBuffer type, so there is no need to copy the byte buffer twice. If heap memory is used, the JVM will copy it to the heap first and then write it to the Socket, which will result in an extra copy.
• 2 CompositeByteBuf encapsulates multiple ByteBufs into one ByteBuf, and no process copying is required when adding ByteBufs.
• 3 The transferTo method of Netty's file transfer class DefaultFileRegion sends files to the target channel without the need for circular copying, which improves performance.

3.2 EventLoop Task Scheduling

 channel.eventLoop(). execute (new Runnable() {
 @Override
 public void run() {
     channel.writeAndFlush(data)
 }
 });

Instead of using hannel.writeAndFlush(data); EventLoop's task scheduling is directly put into the execution queue of EventLoop corresponding to the channel, which will cause thread switching. Note: At the bottom layer of writeAndFlush, if it is not executed through eventLoop, a new thread will be started.

3.3 Reduce the call length of ChannelPipline

 public class YourHandler extends ChannelInboundHandlerAdapter {
 @Override
 public void channelActive(ChannelHandlerContext ctx) {
 //msg goes through the entire ChannelPipline and all handlers.
    ctx.channel().writeAndFlush(msg);
 //From the current handler to the end of the pipeline, the call is shorter.
 ctx.writeAndFlush(msg);
 }
 }

3.4 Reduce the creation of ChannelHandler (basically no configuration)

If the channelhandler is stateless (that is, no state parameters need to be saved), then use the Sharable annotation and only create one instance during bootstrap to reduce GC. Otherwise, a new handler object will be created for each connection.

 @ChannelHandler.Shareable
 public class StatelessHandler extends ChannelInboundHandlerAdapter {
 @Override
 public void channelActive(ChannelHandlerContext ctx) {}
 }
 public class MyInitializer extends ChannelInitializer<Channel> {
    private static final ChannelHandler INSTANCE = new StatelessHandler();
 @Override
 public void initChannel(Channel ch) {
     ch.pipeline().addLast(INSTANCE);
 }
 }

Notice:

Codecs such as ByteToMessageDecoder are stateful and cannot use the Sharable annotation.

3.5 Configuration parameter settings

 The bossGroup on the server only needs to be set to 1, because ServerSocketChannel will only
 Register to an eventLoop, and this eventLoop will only have one thread running, so there is no need to set
 Multithreading. For IO threads, in order to fully utilize the CPU and reduce the overhead of online and context switching, workGroup
 Set to twice the number of CPU cores, which is also the default value provided by Netty. 
 
 In scenarios where high response time is required, use .childOption(ChannelOption.TCP_NODELAY, true )
 And option (ChannelOption.TCP_NODELAY, true ) to disable the Nagle algorithm, do not wait, send immediately.

The knowledge points related to Netty have been shared. I will continue to share content related to Netty in the future, mainly about problems encountered at work and new insights.