Netty网络编程第七卷

转x33g5p2x  于2022-02-07 转载在 其他  
字(34.7k)|赞(0)|评价(0)|浏览(518)

本卷重点在netty的知识盲区整理，列举了一些常用的技巧，包括心跳检测，源码阅读，退出机制

Netty的高性能之道

首先给大家讲讲netty那些高性能的体现之处,前六卷也讲过，不过这次是全部整合在一起讲，方便阅读

传统RPC调用性能差的三宗罪

  • 1网络传输方式问题

传统的RPC框架或者基于RMI等方式的远程服务（过程）调用采用了同步阻塞IO，当客户端的并发压力或者网络时延增大之后，同步阻塞IO会由于频繁的wait导致IO线程经常性的阻塞，由于线程无法高效的工作，IO处理能力自然下降。

下面，我们通过BIO通信模型图看下BIO通信的弊端：

采用BIO通信模型的服务端，通常由一个独立的Acceptor线程负责监听客户端的连接，接收到客户端连接之后为客户端连接创建一个新的线程处理请求消息，处理完成之后，返回应答消息给客户端，线程销毁，这就是典型的一请求一应答模型。

该架构最大的问题就是不具备弹性伸缩能力，当并发访问量增加后，服务端的线程个数和并发访问数成线性正比，由于线程是JAVA虚拟机非常宝贵的系统资源，当线程数膨胀之后，系统的性能急剧下降，随着并发量的继续增加，可能会发生句柄溢出、线程堆栈溢出等问题，并导致服务器最终宕机。

  • 2序列化方式问题

Java序列化存在如下几个典型问题：

Java序列化机制是Java内部的一种对象编解码技术，无法跨语言使用；例如对于异构系统之间的对接，Java序列化后的码流需要能够通过其它语言反序列化成原始对象（副本），目前很难支持；

相比于其它开源的序列化框架，Java序列化后的码流太大，无论是网络传输还是持久化到磁盘，都会导致额外的资源占用；
序列化性能差（CPU资源占用高）。

  • 3线程模型问题

由于采用同步阻塞IO，这会导致每个TCP连接都占用1个线程，由于线程资源是JVM虚拟机非常宝贵的资源，当IO读写阻塞导致线程无法及时释放时，会导致系统性能急剧下降，严重的甚至会导致虚拟机无法创建新的线程。

高性能的三个主题

1）传输：
用什么样的通道将数据发送给对方，BIO、NIO或者AIO，IO模型在很大程度上决定了框架的性能。

2）协议：
采用什么样的通信协议，HTTP或者内部私有协议。协议的选择不同，性能模型也不同。相比于公有协议，内部私有协议的性能通常可以被设计的更优。

3）线程：
数据报如何读取？读取之后的编解码在哪个线程进行，编解码后的消息如何派发，Reactor线程模型的不同，对性能的影响也非常大。

RPC调用性能三要素：

Netty的高性能之道详解

异步非阻塞通信

在IO编程过程中，当需要同时处理多个客户端接入请求时，可以利用多线程或者IO多路复用技术进行处理。IO多路复用技术通过把多个IO的阻塞复用到同一个select的阻塞上，从而使得系统在单线程的情况下可以同时处理多个客户端请求。与传统的多线程/多进程模型比，I/O多路复用的最大优势是系统开销小，系统不需要创建新的额外进程或者线程，也不需要维护这些进程和线程的运行，降低了系统的维护工作量，节省了系统资源。

JDK1.4提供了对非阻塞IO（NIO）的支持，JDK1.5_update10版本使用epoll替代了传统的select/poll，极大的提升了NIO通信的性能。

JDK NIO通信模型如下所示：

与Socket类和ServerSocket类相对应，NIO也提供了SocketChannel和ServerSocketChannel两种不同的套接字通道实现。这两种新增的通道都支持阻塞和非阻塞两种模式。阻塞模式使用非常简单，但是性能和可靠性都不好，非阻塞模式正好相反。开发人员一般可以根据自己的需要来选择合适的模式，一般来说，低负载、低并发的应用程序可以选择同步阻塞IO以降低编程复杂度。但是对于高负载、高并发的网络应用，需要使用NIO的非阻塞模式进行开发。

Netty架构按照Reactor模式设计和实现，它的服务端通信序列图如下：

客户端通信序列图如下：

Netty的IO线程NioEventLoop由于聚合了多路复用器Selector，可以同时并发处理成百上千个客户端Channel，由于读写操作都是非阻塞的，这就可以充分提升IO线程的运行效率，避免由于频繁IO阻塞导致的线程挂起。另外，由于Netty采用了异步通信模式，一个IO线程可以并发处理N个客户端连接和读写操作，这从根本上解决了传统同步阻塞IO一连接一线程模型，架构的性能、弹性伸缩能力和可靠性都得到了极大的提升。

零拷贝

很多用户都听说过Netty具有“零拷贝”功能，但是具体体现在哪里又说不清楚，本小节就详细对Netty的“零拷贝”功能进行讲解。

Netty的“零拷贝”主要体现在如下三个方面：

  • Netty的接收和发送ByteBuffer采用DIRECT BUFFERS，使用堆外直接内存进行Socket读写，不需要进行字节缓冲区的二次拷贝。如果使用传统的堆内存（HEAP BUFFERS）进行Socket读写，JVM会将堆内存Buffer拷贝一份到直接内存中，然后才写入Socket中。相比于堆外直接内存，消息在发送过程中多了一次缓冲区的内存拷贝。
  • Netty提供了组合Buffer对象，可以聚合多个ByteBuffer对象，用户可以像操作一个Buffer那样方便的对组合Buffer进行操作，避免了传统通过内存拷贝的方式将几个小Buffer合并成一个大的Buffer。
  • Netty的文件传输采用了transferTo方法，它可以直接将文件缓冲区的数据发送到目标Channel，避免了传统通过循环write方式导致的内存拷贝问题。

下面，我们对上述三种“零拷贝”进行说明，先看Netty 接收Buffer的创建（异步消息读取“零拷贝”）：

每循环读取一次消息，就通过ByteBufAllocator的ioBuffer方法获取ByteBuf对象，下面继续看它的接口定义。

ByteBufAllocator 通过ioBuffer分配堆外内存：

当进行Socket IO读写的时候，为了避免从堆内存拷贝一份副本到直接内存，Netty的ByteBuf分配器直接创建非堆内存避免缓冲区的二次拷贝，通过“零拷贝”来提升读写性能。

下面我们继续看第二种“零拷贝”的实现CompositeByteBuf，它对外将多个ByteBuf封装成一个ByteBuf，对外提供统一封装后的ByteBuf接口，它的类定义如下（CompositeByteBuf类继承关系）：

通过继承关系我们可以看出CompositeByteBuf实际就是个ByteBuf的包装器，它将多个ByteBuf组合成一个集合，然后对外提供统一的ByteBuf接口，相关定义如下（CompositeByteBuf类定义）：

添加ByteBuf，不需要做内存拷贝，相关代码如下（新增ByteBuf的“零拷贝”）：

最后，我们看下文件传输的“零拷贝”：

Netty文件传输DefaultFileRegion通过transferTo方法将文件发送到目标Channel中，下面重点看FileChannel的transferTo方法，它的API DOC说明如下：

对于很多操作系统它直接将文件缓冲区的内容发送到目标Channel中，而不需要通过拷贝的方式，这是一种更加高效的传输方式，它实现了文件传输的“零拷贝”。

内存池

随着JVM虚拟机和JIT即时编译技术的发展，对象的分配和回收是个非常轻量级的工作。但是对于缓冲区Buffer，情况却稍有不同，特别是对于堆外直接内存的分配和回收，是一件耗时的操作。为了尽量重用缓冲区，Netty提供了基于内存池的缓冲区重用机制。下面我们一起看下Netty ByteBuf的实现：

Netty提供了多种内存管理策略，通过在启动辅助类中配置相关参数，可以实现差异化的定制。下面通过性能测试，我们看下基于内存池循环利用的ByteBuf和普通ByteBuf的性能差异。

用例一，使用内存池分配器创建直接内存缓冲区：

用例二，使用非堆内存分配器创建的直接内存缓冲区：

各执行300万次，性能对比结果如下所示：

性能测试表明，采用内存池的ByteBuf相比于朝生夕灭的ByteBuf，性能高23倍左右（性能数据与使用场景强相关）。

下面我们一起简单分析下Netty内存池的内存分配：

继续看newDirectBuffer方法，我们发现它是一个抽象方法，由AbstractByteBufAllocator的子类负责具体实现，代码如下：

代码跳转到PooledByteBufAllocator的newDirectBuffer方法，从Cache中获取内存区域PoolArena，调用它的allocate方法进行内存分配：

PoolArena的allocate方法如下：

我们重点分析newByteBuf的实现，它同样是个抽象方法，由子类DirectArena和HeapArena来实现不同类型的缓冲区分配，由于测试用例使用的是堆外内存：

因此重点分析DirectArea的实现：如果没有开启使用sun的unsafe，则：

执行PooledDirectByteBuf的newInstance方法，代码如下：

通过RECYCLER的get方法循环使用ByteBuf对象，如果是非内存池实现，则直接创建一个新的ByteBuf对象。从缓冲池中获取ByteBuf之后，调用AbstractReferenceCountedByteBuf的setRefCnt方法设置引用计数器，用于对象的引用计数和内存回收（类似JVM垃圾回收机制）。

高效的Reactor线程模型

  • 常用的Reactor线程模型有三种，分别如下：

Reactor单线程模型；
Reactor多线程模型；
主从Reactor多线程模型。

  • Reactor单线程模型，指的是所有的IO操作都在同一个NIO线程上面完成，NIO线程的职责如下：

作为NIO服务端，接收客户端的TCP连接；
作为NIO客户端，向服务端发起TCP连接；
读取通信对端的请求或者应答消息；
向通信对端发送消息请求或者应答消息。

Reactor单线程模型示意图如下所示：

由于Reactor模式使用的是异步非阻塞IO，所有的IO操作都不会导致阻塞，理论上一个线程可以独立处理所有IO相关的操作。从架构层面看，一个NIO线程确实可以完成其承担的职责。例如，通过Acceptor接收客户端的TCP连接请求消息，链路建立成功之后，通过Dispatch将对应的ByteBuffer派发到指定的Handler上进行消息解码。用户Handler可以通过NIO线程将消息发送给客户端。

对于一些小容量应用场景，可以使用单线程模型。但是对于高负载、大并发的应用却不合适，主要原因如下：

  • 一个NIO线程同时处理成百上千的链路，性能上无法支撑，即便NIO线程的CPU负荷达到100%，也无法满足海量消息的编码、解码、读取和发送；
  • 当NIO线程负载过重之后，处理速度将变慢，这会导致大量客户端连接超时，超时之后往往会进行重发，这更加重了NIO线程的负载，最终会导致大量消息积压和处理超时，NIO线程会成为系统的性能瓶颈；
  • 可靠性问题：一旦NIO线程意外跑飞，或者进入死循环，会导致整个系统通信模块不可用，不能接收和处理外部消息，造成节点故障。

为了解决这些问题，演进出了Reactor多线程模型，下面我们一起学习下Reactor多线程模型。

Rector多线程模型与单线程模型最大的区别就是有一组NIO线程处理IO操作，它的原理图如下：

Reactor多线程模型的特点：

  • 有专门一个NIO线程-Acceptor线程用于监听服务端，接收客户端的TCP连接请求；
  • 网络IO操作-读、写等由一个NIO线程池负责，线程池可以采用标准的JDK线程池实现，它包含一个任务队列和N个可用的线程，由这些NIO线程负责消息的读取、解码、编码和发送；
  • 1个NIO线程可以同时处理N条链路，但是1个链路只对应1个NIO线程，防止发生并发操作问题。

在绝大多数场景下，Reactor多线程模型都可以满足性能需求；但是，在极特殊应用场景中，一个NIO线程负责监听和处理所有的客户端连接可能会存在性能问题。例如百万客户端并发连接，或者服务端需要对客户端的握手消息进行安全认证，认证本身非常损耗性能。在这类场景下，单独一个Acceptor线程可能会存在性能不足问题，为了解决性能问题，产生了第三种Reactor线程模型-主从Reactor多线程模型。

主从Reactor线程模型的特点是：服务端用于接收客户端连接的不再是个1个单独的NIO线程，而是一个独立的NIO线程池。Acceptor接收到客户端TCP连接请求处理完成后（可能包含接入认证等），将新创建的SocketChannel注册到IO线程池（sub reactor线程池）的某个IO线程上，由它负责SocketChannel的读写和编解码工作。Acceptor线程池仅仅只用于客户端的登陆、握手和安全认证，一旦链路建立成功，就将链路注册到后端subReactor线程池的IO线程上，由IO线程负责后续的IO操作。

它的线程模型如下图所示：

利用主从NIO线程模型，可以解决1个服务端监听线程无法有效处理所有客户端连接的性能不足问题。因此，在Netty的官方demo中，推荐使用该线程模型。

事实上，Netty的线程模型并非固定不变，通过在启动辅助类中创建不同的EventLoopGroup实例并通过适当的参数配置，就可以支持上述三种Reactor线程模型。正是因为Netty 对Reactor线程模型的支持提供了灵活的定制能力，所以可以满足不同业务场景的性能诉求。
调用select函数等待事件的始终是一个线程，当发生客户端连接事件的时候，会去boss线程池中找其他线程进行处理,因为客户端连接事件处理可能会比较耗时

无锁化的串行设计理念

在大多数场景下，并行多线程处理可以提升系统的并发性能。但是，如果对于共享资源的并发访问处理不当，会带来严重的锁竞争，这最终会导致性能的下降。为了尽可能的避免锁竞争带来的性能损耗，可以通过串行化设计，即消息的处理尽可能在同一个线程内完成，期间不进行线程切换，这样就避免了多线程竞争和同步锁。

为了尽可能提升性能，Netty采用了串行无锁化设计，在IO线程内部进行串行操作，避免多线程竞争导致的性能下降。表面上看，串行化设计似乎CPU利用率不高，并发程度不够。但是，通过调整NIO线程池的线程参数，可以同时启动多个串行化的线程并行运行，这种局部无锁化的串行线程设计相比一个队列-多个工作线程模型性能更优。

Netty的串行化设计工作原理图如下：

Netty的NioEventLoop读取到消息之后，直接调用ChannelPipeline的fireChannelRead(Object msg)，只要用户不主动切换线程，一直会由NioEventLoop调用到用户的Handler，期间不进行线程切换，这种串行化处理方式避免了多线程操作导致的锁的竞争，从性能角度看是最优的。
其实就是一个客户端SocketChannel只能绑定到一个worker线程的selector上，而一个worker线程的selector上面可以同时注册多个客户端的socketChannel，同时监听这些客户端的读写事件

高效的并发编程

Netty的高效并发编程主要体现在如下几点：

  • volatile的大量、正确使用;
  • CAS和原子类的广泛使用；
  • 线程安全容器的使用；
  • 通过读写锁提升并发性能。

高性能的序列化框架

影响序列化性能的关键因素总结如下：

  • 序列化后的码流大小（网络带宽的占用）；
  • 序列化&反序列化的性能（CPU资源占用）；
  • 是否支持跨语言（异构系统的对接和开发语言切换）。

Netty默认提供了对Google Protobuf的支持，通过扩展Netty的编解码接口，用户可以实现其它的高性能序列化框架，例如Thrift的压缩二进制编解码框架。

下面我们一起看下不同序列化&反序列化框架序列化后的字节数组对比：

从上图可以看出，Protobuf序列化后的码流只有Java序列化的1/4左右。正是由于Java原生序列化性能表现太差，才催生出了各种高性能的开源序列化技术和框架（性能差只是其中的一个原因，还有跨语言、IDL定义等其它因素）。

灵活的TCP参数配置能力

合理设置TCP参数在某些场景下对于性能的提升可以起到显著的效果，例如SO_RCVBUF和SO_SNDBUF。如果设置不当，对性能的影响是非常大的。

下面我们总结下对性能影响比较大的几个配置项：

  • SO_RCVBUF和SO_SNDBUF：通常建议值为128K或者256K；
  • SO_TCPNODELAY：NAGLE算法通过将缓冲区内的小封包自动相连，组成较大的封包，阻止大量小封包的发送阻塞网络，从而提高网络应用效率。但是对于时延敏感的应用场景需要关闭该优化算法；
  • 软中断：如果Linux内核版本支持RPS（2.6.35以上版本），开启RPS后可以实现软中断，提升网络吞吐量。RPS根据数据包的源地址，目的地址以及目的和源端口，计算出一个hash值，然后根据这个hash值来选择软中断运行的cpu，从上层来看，也就是说将每个连接和cpu绑定，并通过这个hash值，来均衡软中断在多个cpu上，提升网络并行处理性能。

Netty在启动辅助类中可以灵活的配置TCP参数，满足不同的用户场景。相关配置接口定义如下：

实现心跳机制、断线重连机制

为何需要心跳机制？

所谓心跳, 即在 TCP 长连接中, 客户端和服务器之间定期发送的一种特殊的数据包, 通知对方自己还在线, 以确保 TCP 连接的有效性。
注：心跳包还有另一个作用，经常被忽略，即：一个连接如果长时间不用，防火墙或者路由器就会断开该连接（PS：术语叫“端口老化”）。

如何用Netty实现心跳机制？

先理解一下核心Handler：IdleStateHandler

在 Netty 中, 实现心跳机制的关键是 IdleStateHandler, 那么这个 Handler 如何使用呢?

先看下它的构造器：

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 事件.

注：这三个参数默认的时间单位是秒。若需要指定其他时间单位，可以使用另一个构造方法：IdleStateHandler(boolean observeOutput, long readerIdleTime, long writerIdleTime, long allIdleTime, TimeUnit unit)

使用IdleStateHandler实现心跳（客户端篇）

下面将使用IdleStateHandler来实现心跳，Client端连接到Server端后，会循环执行一个任务：随机等待几秒，然后ping一下Server端，即发送一个心跳包。

当等待的时间超过规定时间，将会发送失败，以为Server端在此之前已经主动断开连接了。

实现代码如下。

ClientIdleStateTrigger —— 自定义心跳触发器类
ClientIdleStateTrigger也是一个Handler，只是重写了userEventTriggered方法，用于捕获IdleState.WRITER_IDLE事件（未在指定时间内向服务器发送数据），然后向Server端发送一个心跳包。

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;

/**
 * <p>
 *  用于捕获{@link IdleState#WRITER_IDLE}事件（未在指定时间内向服务器发送数据），然后向Server端发送一个心跳包。
 * </p>
 */
public class ClientIdleStateTrigger extends ChannelInboundHandlerAdapter {
 
    public static final String HEART_BEAT = "heart beat!";
 
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleState state = ((IdleStateEvent) evt).state();
            if (state == IdleState.WRITER_IDLE) {
                // write heartbeat to server
                ctx.writeAndFlush(HEART_BEAT);
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }
 
}

Pinger —— 心跳发射器：

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import io.netty.util.concurrent.ScheduledFuture;

import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * <p>客户端连接到服务器端后，会循环执行一个任务：随机等待几秒，然后ping一下Server端，即发送一个心跳包。</p>
 */
public class Pinger extends ChannelInboundHandlerAdapter {
 
    private Random random = new Random();
    private int baseRandom = 8;
 
    private Channel channel;
 
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
        this.channel = ctx.channel();
 
        ping(ctx.channel());
    }
 
    private void ping(Channel channel) {
        int second = Math.max(1, random.nextInt(baseRandom));
        System.out.println("next heart beat will send after " + second + "s.");
        ScheduledFuture<?> future = channel.eventLoop().schedule(new Runnable() {
            @Override
            public void run() {
                if (channel.isActive()) {
                    System.out.println("sending heart beat to the server...");
                    channel.writeAndFlush(ClientIdleStateTrigger.HEART_BEAT);
                } else {
                    System.err.println("The connection had broken, cancel the task that will send a heart beat.");
                    channel.closeFuture();
                    throw new RuntimeException();
                }
            }
        }, second, TimeUnit.SECONDS);
 
        future.addListener(new GenericFutureListener() {
            @Override
            public void operationComplete(Future future) throws Exception {
                if (future.isSuccess()) {
                    ping(channel);
                }
            }
        });
    }
 
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        // 当Channel已经断开的情况下, 仍然发送数据, 会抛异常, 该方法会被调用.
        cause.printStackTrace();
        ctx.close();
    }
}

ClientHandlersInitializer —— 客户端处理器集合的初始化类：

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.CharsetUtil;

public class ClientHandlersInitializer extends ChannelInitializer<SocketChannel> {
 
    private ReconnectHandler reconnectHandler;
    private EchoHandler echoHandler;
 
    public ClientHandlersInitializer(TcpClient tcpClient) {
        Assert.notNull(tcpClient, "TcpClient can not be null.");
        this.reconnectHandler = new ReconnectHandler(tcpClient);
        this.echoHandler = new EchoHandler();
    }
 
    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
        pipeline.addLast(new LengthFieldPrepender(4));
        pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
        pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
        pipeline.addLast(new Pinger());
    }
}
 
class EchoHandler extends SimpleChannelInboundHandler<String> {
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, String data) throws Exception {
                try {
                        System.out.println("receive data from server: " + data);
                } catch (Exception e) {
                        e.printStackTrace();
                }
        }
 
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
                cause.printStackTrace();
                ctx.close();
        }
}

注：上面的Handler集合，除了Pinger，其他都是编解码器和解决粘包，可以忽略。

TcpClient —— TCP连接的客户端：

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;

public class TcpClient {
 
    private String host;
    private int port;
    private Bootstrap bootstrap;
    /** 将<code>Channel</code>保存起来, 可用于在其他非handler的地方发送数据 */
    private Channel channel;
 
    public TcpClient(String host, int port) {
        this(host, port, new ExponentialBackOffRetry(1000, Integer.MAX_VALUE, 60 * 1000));
    }
 
    public TcpClient(String host, int port, RetryPolicy retryPolicy) {
        this.host = host;
        this.port = port;
        init();
    }
 
    /**
     * 向远程TCP服务器请求连接
     */
    public void connect() {
        synchronized (bootstrap) {
            ChannelFuture future = bootstrap.connect(host, port);
            this.channel = future.channel();
        }
    }
 
    private void init() {
        EventLoopGroup group = new NioEventLoopGroup();
        // bootstrap 可重用, 只需在TcpClient实例化的时候初始化即可.
        bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ClientHandlersInitializer(TcpClient.this));
    }
 
    public static void main(String[] args) {
        TcpClient tcpClient = new TcpClient("localhost", 2222);
        tcpClient.connect();
    }
}

使用IdleStateHandler实现心跳（服务端篇）

ServerIdleStateTrigger —— 断连触发器：

/**
 * <p>在规定时间内未收到客户端的任何数据包, 将主动断开该连接</p>
 */
public class ServerIdleStateTrigger extends ChannelInboundHandlerAdapter {
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleState state = ((IdleStateEvent) evt).state();
            if (state == IdleState.READER_IDLE) {
                // 在规定时间内没有收到客户端的上传的数据, 主动断开连接
                ctx.disconnect();
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }
}

ServerBizHandler —— 服务器端的业务处理器：

/**
 * <p>收到来自客户端的数据包后, 直接在控制台打印出来.</p>
 */
@ChannelHandler.Sharable
public class ServerBizHandler extends SimpleChannelInboundHandler<String> {
 
    private final String REC_HEART_BEAT = "I had received the heart beat!";
 
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String data) throws Exception {
        try {
            System.out.println("receive data: " + data);
//            ctx.writeAndFlush(REC_HEART_BEAT);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
 
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("Established connection with the remote client.");
 
        // do something
 
        ctx.fireChannelActive();
    }
 
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("Disconnected with the remote client.");
 
        // do something
 
        ctx.fireChannelInactive();
    }
 
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

ServerHandlerInitializer —— 服务器端处理器集合的初始化类：

import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.timeout.IdleStateHandler;

/**
 * <p>用于初始化服务器端涉及到的所有<code>Handler</code></p>
 */
public class ServerHandlerInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ch.pipeline().addLast("idleStateHandler", new IdleStateHandler(5, 0, 0));
        ch.pipeline().addLast("idleStateTrigger", new ServerIdleStateTrigger());
        ch.pipeline().addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
        ch.pipeline().addLast("frameEncoder", new LengthFieldPrepender(4));
        ch.pipeline().addLast("decoder", new StringDecoder());
        ch.pipeline().addLast("encoder", new StringEncoder());
        ch.pipeline().addLast("bizHandler", new ServerBizHandler());
    }
}

注：new IdleStateHandler(5, 0, 0)该handler代表如果在5秒内没有收到来自客户端的任何数据包（包括但不限于心跳包），将会主动断开与该客户端的连接。

TcpServer —— 服务器端：

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;

public class TcpServer {
    private int port;
    private ServerHandlerInitializer serverHandlerInitializer;
 
    public TcpServer(int port) {
        this.port = port;
        this.serverHandlerInitializer = new ServerHandlerInitializer();
    }
 
    public void start() {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(this.serverHandlerInitializer);
            // 绑定端口，开始接收进来的连接
            ChannelFuture future = bootstrap.bind(port).sync();
 
            System.out.println("Server start listen at " + port);
            future.channel().closeFuture().sync();
        } catch (Exception e) {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
            e.printStackTrace();
        }
    }
 
    public static void main(String[] args) throws Exception {
        int port = 2222;
        new TcpServer(port).start();
    }
}

运行测试

首先启动服务端，再启动客户端。

启动完成后，在客户端的控制台上，可以看到打印如下类似日志：

在服务器端可以看到控制台输出了类似如下的日志：

可以看到，客户端在发送4个心跳包后，第5个包因为等待时间较长，等到真正发送的时候，发现连接已断开了；而服务器端收到客户端的4个心跳数据包后，迟迟等不到下一个数据包，所以果断断开该连接。

在测试过程中，有可能会出现如下异常情况：

出现这种情况的原因是：在连接已断开的情况下，仍然向服务器端发送心跳包。虽然在发送心跳包之前会使用 channel.isActive() 判断连接是否可用，但也有可能上一刻判断结果为可用，但下一刻发送数据包之前，连接就断了。

如何用Netty实现网络断线重连机制？

断线重连对于复杂网络非常有用（没有这个，一旦断网，通信就无法自动恢复了），这里就不过多介绍，相信各位都知道是怎么回事。这里只说大致思路，然后直接上代码。

实现思路

客户端在监测到与服务器端的连接断开后，或者一开始就无法连接的情况下，使用指定的重连策略进行重连操作，直到重新建立连接或重试次数耗尽。

对于如何监测连接是否断开，则是通过重写ChannelInboundHandler#channelInactive来实现，但连接不可用，该方法会被触发，所以只需要在该方法做好重连工作即可。

代码实现

注：以下代码都是在上一节心跳机制代码的基础上修改/添加的。因为断线重连是客户端的工作，所以只需对客户端代码进行修改。

重试策略：

1）RetryPolicy —— 重试策略接口：

public interface RetryPolicy {
 
    /**
     * Called when an operation has failed for some reason. This method should return
     * true to make another attempt.
     *
     * @param retryCount the number of times retried so far (0 the first time)
     * @return true/false
     */
    boolean allowRetry(int retryCount);
 
    /**
     * get sleep time in ms of current retry count.
     *
     * @param retryCount current retry count
     * @return the time to sleep
     */
    long getSleepTimeMs(int retryCount);
}

2）ExponentialBackOffRetry —— 重连策略的默认实现：

/**
 * <p>Retry policy that retries a set number of times with increasing sleep time between retries</p>
 */
public class ExponentialBackOffRetry implements RetryPolicy {
 
    private static final int MAX_RETRIES_LIMIT = 29;
    private static final int DEFAULT_MAX_SLEEP_MS = Integer.MAX_VALUE;
 
    private final Random random = new Random();
    private final long baseSleepTimeMs;
    private final int maxRetries;
    private final int maxSleepMs;
 
    public ExponentialBackOffRetry(int baseSleepTimeMs, int maxRetries) {
        this(baseSleepTimeMs, maxRetries, DEFAULT_MAX_SLEEP_MS);
    }
 
    public ExponentialBackOffRetry(int baseSleepTimeMs, int maxRetries, int maxSleepMs) {
        this.maxRetries = maxRetries;
        this.baseSleepTimeMs = baseSleepTimeMs;
        this.maxSleepMs = maxSleepMs;
    }
 
    @Override
    public boolean allowRetry(int retryCount) {
        if (retryCount < maxRetries) {
            return true;
        }
        return false;
    }
 
    @Override
    public long getSleepTimeMs(int retryCount) {
        if (retryCount < 0) {
            throw new IllegalArgumentException("retries count must greater than 0.");
        }
        if (retryCount > MAX_RETRIES_LIMIT) {
            System.out.println(String.format("maxRetries too large (%d). Pinning to %d", maxRetries, MAX_RETRIES_LIMIT));
            retryCount = MAX_RETRIES_LIMIT;
        }
        long sleepMs = baseSleepTimeMs * Math.max(1, random.nextInt(1 << retryCount));
        if (sleepMs > maxSleepMs) {
            System.out.println(String.format("Sleep extension too large (%d). Pinning to %d", sleepMs, maxSleepMs));
            sleepMs = maxSleepMs;
        }
        return sleepMs;
    }
}

ReconnectHandler—— 重连处理器：

import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.EventLoop;

import java.util.concurrent.TimeUnit;

@ChannelHandler.Sharable
public class ReconnectHandler extends ChannelInboundHandlerAdapter {
 
    private int retries = 0;
    private RetryPolicy retryPolicy;
 
    private TcpClient tcpClient;
 
    public ReconnectHandler(TcpClient tcpClient) {
        this.tcpClient = tcpClient;
    }
 
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("Successfully established a connection to the server.");
        retries = 0;
        ctx.fireChannelActive();
    }
 
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        if (retries == 0) {
            System.err.println("Lost the TCP connection with the server.");
            ctx.close();
        }
 
        boolean allowRetry = getRetryPolicy().allowRetry(retries);
        if (allowRetry) {
 
            long sleepTimeMs = getRetryPolicy().getSleepTimeMs(retries);
 
            System.out.println(String.format("Try to reconnect to the server after %dms. Retry count: %d.", sleepTimeMs, ++retries));
 
            final EventLoop eventLoop = ctx.channel().eventLoop();
            eventLoop.schedule(() -> {
                System.out.println("Reconnecting ...");
                tcpClient.connect();
            }, sleepTimeMs, TimeUnit.MILLISECONDS);
        }
        ctx.fireChannelInactive();
    }
 
 
    private RetryPolicy getRetryPolicy() {
        if (this.retryPolicy == null) {
            this.retryPolicy = tcpClient.getRetryPolicy();
        }
        return this.retryPolicy;
    }
}

ClientHandlersInitializer：

在之前的基础上，添加了重连处理器ReconnectHandler。

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.CharsetUtil;

public class ClientHandlersInitializer extends ChannelInitializer<SocketChannel> {
 
    private ReconnectHandler reconnectHandler;
    private EchoHandler echoHandler;
 
    public ClientHandlersInitializer(TcpClient tcpClient) {
        Assert.notNull(tcpClient, "TcpClient can not be null.");
        this.reconnectHandler = new ReconnectHandler(tcpClient);
        this.echoHandler = new EchoHandler();
    }
 
    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        pipeline.addLast(this.reconnectHandler);
        pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
        pipeline.addLast(new LengthFieldPrepender(4));
        pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
        pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
        pipeline.addLast(new Pinger());
    }
}
 
class EchoHandler extends SimpleChannelInboundHandler<String> {
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, String data) throws Exception {
                try {
                        System.out.println("receive data from server: " + data);
                } catch (Exception e) {
                        e.printStackTrace();
                }
        }
 
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
                cause.printStackTrace();
                ctx.close();
        }
}

TcpClient：
在之前的基础上添加重连、重连策略的支持。

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;

public class TcpClient {
 
    private String host;
    private int port;
    private Bootstrap bootstrap;
    /** 将<code>Channel</code>保存起来, 可用于在其他非handler的地方发送数据 */
    private Channel channel;
    /** 重连策略 */
    private RetryPolicy retryPolicy;

    public TcpClient(String host, int port) {
        this(host, port, new ExponentialBackOffRetry(1000, Integer.MAX_VALUE, 60 * 1000));
    }
 
    public TcpClient(String host, int port, RetryPolicy retryPolicy) {
        this.host = host;
        this.port = port;
        this.retryPolicy=retryPolicy;
        init();
    }
 
    /**
     * 向远程TCP服务器请求连接
     */
    public void connect() {
        synchronized (bootstrap) {
            ChannelFuture future = bootstrap.connect(host, port);
            future.addListener(getConnectionListener());
            this.channel = future.channel();
        }
    }

    private ChannelFutureListener getConnectionListener() {
        return new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (!future.isSuccess()) {
                    future.channel().pipeline().fireChannelInactive();
                }
            }
        };
    }

    private void init() {
        EventLoopGroup group = new NioEventLoopGroup();
        // bootstrap 可重用, 只需在TcpClient实例化的时候初始化即可.
        bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ClientHandlersInitializer(TcpClient.this));
    }
 
    public static void main(String[] args) {
        TcpClient tcpClient = new TcpClient("localhost", 2222);
        tcpClient.connect();
    }
        public RetryPolicy getRetryPolicy() {
    return retryPolicy;
    }
}

运行测试

在测试之前，为了避开 Connection reset by peer 异常，可以稍微修改Pinger的ping()方法，添加if (second == 5)的条件判断。如下：

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import io.netty.util.concurrent.ScheduledFuture;

import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * <p>客户端连接到服务器端后，会循环执行一个任务：随机等待几秒，然后ping一下Server端，即发送一个心跳包。</p>
 */
public class Pinger extends ChannelInboundHandlerAdapter {
 
    private Random random = new Random();
    private int baseRandom = 8;
 
    private Channel channel;
 
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
        this.channel = ctx.channel();
 
        ping(ctx.channel());
    }
 
    private void ping(Channel channel) {
        int second = Math.max(1, random.nextInt(baseRandom));
        if (second == 5) {
            second = 6;
        }
        System.out.println("next heart beat will send after " + second + "s.");
        ScheduledFuture<?> future = channel.eventLoop().schedule(new Runnable() {
            @Override
            public void run() {
                if (channel.isActive()) {
                    System.out.println("sending heart beat to the server...");
                    channel.writeAndFlush(ClientIdleStateTrigger.HEART_BEAT);
                } else {
                    System.err.println("The connection had broken, cancel the task that will send a heart beat.");
                    channel.closeFuture();
                    throw new RuntimeException();
                }
            }
        }, second, TimeUnit.SECONDS);
 
        future.addListener(new GenericFutureListener() {
            @Override
            public void operationComplete(Future future) throws Exception {
                if (future.isSuccess()) {
                    ping(channel);
                }
            }
        });
    }
 
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        // 当Channel已经断开的情况下, 仍然发送数据, 会抛异常, 该方法会被调用.
        cause.printStackTrace();
        ctx.close();
    }
}

1）启动客户端：

先只启动客户端，观察控制台输出，可以看到类似如下日志：

可以看到，当客户端发现无法连接到服务器端，所以一直尝试重连。随着重试次数增加，重试时间间隔越大，但又不想无限增大下去，所以需要定一个阈值，比如60s。如上图所示，当下一次重试时间超过60s时，会打印Sleep extension too large(*). Pinning to 60000，单位为ms。出现这句话的意思是，计算出来的时间超过阈值（60s），所以把真正睡眠的时间重置为阈值（60s）。

2）启动服务器端：

接着启动服务器端，然后继续观察客户端控制台输出。

可以看到，在第9次重试失败后，第10次重试之前，启动的服务器，所以第10次重连的结果为Successfully established a connection to the server.，即成功连接到服务器。接下来因为还是不定时ping服务器，所以出现断线重连、断线重连的循环。

3）扩展：

在不同环境，可能会有不同的重连需求。有不同的重连需求的，只需自己实现RetryPolicy接口，然后在创建TcpClient的时候覆盖默认的重连策略即可。

源码剖析

IdleStateHandler

Netty提供了对心跳机制的天然支持，心跳可以检测远程端是否存活，或者活跃

今天我们就一起初识一下Netty4的心跳机制

Netty4.0提供了一个类，名为IdleStateHandler，这个类可以对三种类型的心跳检测

这个类的构造参数是这样的：

前三个的参数解释如下：

1）readerIdleTime：为读超时时间（即测试端一定时间内未接受到被测试端消息）

2）writerIdleTime：为写超时时间（即测试端一定时间内向被测试端发送消息）

3）allIdleTime：所有类型的超时时间

这个类主要也是一个ChannelHandler，也需要被载入到ChannelPipeline中，加入我们在服务器端的ChannelInitializer中加入如下的代码：

我们在channel链中加入了IdleSateHandler，第一个参数是5，单位是秒，那么这样做的意思就是：在服务器端会每隔5秒来检查一下channelRead方法被调用的情况，如果在5秒内该链上的channelRead方法都没有被触发，就会调用userEventTriggered方法：

先看channelActive方法：

我们知道ChannelActive会在通道准备就绪后被调用，因此初始化方法在这里就开始计时了，我们可以追踪看一下

我们下面看一下这个任务的run方法到底是执行什么操作:

这上面schedule第二个参数的this,就是当前任务队列，表示下一次时间一到继续调用该run方法

firstReaderIdleEvent主要作用是标注当前读超时事件是否是第一次发生，这样在newIdleStateEvent的时候，会有小区别

channelIdle本质是触发上面的事件，传入对应的事件类型，供我们进行判断

下面我们需要追踪一下，最后一次读事件的时间会在哪里被更新

至于写空闲事件和读写空闲事件，实现的思路类似，大家可以自行翻找源码

Netty的优雅退出机制和原理

进程的优雅退出

Kill -9 PID带来的问题

  • 在Linux上通常会通过kill -9 pid的方式强制将某个进程杀掉，这种方式简单高效，因此很多程序的停止脚本经常会选择使用kill -9 pid的方式。
  • 无论是Linux的Kill -9 pid还是windows的taskkill /f /pid强制进程退出,都会带来一些副作用：对应用软件而言其效果等同于突然掉电，可能会导致如下一些问题：
  • 缓存中的数据尚未持久化到磁盘中，导致数据丢失；
  • 正在进行文件的write操作，没有更新完成，突然退出，导致文件损坏；
  • 线程的消息队列中尚有接收到的请求消息还没来得及处理，导致请求消息丢失；
  • 数据库操作已经完成，例如账户余额更新，准备返回应答消息给客户端时，消息尚在通信线程的发送队列中排队等待发送，进程强制退出导致应答消息没有返回给客户端，客户端发起超时重试，会带来重复更新问题；
  • 其它问题等…

Java如何优雅地退出

Java的优雅停机通常通过注册JDK的ShutdownHook来实现，当系统接收到退出指令后，首先标记系统处于退出状态，不再接收新的消息，然后将积压的消息处理完，最后调用资源回收接口将资源销毁，最后各线程退出执行。

通常优雅退出需要有超时控制机制，例如30S，如果到达超时时间仍然没有完成退出前的资源回收等操作，则由停机脚本直接调用kill -9 pid，强制退出。

如何实现Netty的优雅退出

要实现Netty的优雅退出，首先需要了解通用Java进程的优雅退出如何实现。下面我们先讲解下优雅退出的实现原理，并结合实际代码进行讲解。最后看下如何实现Netty的优雅退出。

信号简介

信号是在软件层次上对中断机制的一种模拟，在原理上，一个进程收到一个信号与处理器收到一个中断请求可以说是一样的，它是进程间一种异步通信的机制。以Linux的kill命令为例，kill -s SIGKILL pid (即kill -9 pid) 立即杀死指定pid的进程，SIGKILL就是发送给pid进程的信号。

信号具有平台相关性，Linux平台支持的一些终止进程信号如下所示：

Windows平台存在一些差异，它的一些信号举例如下：SIGINT（Ctrl+C中断）、SIGILL、SIGTERM （kill发出的软件终止）、SIGBREAK （Ctrl+Break中断）

信号选择：为了不干扰正常信号的运作，又能模拟Java异步通知，在Linux上我们需要先选定一种特殊的信号。通过查看信号列表上的描述，发现 SIGUSR1 和 SIGUSR2 是允许用户自定义的信号,我们可以选择SIGUSR2，为了测试方便，在Windows上我们可以选择SIGINT。

Java程序的优雅退出

首先看下通用的Java进程优雅退出的流程图：

第一步，应用进程启动的时候，初始化Signal实例，它的代码示例如下：

Signal sig = new Signal(getOSSignalType());

其中Signal构造函数的参数为String字符串，也就是上面介绍的信号量名称。

第二步，根据操作系统的名称来获取对应的信号名称，代码如下：

private String getOSSignalType()
   {
       return System.getProperties().getProperty("os.name").
                 toLowerCase().startsWith("win") ? "INT" : "USR2";
    }

判断是否是windows操作系统，如果是则选择SIGINT，接收Ctrl+C中断的指令；否则选择USR2信号，接收SIGUSR2（等价于kill -12 pid）指令。

第三步，将实例化之后的SignalHandler注册到JDK的Signal，一旦Java进程接收到kill -12 或者 Ctrl+C则回调handle接口，代码示例如下：

Signal.handle(sig, shutdownHandler);

其中shutdownHandler实现了SignalHandler接口的handle(Signal sgin)方法，代码示例如

第四步，在接收到信号回调的handle接口中，初始化JDK的ShutdownHook线程，并将其注册到Runtime中，示例代码如下：

private void invokeShutdownHook()
 {
        Thread t = new Thread(new ShutdownHook(), "ShutdownHook-Thread");
        Runtime.getRuntime().addShutdownHook(t);
 }

第五步，接收到进程退出信号后，在回调的handle接口中执行虚拟机的退出操作，示例代码如下：

Runtime.getRuntime().exit(0);

虚拟机退出时，底层会自动检测用户是否注册了ShutdownHook任务，如果有，则会自动将ShutdownHook线程拉起，执行它的Run方法，用户只需要在ShutdownHook中执行资源释放操作即可，示例代码如下：

class ShutdownHook implements Runnable
{
        @Override
        public void run() {
                System.out.println("ShutdownHook execute start...");
                System.out.print("Netty NioEventLoopGroup shutdownGracefully...");
                try {
                        TimeUnit.SECONDS.sleep(10);//模拟应用进程退出前的处理操作
                } catch (InterruptedException e) {
                                e.printStackTrace();
                }
                System.out.println("ShutdownHook execute end...");
        System.out.println("Sytem shutdown over, the cost time is 10000MS");
                }
}

下面我们在Windows环境中对通用的Java优雅退出程序进行测试，打开CMD控制台，拉起待测试程序，如下所示。

启动进程：

查看线程信息，发现注册的ShutdownHook线程没有启动，符合预期：

在控制台执行Ctrl+C，使进程退出，示例如下：

如上图所示，我们定义的ShutdownHook线程在JVM退出时被执行，作为测试程序，它休眠10S之后退出，控制台打印的相关信息如下：

下面我们总结下通用的Java程序优雅退出的技术要点：

Netty的优雅退出

在实际项目中，Netty作为高性能的异步NIO通信框架，往往用作基础通信框架负责各种协议的接入、解析和调度等，例如在RPC和分布式服务框架中，往往会使用Netty作为内部私有协议的基础通信框架。

当应用进程优雅退出时，作为通信框架的Netty也需要优雅退出，主要原因如下：

  • 尽快的释放NIO线程、句柄等资源；
  • 如果使用flush做批量消息发送，需要将积攒在发送队列中的待发送消息发送完成；
  • 正在write或者read的消息，需要继续处理；
  • 设置在NioEventLoop线程调度器中的定时任务，需要执行或者清理。

下面我们看下Netty优雅退出涉及的主要操作和资源对象：

Netty的优雅退出总结起来有三大步操作：

  • 把NIO线程的状态位设置成ST_SHUTTING_DOWN状态，不再处理新的消息（不允许再对外发送消息）；
  • 退出前的预处理操作：把发送队列中尚未发送或者正在发送的消息发送完、把已经到期或者在退出超时之前到期的定时任务执行完成、把用户注册到NIO线程的退出Hook任务执行完成；
  • 资源的释放操作：所有Channel的释放、多路复用器的去注册和关闭、所有队列和定时任务的清空取消，最后是NIO线程的退出。

下面我们具体看下如何实现Netty的优雅退出。

Netty优雅退出的接口和总入口在EventLoopGroup，调用它的shutdownGracefully方法即可，相关代码如下：

bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();

除了无参的shutdownGracefully方法，还可以指定退出的超时时间和周期，相关接口定义如下：

EventLoopGroup的shutdownGracefully工作原理下面会做详细讲解，结合Java通用的优雅退出机制，即可实现Netty的优雅退出，相关伪代码如下：

//统一定义JVM退出事件，并将JVM退出事件作为主题对进程内部发布
//所有需要优雅退出的消费者订阅JVM退出事件主题
//监听JVM退出的ShutdownHook被启动之后，发布JVM退出事件
//消费者监听到JVM退出事件，开始执行自身的优雅退出
//如果所有的非守护线程都成功完成优雅退出，进程主动退出
//如果到了退出的超时时间仍然没正常退出，则由停机脚本通过kill -9 pid强杀进程，强制退出

总结一下：

JVM的ShutdownHook被触发之后，调用所有EventLoopGroup实例的shutdownGracefully方法进行优雅退出。由于Netty自身对优雅退出有较完善的支持，所以实现起来相对比较简单。

一些误区

在实际工作中，由于对优雅退出和资源释放的原理不太清楚，或者对Netty的接口不太了解，很容易把优雅退出和资源释放混淆，导致出现各种问题。

如下案例：本意是想把某个Channel关闭，但是却调用了Channel关联的EventLoop的shutdownGracefully，导致把EventLoop线程和注册在该线程持有的多路复用器上所有的Channel都关闭了。

错误代码如下所示：

ctx.channel().eventLoop().shutdownGracefully();

正确的做法如下所示：调用channel的close方法，关闭链路，释放与该Channel相关的资源：

ctx.channel().close();

除非是整个进程优雅退出，一般情况下不会调用EventLoopGroup和EventLoop的shutdownGracefully方法，更多的是链路channel的关闭和资源释放。

Netty优雅退出原理分析

Netty优雅退出涉及到线程组、线程、链路、定时任务等，底层实现细节非常复杂，下面我们就层层分解，通过源码来剖析它的实现原理。

NioEventLoopGroup

NioEventLoopGroup实际是NioEventLoop的线程组，它的优雅退出比较简单，直接遍历EventLoop数组，循环调用它们的shutdownGracefully方法，源码如下：

NioEventLoop

调用NioEventLoop的shutdownGracefully方法，首先就是要修改线程状态为正在关闭状态，它的实现在父类SingleThreadEventExecutor中，它们的继承关系如下：

SingleThreadEventExecutor的shutdownGracefully代码比较简单，就是修改线程的状态位，需要注意的是修改时需要对并发调用做判断，如果是由NioEventLoop自身调用，则不需要加锁，否则需要加锁，代码如下：

解释下为什么要加锁，因为shutdownGracefully是public的方法，任何能够获取到NioEventLoop的代码都可以调用它，在Netty中，业务代码通常不需要直接获取NioEventLoop并操作它，但是Netty对NioEventLoop做了比较厚的封装，它不仅仅只能读写消息，还能够执行定时任务，并作为线程池执行用户自定义Task。因此在Channel中将获取NioEventLoop的方法开放了出来，这就意味着用户只要能够获取到Channel，理论上就会存在并发执行shutdownGracefully的可能，因此在优雅退出的时候做了并发保护。

完成状态修改之后，剩下的操作主要在NioEventLoop中进行，代码如下：

我们继续看下closeAll的实现，它的原理是把注册在selector上的所有Channel都关闭，但是有些Channel正在发送消息，暂时还不能关，需要稍后再执行，核心代码如下：

循环调用Channel Unsafe的close方法，下面我们跳转到Unsafe中，对close方法进行分析。

AbstractUnsafe

AbstractUnsafe的close方法主要做了如下几件事：

1．判断当前该链路是否有消息正在发送，如果有则将关闭操作封装成Task放到eventLoop中稍后再执行：

2．将发送队列清空，不再允许发送新的消息：

3．调用SocketChannel的close方法，关闭链路：

4．调用pipeline的fireChannelInactive，触发链路关闭通知事件：

5．最后是调用deregister，从多路复用器上取消SelectionKey：

至此，优雅退出流程已经完成，这是否意味着NioEventLoop线程可以退出了，其实并非如此。

在此处，只是做了Channel的关闭和从Selector上的去注册，总结如下：

  1. 通过inFlush0来判断当前是否正在发送消息，如果是，则不执行Channel关闭动作，放入NIO线程的任务队列中稍后再执行close()操作；
  2. 因为已经不允许新的发送消息加入，一旦发送操作完成，就执行链路关闭、触发链路关闭事件和从Selector上取消注册操作。

之前已经说了，NioEventLoop除了I/O读写之外，还兼具定时任务执行、关闭ShutdownHook的执行等，如果此时有到期的定时任务，即使Chanel已经关闭，但是仍然需要继续执行，线程不能退出。下面我们具体分析下TaskQueue的处理流程。

TaskQueue

NioEventLoop执行完closeAll（）操作之后，需要调用confirmShutdown看是否真的能够退出，它的处理逻辑如下。

1．执行TaskQueue中排队的Task，代码如下：

2．执行注册到NioEventLoop中的ShutdownHook，代码如下：

3．判断是否到达优雅退出的指定超时时间，如果达到或者过了超时时间，则立即退出，代码如下：

4．如果没到达指定的超时时间，暂时不退出，每隔100MS检测下是否有新的任务加入，有则继续执行：

在confirmShutdown方法中，夹杂了一些对已经废弃的shutdown（）方法的处理，例如：

调用新的shutdownGracefully系列方法，该判断条件是永远都不会成立的，因此对于已经废弃的shutdown相关的处理逻辑，不再详细分析。

到此为止，confirmShutdown方法讲解完毕，confirmShutdown返回true，则NioEventLoop线程正式退出，Netty的优雅退出完成，代码如下：

疑问解答

runAllTasks重复执行问题

在NioEventLoop的run方法中，已经调用了runAllTasks方法，为何紧随其后，在confirmShutdown中有继续调用runAllTasks方法呢，疑问代码如下：

原因主要有两个：

  • 为了防止定时任务Task或者用户自定义的线程Task的执行过多占用NioEventLoop线程的调度资源，Netty对NioEventLoop线程I/O操作和非I/O操作时间做了比例限制，即限制非I/O操作的执行时间，如上图红框中代码所示。有了执行时间限制，因此可能会导致已经到期的定时任务、普通任务没有执行完，需要等待下次Selector轮询继续执行。在线程退出之前，需要对本该执行但是没有执行完成的Task进行扫尾处理，所以在confirmShutdown中再次调用了runAllTasks方法；
  • 在调用runAllTasks方法之后，执行confirmShutdown之前，用户向NioEventLoop中添加了新的普通任务或者定时任务，因此需要在退出之前再次遍历并处理一遍Task Queue。

优雅退出是否能够保证所有在通信线程排队的消息全部发送出去

实际是无法保证的，它只能保证如果现在正在发送消息过程中，调用了优雅退出方法，此时不会关闭链路，继续发送，如果发送操作完成，无论是否还有消息尚未发送出去，在下一轮Selector的轮询中，链路将会关闭，没有发送完成的消息将会被丢弃，甚至是半包消息。

它的处理原理图如下：

它的原理比较复杂，现对主要逻辑处理进行解读：

  • 调用优雅退出之后，是否关闭链路，判断标准是inFlush0是否为true，如果为False，则会执行链路关闭操作；
  • 如果用户是类似批量发送，例如每达到N条或者定时触发flush操作，则在此期间调用优雅退出方法，inFlush0为False，链路关闭，积压的待发送消息会被丢弃掉；
  • 如果优雅退出时链路正好在发送消息过程中，则它不会立即退出，等待发送完成之后，下次Selector轮询的时候才退出。在这种场景下，又有两种可能的场景。

场景A：如果一次把积压的消息全部发送完，没有发生写半包，则不会发生消息丢失。

场景B：如果一次没有把消息发送完成，此时Netty会监听写事件，触发Selector的下一次轮询并发送消息，代码如下：

Selector轮询时，首先处理读写事件，然后再处理定时任务和普通任务，因此在链路关闭之前，还有最后一次继续发送的机会，代码如下：

如果非常不幸，再次发送仍然没有把积压的消息全部发送完毕，再次发生了写半包，那无论是否有积压消息，执行AbstractUnsafe.close的Task还是会把链路给关闭掉，原因是只要完成一次消息发送操作，Netty就会把inFlush0置为false，代码如下：

链路关闭之后，所有尚未发送的消息都将被丢弃。

可能有些读者会有疑问，如果在第二次发送之后，执行AbstractUnsafe.close之前，业务正好又调用了flush操作，inFlush0是否会被修改成True呢？这个是不可能的，因为从Netty 4.X之后线程模型发生了变更，flush操作不是由用户线程执行，而是由Channel对应的NioEventLoop线程执行，所以在两者之间不会发生inFlush0被修改的情况。

Netty 4.X之后的线程模型如下所示：

另外，由于优雅退出有超时时间，如果在超时时间内没有完成积压消息的发送，也会发生消息丢弃的情况。对于上述场景，需要应用层来保证相关的可靠性，或者对Netty的优雅退出机制进行优化。
本卷内容已经比较长了，暂时到此结束，更多的内容将会在第八卷呈现

版权说明 : 本文为转载文章, 版权归原作者所有 版权申明

原文链接 : https://cjdhy.blog.csdn.net/article/details/122736765

内容来源于网络，如有侵权，请联系作者删除！

Javarpcnio网络

相关文章

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多

    目录

    • 蜀ICP备2022004421号-2 NULL123 https://www.null123.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com