#Netty 中的編解碼器的順序問題:
>> 出站處理器(ChannelOutboundHandler):
其write操作是從ChannelPipeline的尾部到頭部依次進行編碼
>> 入站處理器(ChannelInboundHandler):
其read操作是從ChannelPipeline的頭部到尾部依次進行解碼

#假設 Server 與 Client 處的 ChannelHandler 都是:
ch.pipeline()
    .addLast(new LengthFieldBasedFrameDecoder(4096, 0, 4, 0, 4)
    .addLast(new StringDecoder())
    .addLast(new LengthFieldPrepender(4))
    .addLast(new StringEncoder())
則流程為:
>> client 發(fā)消息: 先通過StringEncoder編碼, 再通過LengthFieldPrepender編碼
>> server 收消息: 先通過LengthFieldBasedFrameDecoder解碼, 再通過StringDecoder解碼
>> server 回消息: 先通過StringEncoder編碼, 再通過LengthFieldPrepender編碼
>> client 收消息: 先通過LengthFieldBasedFrameDecoder解碼, 再通過StringDecoder解碼

1.概述

1.1 粘包拆包問題描述

傳輸層除了有TCP協(xié)議外還有UDP協(xié)議左刽。

#UDP
不會發(fā)生粘包或拆包現(xiàn)象
因為UDP是基于報文發(fā)送的，從UDP的幀結(jié)構(gòu)可以看出框弛，
在UDP首部采用了16bit來指示UDP數(shù)據(jù)報文的長度，
因此在應用層能很好的將不同的數(shù)據(jù)報文區(qū)分開，從而避免粘包和拆包的問題酱讶。

#TCP
是基于字節(jié)流的何暮，在基于流的傳輸里（如TCP/IP），接收到的數(shù)據(jù)會先被存儲到一個socket接收緩沖里婆赠。
不幸的是绵脯，基于流的傳輸并不是一個數(shù)據(jù)包隊列，而是一個字節(jié)隊列休里。
TCP底層并不了解上層業(yè)務數(shù)據(jù)的具體含義蛆挫，它會根據(jù)TCP緩沖區(qū)的實際情況進行化包的劃分，
所以在業(yè)務上認為妙黍，一個完整的包可能會被TCP拆成多個包進行發(fā)送悴侵，
也有多個小的包封裝成一個大的數(shù)據(jù)包發(fā)送，這就是所謂的TCP粘包和拆包問題拭嫁。

粘包拆包問題.png

1.2 粘包拆包產(chǎn)生的原因

服務端和客戶端都會造成粘包畜挨、半包問題，以下列出常見原因噩凹。

#服務端：
>> 要發(fā)送的數(shù)據(jù)大于TCP發(fā)送緩沖區(qū)剩余空間大小巴元，將會發(fā)生拆包。
>> 要發(fā)送的數(shù)據(jù)大于MSS（最大報文長度）驮宴，TCP在傳輸前將進行拆包逮刨。
>> 要發(fā)送的數(shù)據(jù)小于TCP發(fā)送緩沖區(qū)的大小，TCP將多次寫入緩沖區(qū)的數(shù)據(jù)一次發(fā)送出去，將會發(fā)生粘包修己。

#接收端：
>> 接收數(shù)據(jù)端的應用層沒有及時讀取接收緩沖區(qū)中的數(shù)據(jù)恢总，將發(fā)生粘包。

1.3 粘包拆包問題的解決思路

TCP以流的方式進行數(shù)據(jù)傳輸睬愤，由于底層TCP無法理解上層的業(yè)務數(shù)據(jù)片仿，
所以在底層是無法保證數(shù)據(jù)包不被拆分和重組的，
這個問題只能通過上層的應用協(xié)議棧設計來解決尤辱，上層的應用協(xié)議為了對消息進行區(qū)分砂豌。

#業(yè)界主流的解決方案歸納如下：
>> 客戶端在發(fā)送數(shù)據(jù)包的時候，每個包都固定長度光督，比如1024個字節(jié)大小阳距，
如果客戶端發(fā)送的數(shù)據(jù)長度不足1024個字節(jié)，則通過補充空格的方式補全到指定長度结借；
>> 客戶端在每個包的末尾使用固定的分隔符筐摘，例如\r\n，
如果一個包被拆分了船老，則等待下一個包發(fā)送過來之后找到其中的\r\n咖熟，
然后對其拆分后的頭部部分與前一個包的剩余部分進行合并，這樣就得到了一個完整的包柳畔；
>> 將消息分為頭部和消息體球恤，在頭部中保存有當前整個消息的長度，
只有在讀取到足夠長度的消息之后才算是讀到了一個完整的消息荸镊；
>> 更復雜的應用層協(xié)議。
>> 自定義協(xié)議解決 (可參考 Dubbo)堪置。

2.Netty中粘包拆包問題的解決方案

#1.FixedLengthFrameDecoder
對于使用固定長度的粘包和拆包場景躬存，可以使用FixedLengthFrameDecoder，該解碼一器會每次讀取固定長度的消息舀锨，
如果當前讀取到的消息不足指定長度岭洲，那么就會等待下一個消息到達后進行補足。
其使用也比較簡單，只需要在構(gòu)造函數(shù)中指定每個消息的長度即可。
這里需要注意的是添瓷，F(xiàn)ixedLengthFrameDecoder只是一個解碼器菲宴，Netty也只提供了一個解碼器，
這是因為對于解碼是需要等待下一個包的進行補全的馁龟，代碼相對復雜，
對于編碼器，用戶可以自行編寫驻粟，因為編碼時只需要將不足指定長度的部分進行補全即可。

#2.LineBasedFrameDecoder與DelimiterBasedFrameDecoder
這倆適用于通過分隔符進行粘包和拆包問題的處理凶异。
>> LineBasedFrameDecoder的作用主要是通過換行符蜀撑，即\n或者\r\n對數(shù)據(jù)進行處理挤巡；
>> DelimiterBasedFrameDecoder的作用則是通過用戶指定的分隔符對數(shù)據(jù)進行粘包和拆包處理。
這兩個類都是解碼器類酷麦，而對于數(shù)據(jù)的編碼矿卑，
也即在每個數(shù)據(jù)包最后添加換行符或者指定分割符的部分需要用戶自行進行處理。

#3.LengthFieldBasedFrameDecoder與LengthFieldPrepender
二者需要配合起來使用沃饶，其實本質(zhì)上來講母廷，這兩者一個是解碼，一個是編碼的關(guān)系绍坝。
它們處理粘拆包的主要思想是在生成的數(shù)據(jù)包中添加一個長度字段徘意，用于記錄當前數(shù)據(jù)包的長度。
##3.1 LengthFieldBasedFrameDecoder(基于數(shù)據(jù)包長度的拆包器):
會按照參數(shù)指定的包長度偏移量數(shù)據(jù)對接收到的數(shù)據(jù)進行解碼轩褐，從而得到目標消息體數(shù)據(jù)椎咧；
>> maxFrameLength：指定了每個包所能傳遞的最大數(shù)據(jù)包大小把介；
>> lengthFieldOffset：指定了長度字段在字節(jié)碼中的偏移量勤讽；
>> lengthFieldLength：指定了長度字段所占用的字節(jié)長度；
>> lengthAdjustment：對一些不僅包含有消息頭和消息體的數(shù)據(jù)進行消息頭的長度的調(diào)整拗踢，
這樣就可以只得到消息體的數(shù)據(jù)脚牍，這里的lengthAdjustment指定的就是消息頭的長度；
>> initialBytesToStrip：對于長度字段在消息頭中間的情況巢墅，
可以通過initialBytesToStrip忽略掉消息頭以及長度字段占用的字節(jié)诸狭。
將應用層數(shù)據(jù)包的長度，作為接收端應用層數(shù)據(jù)包的拆分依據(jù)君纫。
按照應用層數(shù)據(jù)包的大小驯遇，拆包。
這個拆包器蓄髓，有一個要求叉庐，就是應用層協(xié)議中包含數(shù)據(jù)包的長度。
>> failFast 
true: 讀取到長度域超過maxFrameLength会喝，就拋出一個 TooLongFrameException陡叠。
false: 只有真正讀取完長度域的值表示的字節(jié)之后，才會拋出 TooLongFrameException肢执，
默認情況下設置為true枉阵，建議不要修改，否則可能會造成內(nèi)存溢出
>> ByteOrder: 數(shù)據(jù)存儲采用大端模式或小端模式


LengthFieldBasedFrameDecoder與LengthFieldPrepender需要配合起來使用预茄，
其實本質(zhì)上來講岭妖，這兩者一個是解碼，一個是編碼的關(guān)系。
它們處理粘拆包的主要思想是在生成的數(shù)據(jù)包中添加一個長度字段昵慌，用于記錄當前數(shù)據(jù)包的長度假夺。
LengthFieldBasedFrameDecoder會按照參數(shù)指定的包長度
偏移量數(shù)據(jù)對接收到的數(shù)據(jù)進行解碼，從而得到目標消息體數(shù)據(jù)斋攀；
LengthFieldPrepender則會在響應的數(shù)據(jù)前面添加指定的字節(jié)數(shù)據(jù)已卷，
這個字節(jié)數(shù)據(jù)中保存了當前消息體的整體字節(jié)數(shù)據(jù)長度。
數(shù)據(jù)在編碼發(fā)送的時候淳蔼，會指定當前這條消息的長度侧蘸。


2.LengthFieldPrepender則會在響應的數(shù)據(jù)前面添加指定的字節(jié)數(shù)據(jù)，
這個字節(jié)數(shù)據(jù)中保存了當前消息體的整體字節(jié)數(shù)據(jù)長度鹉梨。

#自定義粘包與拆包器
>> 方案1:
通過繼承LengthFieldBasedFrameDecoder和LengthFieldPrepender來實現(xiàn)粘包和拆包的處理讳癌。
>> 方案2:
通過繼承MessageToByteEncoder和ByteToMessageDecoder來實現(xiàn)。
這里MessageToByteEncoder的作用是將響應數(shù)據(jù)編碼為一個ByteBuf對象存皂，
而ByteToMessageDecoder則是將接收到的ByteBuf數(shù)據(jù)轉(zhuǎn)換為某個對象數(shù)據(jù)晌坤。
通過實現(xiàn)這兩個抽象類，用戶就可以達到實現(xiàn)自定義粘包和拆包處理的目的旦袋。

2.1 LineBasedFrameDecoder + StringDecoder (基于回車換行符解決粘包拆包問題)

package com.zy.netty.netty01;

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.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LineBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.concurrent.DefaultThreadFactory;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.util.concurrent.atomic.LongAdder;

@Slf4j
public class Server01 {
    public static void main(String[] args) {
        ServerBootstrap server = new ServerBootstrap();
        NioEventLoopGroup bossGroup = new NioEventLoopGroup(1, new DefaultThreadFactory("server-bossGroup", true));
        NioEventLoopGroup workerGroup = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors(), new DefaultThreadFactory("server-workerGroup", true));

        try {
            server.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .handler(new LoggingHandler(LogLevel.DEBUG))
                    .childHandler(new ChannelInitializer<NioSocketChannel>() {
                        @Override
                        protected void initChannel(NioSocketChannel ch) throws Exception {
                            ch.pipeline()
                                    // 這里結(jié)合 LineBasedFrameDecoder + StringDecoder 實現(xiàn)粘包拆包的處理, 其中 LineBasedFrameDecoder 是基于 \n 或 \r\n 實現(xiàn)
                                    // 這里需要注意: 單條消息不能超過給定的最大限度, 否則會拋出異常
                                    .addLast("lineBasedFrameDecoder", new LineBasedFrameDecoder(1024))
                                    .addLast("stringDecoder", new StringDecoder())
                                    .addLast("serverhandler01", new ServerHandler01());
                        }
                    });

            ChannelFuture channelFuture = server.bind("127.0.0.1", 8099).sync();
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            log.error("server is error ..........", e);
        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }

    private static class ServerHandler01 extends SimpleChannelInboundHandler<Object> {

        private LongAdder counter = new LongAdder();

        @Override
        protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
            counter.increment();

            String body = (String) msg;
            System.out.println("server01 received msg: " + body + "; the counter is: " + counter.doubleValue());

            String currentTime = "Query Time Order".equalsIgnoreCase(body) ? LocalDateTime.now().toString() : "Bad Order";
            // 發(fā)送消息時, 每條消息結(jié)尾需要添加回車換行符
            currentTime += System.getProperty("line.separator");
            ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes(StandardCharsets.UTF_8));
            ctx.writeAndFlush(resp);
        }
    }
}

package com.zy.netty.netty01;

import io.netty.bootstrap.Bootstrap;
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.nio.NioSocketChannel;
import io.netty.handler.codec.LineBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.concurrent.DefaultThreadFactory;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.atomic.LongAdder;
@Slf4j
public class Client01 {
    public static void main(String[] args) {
        Bootstrap client = new Bootstrap();
        NioEventLoopGroup executors = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors(), new DefaultThreadFactory("client-executor", true));

        try {
            client.group(executors)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .handler(new ChannelInitializer<NioSocketChannel>() {
                        @Override
                        protected void initChannel(NioSocketChannel ch) throws Exception {
                            ch.pipeline()
                                    .addLast("clentLogginHandler", new LoggingHandler(LogLevel.DEBUG))
                                    // 這里結(jié)合 LineBasedFrameDecoder + StringDecoder 實現(xiàn)粘包拆包的處理, 其中 LineBasedFrameDecoder 是基于 \n 或 \r\n 實現(xiàn)
                                    // 這里需要注意: 單條消息不能超過給定的最大限度, 否則會拋出異常
                                    .addLast("lineBasedFrameDecoder", new LineBasedFrameDecoder(1024))
                                    .addLast("stringDecoder", new StringDecoder())
                                    .addLast("clienthandler01", new ClientHandler01());
                        }
                    });

            ChannelFuture channelFuture = client.connect("127.0.0.1", 8099).sync();
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            log.error("client is error -----------", e);
        } finally {
            executors.shutdownGracefully();
        }
    }

    private static class ClientHandler01 extends SimpleChannelInboundHandler<Object> {

        private LongAdder counter = new LongAdder();

        @Override
        protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
            counter.increment();

            String body = (String) msg;
            System.out.println("client receive msg: " + body + "; the counter is: " + counter.doubleValue());
        }

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            String req = "Query Time Order";
            // 發(fā)送消息時, 每條消息結(jié)尾需要添加回車換行符
            req += System.getProperty("line.separator");
            byte[] reqBytes = req.getBytes(StandardCharsets.UTF_8);
            ByteBuf buf;
            for (int i = 0; i < 50; i++) {
                buf = Unpooled.buffer(reqBytes.length);
                buf.writeBytes(reqBytes);
                ctx.writeAndFlush(buf);
            }
        }
    }
}

2.2 DelimiterBasedFrameDecoder + StringDecoder (自定義分隔符解決粘包拆包問題)

這里模擬一個簡易聊天室

package com.zy.netty.netty02;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
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;
import io.netty.util.concurrent.DefaultThreadFactory;
import io.netty.util.concurrent.GlobalEventExecutor;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.StandardCharsets;
import java.util.Objects;

@Slf4j
public class Server02 {

    private static final String DELIMITER_SEPARATOR = "$_";

    public static void main(String[] args) {
        ServerBootstrap server = new ServerBootstrap();
        NioEventLoopGroup bossGroup = new NioEventLoopGroup(1, new DefaultThreadFactory("server-bossGroup", true));
        NioEventLoopGroup workerGroup = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors(), new DefaultThreadFactory("server-workerGroup", true));

        try {
            server.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .childHandler(new ChannelInitializer<NioSocketChannel>() {
                        @Override
                        protected void initChannel(NioSocketChannel ch) throws Exception {
                            ch.pipeline()
                                    // 這里結(jié)合 DelimiterBasedFrameDecoder + StringDecoder 實現(xiàn)粘包拆包的處理, 其中 DelimiterBasedFrameDecoder 需要自定義分隔符, 否則走默認值
                                    // 這里需要注意: 單條消息不能超過給定的最大限度, 否則會拋出異常
                                    .addLast("$_delimiterBasedFrameDecoder", new DelimiterBasedFrameDecoder(4096, Unpooled.copiedBuffer(DELIMITER_SEPARATOR.getBytes(StandardCharsets.UTF_8))))
                                    .addLast("stringDecoder", new StringDecoder())
                                    .addLast("stringEncoder", new StringEncoder())
                                    .addLast(new ServerHandler02());
                        }
                    });

            ChannelFuture channelFuture = server.bind("127.0.0.1", 8090).sync();
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            log.error("server is error ..........", e);
        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }

    @ChannelHandler.Sharable
    private static class ServerHandler02 extends SimpleChannelInboundHandler<Object> {
        // 保存所有與 服務端 建立好連接的 客戶端的 channel 對象
        // FIXME 分布式場景下, 是否可以存儲到 redis 中 ?
        private static ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

        @Override
        protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
            Channel channel = ctx.channel();
            channels.forEach(ch -> {
                if (Objects.equals(ch, channel)) {
                    ch.writeAndFlush("self send msg: " + msg + DELIMITER_SEPARATOR);
                } else {
                    ch.writeAndFlush("remoteAddress: " + channel.remoteAddress() + " send msg: " + msg + DELIMITER_SEPARATOR);
                }
            });
        }

        @Override
        public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
            // 有客戶端加入的事件
            Channel channel = ctx.channel();
            // 廣播:
            channels.add(channel);
            channels.writeAndFlush("remoteAddress: " + channel.remoteAddress() + " join the server" + DELIMITER_SEPARATOR);
        }

        @Override
        public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
            // 有客戶端離開的事件
            Channel channel = ctx.channel();
            // 廣播:
            channels.writeAndFlush("remoteAddress: " + channel.remoteAddress() + " leave the server" + DELIMITER_SEPARATOR);
            channels.remove(channel);
        }

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            // 客戶端處于活動狀態(tài)
            Channel channel = ctx.channel();
            System.out.println("remoteAddress: " + channel.remoteAddress() + " is online.");
        }

        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            // 客戶端處于非活動狀態(tài)
            Channel channel = ctx.channel();
            System.out.println("remoteAddress: " + channel.remoteAddress() + " is offline.");
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    }
}

package com.zy.netty.netty02;

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.nio.NioSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.concurrent.DefaultThreadFactory;
import lombok.extern.slf4j.Slf4j;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.atomic.LongAdder;

/**
 * 這里多啟動幾個 client, 即可測試
 */
@Slf4j
public class Client02 {
    private static final String DELIMITER_SEPARATOR = "$_";
    public static void main(String[] args) {
        Bootstrap client = new Bootstrap();
        NioEventLoopGroup executors = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors(), new DefaultThreadFactory("client-executor", true));

        try {
            client.group(executors)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .handler(new ChannelInitializer<NioSocketChannel>() {
                        @Override
                        protected void initChannel(NioSocketChannel ch) throws Exception {
                            ch.pipeline()
                                    // 這里結(jié)合 DelimiterBasedFrameDecoder + StringDecoder 實現(xiàn)粘包拆包的處理, 其中 DelimiterBasedFrameDecoder 需要自定義分隔符, 否則走默認值
                                    // 這里需要注意: 單條消息不能超過給定的最大限度, 否則會拋出異常
                                    .addLast("$_delimiterBasedFrameDecoder", new DelimiterBasedFrameDecoder(4096, Unpooled.copiedBuffer(DELIMITER_SEPARATOR.getBytes(StandardCharsets.UTF_8))))
                                    .addLast("stringDecoder", new StringDecoder())
                                    .addLast("stringEncoder", new StringEncoder())
                                    .addLast("clientHandler02", new ClientHandler02());
                        }
                    });

            ChannelFuture channelFuture = client.connect("127.0.0.1", 8090).sync();
            // 死循環(huán), 監(jiān)聽鍵盤輸入, 客戶端啟動后(可以多啟動幾個客戶端), 輸入任意消息, 回車即可
            BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
            for (;;) {
                channelFuture.channel().writeAndFlush(reader.readLine() + DELIMITER_SEPARATOR);
            }
            // channelFuture.channel().closeFuture().sync();
        } catch (Exception e) {
            log.error("client is error -----------", e);
        } finally {
            executors.shutdownGracefully();
        }
    }

    @ChannelHandler.Sharable
    private static class ClientHandler02 extends SimpleChannelInboundHandler<Object> {
        private LongAdder counter = new LongAdder();

        @Override
        protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
            counter.increment();

            String body = (String) msg;
            System.out.println("client receive msg: " + body + "; the counter is: " + counter.doubleValue());
        }
    }
}