深入理解Netty编解码、粘包拆包、心跳机制

connygpt 2024-09-29 11:03 9 浏览

前言

Netty系列文章：

前面我们讲了 BIO、NIO、AIO 等一些基础知识和Netty核心功能与线程模型，本篇重点来理解Netty的编解码、粘包拆包、心跳机制等实现原理进行讲解。

Netty编解码

Netty 涉及到编解码的组件有 Channel 、 ChannelHandler 、 ChannelPipe 等，我们先大概了解下这几个组件的作用。

ChannelHandler

ChannelHandler 充当来处理入站和出站数据的应用程序逻辑容器。例如，实现 ChannelInboundHandler 接口（或 ChannelInboundHandlerAdapter），你就可以接收入站事件和数据，这些数据随后会被你的应用程序的业务逻辑处理。当你要给连接的客户端发送响应时，也可以从 ChannelInboundHandler 刷数据。你的业务逻辑通常下在一个或者多个 ChannelInboundHandler 中。

ChannelOutboundHandler 原理一样，只不过它是用来处理出站数据的。

ChannelPipeline

ChannelPipeline 提供了 ChannelHandler 链的容器。以客户端应用程序为例，如果有事件的运动方向是从客户端到服务端，那么我们称这些事件为出站的，即客户端发送给服务端的数据会通过 pipeline 中的一系列 ChannelOutboundHandler (ChannelOutboundHandler 调用是从 tail 到 head 方向逐个调用每个 handler 的逻辑)，并被这些 Hadnler 处理，反之称为入站的，入站只调用 pipeline 里的 ChannelInboundHandler 逻辑（ChannelInboundHandler 调用是从 head 到 tail 方向逐个调用每个 handler 的逻辑。）

编解码器

当你通过Netty发送或者接受一个消息的时候，就将会发生一次数据转换。入站消息会被解码：从字节转换为另一种格式(比如java对象);如果是出站消息，它会被编码成字节。

Netty提供了一系列实用的编码解码器，它们都实现了ChannelInboundHadnler或者ChannelOutboundHandler接口。在这些类中， channelRead方法已经被重写了。

以入站为例，对于每个从入站Channel读取的消息，这个方法会被调用。随后，它将调用由已知解码器所提供的decode()方法进行解码，并将已经解码的字节转发给ChannelPipeline中的下一个ChannelInboundHandler。

Netty提供了很多编解码器，比如编解码字符串的StringEncoder和StringDecoder，编解码对象的ObjectEncoder和ObjectDecoder 等。

当然也可以通过集成ByteToMessageDecoder自定义编解码器。

示例代码

完整代码在 Github ：
https://github.com/Niuh-Study/niuh-netty.git
对应的包 com.niuh.netty.codec

Netty粘包拆包

TCP 粘包拆包是指发送方发送的若干包数据到接收方接收时粘成一包或某个数据包被拆开接收。如下图所示，client 发送了两个数据包 D1 和 D2，但是 server 端可能会收到如下几种情况的数据。

程序演示

首先准备客户端负责发送消息，连续发送5次消息，代码如下：

public void channelActive(ChannelHandlerContext ctx) throws Exception {
 for (int i = 1; i <= 5; i++) {
     ByteBuf byteBuf = Unpooled.copiedBuffer("msg No" + i + " ", Charset.forName("utf-8"));
        ctx.writeAndFlush(byteBuf);
    }
}

然后服务端作为接收方，接收并且打印结果：

// count 变量，用于计数
private int count;

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
 System.out.println("服务器读取线程 " + Thread.currentThread().getName());

    ByteBuf buf = (ByteBuf) msg;
    byte[] bytes = new byte[buf.readableBytes()];
    // 把ByteBuf的数据读到bytes数组中
    buf.readBytes(bytes);
    String message = new String(bytes, Charset.forName("utf-8"));
    System.out.println("服务器接收到数据：" + message);
    // 打印接收的次数
    System.out.println("接收到的数据量是：" + (++this.count));
}

启动服务端，再启动两个客户端发送消息,服务端的控制台可以看到这样：

粘包的问题其实是随机的，所以每次结果都不太一样。

完整代码在 Github ：
https://github.com/Niuh-Study/niuh-netty.git
对应的包 com.niuh.splitpacket0

为什么出现粘包现象？

TCP 是面向连接的，面向流的，提供高可靠性服务。收发两端（客户端和服务器端）都要有成对的 socket，因此，发送端为了将多个发送给接收端的包，更有效的发送给对方，使用了优化方法（Nagle算法），将多次间隔较少且数据量小的数据，合并成一个大的数据块，然后进行封包，这样做虽然提供了效率，但是接收端就难以分辨出完整的数据包了，因为面向流的通信是无消息保护边界的。

如何理解TCP是面向字节流的

应用程序和 TCP 的交互是一次一个数据块（大小不等），但 TCP 把应用程序交下来的数据仅仅看成是一连串的无结构的字节流。TCP 并不知道所传送的字节流的含义；
因此 TCP 不保证接收方应用程序所收到的数据块和发送方应用程序所发出的数据块具有对应大小的关系（例如，发送方应用程序交给发送方的 TCP 共 10 个数据块，但接收方的 TCP 可能只用了 4 个就把收到的字节流交付上层的应用程序）；
同时，TCP 不关心应用进程一次把多长的报文发送到 TCP 的缓存中，而是根据对方给出的窗口值和当前网络阻塞的程度来决定一个报文段应包含多少个字节（UDP 发送的报文长度是应用进程给出的）。如果应用进程传送到 TCP 缓存的数据块太长，TCP 就可以把它划分短一点再传送。如果应用程序一次只发来一个字节，TCP 也可以等待积累有足够多的字节后再构成报文段发送出去。

TCP发送报文一般是 3 个时机

缓冲区数据达到，最大报文长度 MSS；
由发送端的应用进程指明要求发送报文段，即 TCP 支持的推送（push）操作；
当发送方的一个计时器期限到了，即使长度不超过 MSS，也发送。

解决方案

一般解决粘包拆包问题有 4 中办法

在数据的末尾添加特殊的符号标识数据包的边界。通常会加\n、\r、\t或者其他的符号

学习 HTTP、FTP 等，使用回车换行符号；

在数据的头部声明数据的长度，按长度获取数据

将消息分为 head 和 body，head 中包含 body 长度的字段，一般 head 的第一个字段使用 int 值来表示 body 长度；

规定报文的长度，不足则补空位。读取时按规定好的长度来读取。比如 100 字节，如果不够就补空格；
使用更复杂的应用层协议。

使用LineBasedFrameDecoder

LineBasedFrameDecoder 是Netty内置的一个解码器，对应的编码器是 LineEncoder。

原理是上面讲的第一种思路，在数据末尾加上特殊符号以标识边界。默认是使用换行符\n。

用法很简单，发送方加上编码器：

@Override
protected void initChannel(SocketChannel ch) throws Exception {
 //添加编码器，使用默认的符号\n，字符集是UTF-8
    ch.pipeline().addLast(new LineEncoder(LineSeparator.DEFAULT, CharsetUtil.UTF_8));
    ch.pipeline().addLast(new TcpClientHandler());
}

接收方加上解码器：

@Override
protected void initChannel(SocketChannel ch) throws Exception {
 //解码器需要设置数据的最大长度，我这里设置成1024
 ch.pipeline().addLast(new LineBasedFrameDecoder(1024));
 //给pipeline管道设置业务处理器
 ch.pipeline().addLast(new TcpServerHandler());
}

然后在发送方，发送消息时在末尾加上标识符：

@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
    for (int i = 1; i <= 5; i++) {
  //在末尾加上默认的标识符\n
     ByteBuf byteBuf = Unpooled.copiedBuffer("msg No" + i + StringUtil.LINE_FEED, Charset.forName("utf-8"));
        ctx.writeAndFlush(byteBuf);
 }
}

于是我们再次启动服务端和客户端，在服务端的控制台可以看到：

在数据的末尾添加特殊的符号标识数据包的边界，粘包、拆包的问题就得到解决了。

注意：数据末尾一定是分隔符，分隔符后面不要再加上数据，否则会当做下一条数据的开始部分。下面是错误演示：

@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
    for (int i = 1; i <= 5; i++) {
  //在末尾加上默认的标识符\n
     ByteBuf byteBuf = Unpooled.copiedBuffer("msg No" + i + StringUtil.LINE_FEED + "[我是分隔符后面的字符串]", Charset.forName("utf-8"));
        ctx.writeAndFlush(byteBuf);
 }
}

服务端的控制台就会看到这样的打印信息：

使用自定义长度帧解码器

使用这个解码器解决粘包问题的原理是上面讲的第二种，在数据的头部声明数据的长度，按长度获取数据。这个解码器构造器需要定义5个参数，相对较为复杂一点，先看参数的解释：

maxFrameLength 发送数据包的最大长度
lengthFieldOffset 长度域的偏移量。长度域位于整个数据包字节数组中的开始下标。
lengthFieldLength 长度域的字节数长度。长度域的字节数长度。
lengthAdjustment 长度域的偏移量矫正。如果长度域的值，除了包含有效数据域的长度外，还包含了其他域（如长度域自身）长度，那么，就需要进行矫正。矫正的值为：包长 - 长度域的值 – 长度域偏移 – 长度域长。
initialBytesToStrip 丢弃的起始字节数。丢弃处于此索引值前面的字节。

前面三个参数比较简单，可以用下面这张图进行演示：

矫正偏移量是什么意思呢？

是假设你的长度域设置的值除了包括有效数据的长度还有其他域的长度包含在里面，那么就要设置这个值进行矫正，否则解码器拿不到有效数据。

丢弃的起始字节数。这个比较简单，就是在这个索引值前面的数据都丢弃，只要后面的数据。一般都是丢弃长度域的数据。当然如果你希望得到全部数据，那就设置为0。

下面就在消息接收端使用自定义长度帧解码器，解决粘包的问题：

@Override
protected void initChannel(SocketChannel ch) throws Exception {
 //数据包最大长度是1024
    //长度域的起始索引是0
    //长度域的数据长度是4
    //矫正值为0，因为长度域只有 有效数据的长度的值
    //丢弃数据起始值是4，因为长度域长度为4，我要把长度域丢弃，才能得到有效数据
    ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(1024, 0, 4, 0, 4));
    ch.pipeline().addLast(new TcpClientHandler());
}

接着编写发送端代码，根据解码器的设置，进行发送：

@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
 for (int i = 1; i <= 5; i++) {
     String str = "msg No" + i;
        ByteBuf byteBuf = Unpooled.buffer(1024);
        byte[] bytes = str.getBytes(Charset.forName("utf-8"));
        //设置长度域的值，为有效数据的长度
        byteBuf.writeInt(bytes.length);
        //设置有效数据
        byteBuf.writeBytes(bytes);
        ctx.writeAndFlush(byteBuf);
    }
}

然后启动服务端，客户端，我们可以看到控制台打印结果：

可以看到，利用自定义长度帧解码器解决了粘包问题。

使用Google Protobuf编解码器

Netty官网上是明显写着支持Google Protobuf的，如下图所示：

Google Protobuf是什么

官网的原话： Protocol buffers are Google's language-neutral, platform-neutral, extensible mechanism for serializing structured data – think XML, but smaller, faster, and simpler. You define how you want your data to be structured once, then you can use special generated source code to easily write and read your structured data to and from a variety of data streams and using a variety of languages.

翻译一下：Protocol buffers是Google公司的与语言无关、平台无关、可扩展的序列化数据的机制，类似XML，但是更小、更快、更简单。您只需定义一次数据的结构化方式，然后就可以使用特殊生成的源代码，轻松地将结构化数据写入和读取到各种数据流中，并支持多种语言。

在rpc或tcp通信等很多场景都可以使用。通俗来讲，如果客户端和服务端使用的是不同的语言，那么在服务端定义一个数据结构，通过protobuf转化为字节流，再传送到客户端解码，就可以得到对应的数据结构。这就是protobuf神奇的地方。并且，它的通信效率极高，“一条消息数据，用protobuf序列化后的大小是json的10分之一，xml格式的20分之一，是二进制序列化的10分之一”。

Google Protobuf 官网：
https://developers.google.cn/protocol-buffers/

为什么使用Google Protobuf

在一些场景下，数据需要在不同的平台，不同的程序中进行传输和使用，例如某个消息是用C++程序产生的，而另一个程序是用java写的，当前者产生一个消息数据时，需要在不同的语言编写的不同的程序中进行操作，如何将消息发送并在各个程序中使用呢？这就需要设计一种消息格式，常用的就有json和xml，protobuf出现的则较晚。

Google Protobuf优点

protobuf 的主要优点是简单，快；
protobuf将数据序列化为二进制之后，占用的空间相当小，基本仅保留了数据部分，而xml和json会附带消息结构在数据中；
protobuf使用起来很方便，只需要反序列化就可以了，而不需要xml和json那样层层解析。

Google Protobuf安装

因为我这里是Mac系统，Mac下面除了用dmg、pkg来安装软件外，比较方便的还有用brew命令进行安装，它能帮助安装其他所需要的依赖，从而减少不必要的麻烦。

安装最新版本的protoc

从github上下载 protobuf3

https://github.com/protocolbuffers/protobuf/releases/tag/v3.13.0

Mac系统选择第一个，如下图所示：

下载成功后，切换到root用户

sudo -i

tar xyf protobuf-all-3.13.0.tar.gz
cd protobuf-3.13.0

./configure --prefix=/usr/local/protobuf

安装

make
make install

配置环境变量

第一步：找到.bash_profile文件并编辑

cd ~
open .bash_profile

第二步：然后在打开的bash_profile文件末尾添加如下配置：

export PROTOBUF=/usr/local/protobuf 
export PATH=$PROTOBUF/bin:$PATH

第三步：source一下使文件生效

source .bash_profile

测试安装结果

protoc --version

使用Google Protobuf

以下步骤参考Google Protobuf的github项目的指南。

https://github.com/protocolbuffers/protobuf/tree/master/java

第一步：添加maven依赖

<dependency>
  <groupId>com.google.protobuf</groupId>
  <artifactId>protobuf-java</artifactId>
  <version>3.11.0</version>
</dependency>

第二步：编写proto文件Message.proto

如何编写.proto文件的相关文档说明，可以去官网查看下面写一个例子，请看示范：

syntax = "proto3"; //版本
option java_outer_classname = "MessagePojo";//生成的外部类名，同时也是文件名

message Message {
    int32 id = 1;//Message类的一个属性，属性名称是id，序号为1
    string content = 2;//Message类的一个属性，属性名称是content，序号为2
}

第三步：使用编译器，通过.proto文件生成代码

在执行上面的安装步骤后，进入到 bin 目录下，可以看到一个可执行文件 protoc

cd /usr/local/protobuf/bin/

然后复制前面写好的Message.proto文件到此目录下，如图所示：

输入命令：

protoc --java_out=. Message.proto

然后就可以看到生成的MessagePojo.java文件。最后把文件复制到IDEA项目中。

第四步：在发送端添加编码器，在接收端添加解码器

客户端添加编码器，对消息进行编码。

@Override
protected void initChannel(SocketChannel ch) throws Exception {
 //在发送端添加Protobuf编码器
    ch.pipeline().addLast(new ProtobufEncoder());
 ch.pipeline().addLast(new TcpClientHandler());
}

服务端添加解码器，对消息进行解码。

@Override
protected void initChannel(SocketChannel ch) throws Exception {
 //添加Protobuf解码器，构造器需要指定解码具体的对象实例
 ch.pipeline().addLast(new ProtobufDecoder(MessagePojo.Message.getDefaultInstance()));
 //给pipeline管道设置处理器
 ch.pipeline().addLast(new TcpServerHandler());
}

第五步：发送消息

客户端发送消息，代码如下：

@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
 //使用的是构建者模式进行创建对象
 MessagePojo.Message message = MessagePojo
      .Message
            .newBuilder()
            .setId(1)
            .setContent("一角钱，起飞~")
            .build();
    ctx.writeAndFlush(message);
}

服务端接收到数据，并且打印：

@Override
protected void channelRead0(ChannelHandlerContext ctx, MessagePojo.Message messagePojo) throws Exception {
    System.out.println("id:" + messagePojo.getId());
    System.out.println("content:" + messagePojo.getContent());
}

测试结果正确：

分析Protocol的粘包、拆包

实际上直接使用Protocol编解码器还是存在粘包问题的。

证明一下，发送端循环一百次发送100条"一角钱，起飞"的消息，请看发送端代码演示：

@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
 for (int i = 1; i <= 100; i++) {
  MessagePojo.Message message = MessagePojo
      .Message
            .newBuilder()
            .setId(i)
            .setContent(i + "号一角钱，起飞~")
            .build();
      ctx.writeAndFlush(message);
 }
}

这时，启动服务端，客户端后，可能只有打印几条消息或者在控制台看到如下错误：

com.google.protobuf.InvalidProtocolBufferException: While parsing a protocol message, the input ended unexpectedly in the middle of a field. This could mean either that the input has been truncated or that an embedded message misreported its own length.

意思是：分析protocol消息时，输入意外地在字段中间结束。这可能意味着输入被截断，或者嵌入的消息误报了自己的长度。

其实就是粘包问题，多条数据合并成一条数据了，导致解析出现异常。

解决Protocol的粘包、拆包问题

只需要在发送端加上编码器 ProtobufVarint32LengthFieldPrepender

@Override
protected void initChannel(SocketChannel ch) throws Exception {
 ch.pipeline().addLast(new ProtobufVarint32LengthFieldPrepender());
    ch.pipeline().addLast(new ProtobufEncoder());
    ch.pipeline().addLast(new TcpClientHandler());
}

接收方加上解码器 ProtobufVarint32FrameDecoder

@Override
protected void initChannel(SocketChannel ch) throws Exception {
 ch.pipeline().addLast(new ProtobufVarint32FrameDecoder());
 ch.pipeline().addLast(new ProtobufDecoder(MessagePojo.Message.getDefaultInstance()));
 //给pipeline管道设置处理器
 ch.pipeline().addLast(new TcpServerHandler());
}

然后再启动服务端和客户端，我们可以看到正常了～

ProtobufVarint32LengthFieldPrepender 编码器的工作如下：

 * BEFORE ENCODE (300 bytes)       AFTER ENCODE (302 bytes)
 * +---------------+               +--------+---------------+
 * | Protobuf Data |-------------->| Length | Protobuf Data |
 * |  (300 bytes)  |               | 0xAC02 |  (300 bytes)  |
 * +---------------+               +--------+---------------+
@Sharable
public class ProtobufVarint32LengthFieldPrepender extends MessageToByteEncoder<ByteBuf> {
    @Override
    protected void encode(ChannelHandlerContext ctx, ByteBuf msg, ByteBuf out) throws Exception {
        int bodyLen = msg.readableBytes();
        int headerLen = computeRawVarint32Size(bodyLen);
        //写入请求头，消息长度
        out.ensureWritable(headerLen + bodyLen);
        writeRawVarint32(out, bodyLen);
        //写入数据
        out.writeBytes(msg, msg.readerIndex(), bodyLen);
    }
}

ProtobufVarint32FrameDecoder 解码器的工作如下：

 * BEFORE DECODE (302 bytes)       AFTER DECODE (300 bytes)
 * +--------+---------------+      +---------------+
 * | Length | Protobuf Data |----->| Protobuf Data |
 * | 0xAC02 |  (300 bytes)  |      |  (300 bytes)  |
 * +--------+---------------+      +---------------+
public class ProtobufVarint32FrameDecoder extends ByteToMessageDecoder {
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        //标记读取的下标位置
        in.markReaderIndex();
        //获取读取的下标位置
        int preIndex = in.readerIndex();
        //解码，获取消息的长度,并且移动读取的下标位置
        int length = readRawVarint32(in);
        //比较解码前和解码后的下标位置，如果相等。表示字节数不够读取，跳到下一轮
        if (preIndex == in.readerIndex()) {
            return;
        }
        //如果消息的长度小于0，抛出异常
        if (length < 0) {
            throw new CorruptedFrameException("negative length: " + length);
        }
        //如果不够读取一个完整的数据，reset还原下标位置。
        if (in.readableBytes() < length) {
            in.resetReaderIndex();
        } else {
            //否则，把数据写入到out，接收端就拿到了完整的数据了
            out.add(in.readRetainedSlice(length));
        }
 }

总结：

发送端通过编码器在发送的时候在消息体前面加上一个描述数据长度的数据块。
接收方通过解码器先获取描述数据长度的数据块，知道完整数据的长度，然后根据数据长度获取一条完整的数据。

Netty心跳检测机制

何为心跳

所谓心跳, 即在 TCP 长连接中, 客户端和服务器之间定期发送的一种特殊的数据包, 通知对方自己还在线, 以确保 TCP 连接的有效性.

注：心跳包还有另一个作用，经常被忽略，即：一个连接如果长时间不用，防火墙或者路由器就会断开该连接。

在 Netty 中, 实现心跳机制的关键是 IdleStateHandler, 看下它的构造器:

public IdleStateHandler(int readerIdleTimeSeconds, int writerIdleTimeSeconds, int allIdleTimeSeconds) {
 this((long)readerIdleTimeSeconds, (long)writerIdleTimeSeconds, (long)allIdleTimeSeconds, TimeUnit.SECONDS);
}

三个参数的含义如下：

readerIdleTimeSeconds: 读超时。即当在指定的时间间隔内没有从 Channel 读取到数据时, 会触发一个 READER_IDLE 的 IdleStateEvent 事件。
writerIdleTimeSeconds: 写超时。即当在指定的时间间隔内没有数据写入到 Channel 时, 会触发一个 WRITER_IDLE 的 IdleStateEvent 事件。
allIdleTimeSeconds: 读/写超时。即当在指定的时间间隔内没有读或写操作时, 会触发一个 ALL_IDLE 的 IdleStateEvent 事件。

注：这三个参数默认的时间单位是秒。若需要指定其他时间单位，可以使用另一个构造方法:

public IdleStateHandler(long readerIdleTime, long writerIdleTime, long allIdleTime, TimeUnit unit) {
 this(false, readerIdleTime, writerIdleTime, allIdleTime, unit);
}

要实现Netty服务端心跳检测机制需要在服务器端的ChannelInitializer中加入如下的代码:

pipeline.addLast(new IdleStateHandler(3, 0, 0, TimeUnit.SECONDS));

Netty心跳源码分析

初步地看下IdleStateHandler源码，先看下IdleStateHandler中的channelRead方法:

红框代码其实表示该方法只是进行了透传，不做任何业务逻辑处理，让channelPipe中的下一个handler处理channelRead方法；

我们再看看channelActive方法:

这里有个initialize的方法，这是IdleStateHandler的精髓，接着探究:

这边会触发一个Task，ReaderIdleTimeoutTask，这个task里的run方法源码是这样的:

第一个红框代码是用当前时间减去最后一次channelRead方法调用的时间，假如这个结果是6s，说明最后一次调用channelRead已经是6s 之前的事情了，你设置的是5s，那么nextDelay则为-1，说明超时了，那么第二个红框代码则会触发下一个handler的 userEventTriggered方法:

如果没有超时则不触发userEventTriggered方法。

Netty心跳检测代码示例

服务端

package com.niuh.netty.heartbeat;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.timeout.IdleStateHandler;

import java.util.concurrent.TimeUnit;

public class HeartBeatServer {

    public static void main(String[] args) throws Exception {
        EventLoopGroup boss = new NioEventLoopGroup();
        EventLoopGroup worker = new NioEventLoopGroup();
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(boss, worker)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast("decoder", new StringDecoder());
                            pipeline.addLast("encoder", new StringEncoder());
                            //IdleStateHandler的readerIdleTime参数指定超过3秒还没收到客户端的连接，
                            //会触发IdleStateEvent事件并且交给下一个handler处理，下一个handler必须
                            //实现userEventTriggered方法处理对应事件
                            pipeline.addLast(new IdleStateHandler(3, 0, 0, TimeUnit.SECONDS));
                            pipeline.addLast(new HeartBeatServerHandler());
                        }
                    });
            System.out.println("netty server start。。");
            ChannelFuture future = bootstrap.bind(9000).sync();
            future.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            worker.shutdownGracefully();
            boss.shutdownGracefully();
        }
    }
}

服务端回调处理类

package com.niuh.netty.heartbeat;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleStateEvent;

public class HeartBeatServerHandler extends SimpleChannelInboundHandler<String> {

    int readIdleTimes = 0;

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String s) throws Exception {
        System.out.println(" ====== > [server] message received : " + s);
        if ("Heartbeat Packet".equals(s)) {
            ctx.channel().writeAndFlush("ok");
        } else {
            System.out.println(" 其他信息处理 ... ");
        }
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        IdleStateEvent event = (IdleStateEvent) evt;

        String eventType = null;
        switch (event.state()) {
            case READER_IDLE:
                eventType = "读空闲";
                readIdleTimes++; // 读空闲的计数加1
                break;
            case WRITER_IDLE:
                eventType = "写空闲";
                // 不处理
                break;
            case ALL_IDLE:
                eventType = "读写空闲";
                // 不处理
                break;
        }



        System.out.println(ctx.channel().remoteAddress() + "超时事件：" + eventType);
        if (readIdleTimes > 3) {
            System.out.println(" [server]读空闲超过3次，关闭连接，释放更多资源");
            ctx.channel().writeAndFlush("idle close");
            ctx.channel().close();
        }
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.err.println("=== " + ctx.channel().remoteAddress() + " is active ===");
    }
}

客户端

package com.niuh.netty.heartbeat;

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.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

import java.util.Random;

public class HeartBeatClient {
    public static void main(String[] args) throws Exception {
        EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast("decoder", new StringDecoder());
                            pipeline.addLast("encoder", new StringEncoder());
                            pipeline.addLast(new HeartBeatClientHandler());
                        }
                    });

            System.out.println("netty client start。。");
            Channel channel = bootstrap.connect("127.0.0.1", 9000).sync().channel();
            String text = "Heartbeat Packet";
            Random random = new Random();
            while (channel.isActive()) {
                int num = random.nextInt(10);
                Thread.sleep(2 * 1000);
                channel.writeAndFlush(text);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            eventLoopGroup.shutdownGracefully();
        }
    }

    static class HeartBeatClientHandler extends SimpleChannelInboundHandler<String> {

        @Override
        protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
            System.out.println(" client received :" + msg);
            if (msg != null && msg.equals("idle close")) {
                System.out.println(" 服务端关闭连接，客户端也关闭");
                ctx.channel().closeFuture();
            }
        }
    }
}

PS：以上代码提交在 Github ：
https://github.com/Niuh-Study/niuh-netty.git

文章持续更新，可以公众号搜一搜「一角钱技术」第一时间阅读，
本文 GitHub org_hejianhui/JavaStudy 已经收录，欢迎 Star。

protoc安装

上一篇：gRPC Java入门示例 grpc教程 java
下一篇：Golang使用grpc详解 golang gin rbac