个人专题目录

---

1. Netty 核心模块组件

1.1 Bootstrap、ServerBootstrap

1. Bootstrap 意思是引导，一个 Netty 应用通常由一个 Bootstrap 开始，主要作用是配置整个 Netty 程序，串联各个组件，Netty 中 Bootstrap 类是客户端程序的启动引导类，ServerBootstrap 是服务端启动引导类
2. 常见的方法有

public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup)，该方法用于服务器端，用来设置两个 EventLoop
public B group(EventLoopGroup group) ，该方法用于客户端，用来设置一个 EventLoop
public B channel(Class<? extends C> channelClass)，该方法用来设置一个服务器端的通道实现
public <T> B option(ChannelOption<T> option, T value)，用来给 ServerChannel 添加配置
public <T> ServerBootstrap childOption(ChannelOption<T> childOption, T value)，用来给接收到的通道添加配置
public ServerBootstrap childHandler(ChannelHandler childHandler)，该方法用来设置业务处理类（自定义的 handler）
public ChannelFuture bind(int inetPort) ，该方法用于服务器端，用来设置占用的端口号
public ChannelFuture connect(String inetHost, int inetPort) ，该方法用于客户端，用来连接服务器

1.2 Future、ChannelFuture

1. Netty 中所有的 IO 操作都是异步的，不能立刻得知消息是否被正确处理。但是可以过一会等它执行完成或者直接注册一个监听，具体的实现就是通过 Future 和 ChannelFutures，他们可以注册一个监听，当操作执行成功或失败时监听会自动触发注册的监听事件

2. 常见的方法有

Channel channel()，返回当前正在进行 IO 操作的通道
ChannelFuture sync()，等待异步操作执行完毕

在并发编程中，我们通常会用到一组非阻塞的模型:Promise,Future,Callback。其中的Future表示一个可能还没有实际完成的异步任务的结果,针对这个结果添加Callback以便在执行任务成功或者失败后做出响应的操作。
而经由Promise交给执行者，任务执行者通过Promise可以标记任务完成或者失败。以上这套模型是很多异步非阻塞框架的基础。
具体的理解可参见JDK的FutureTask和Callable。JDK的实现版本，在获取最终结果的时候，不得不做一些阻塞的方法等待最终结果的到来。

Netty的Future机制是JDK机制的一个子版本，它支持给Future添加Listener，以方便EventLoop在任务调度完成之后调用。

Future提供了一套高效便捷的非阻塞并行操作管理方案。其基本思想很简单，所谓Future，指的是一类占位符对象，用于指代某些尚未完成的计算的结果。一般来说，由Future指代的计算都是并行执行的，计算完毕后可另行获取相关计算结果。以这种方式组织并行任务，便可以写出高效、异步、非阻塞的并行代码。

默认情况下，future和promise并不采用一般的阻塞操作，而是依赖回调进行非阻塞操作。为了在语法和概念层面更加简明扼要地使用这些回调。当然，future仍然支持阻塞操作——必要时，可以阻塞(sync、await、awaitUninterruptibly)等待future（不过并不鼓励这样做）。

Future

所谓Future，是一种用于指代某个尚未就绪的值的对象。而这个值，往往是某个计算过程的结果：

• 若该计算过程尚未完成，我们就说该Future未就位；
• 若该计算过程正常结束，或中途抛出异常，我们就说该Future已就位。

Future的就位分为两种情况：

• 当Future带着某个值就位时，我们就说该Future携带计算结果成功就位。
• 当Future因对应计算过程抛出异常而就绪，我们就说这个Future因该异常而失败。

Future的一个重要属性在于它只能被赋值一次。一旦给定了某个值或某个异常，future对象就变成了不可变对象——无法再被改写。

Callbacks(回调函数)

Callback用于对计算的最终结果Future做一些后续的处理，以便我们能够用它来做一些有用的事。我们经常对计算结果感兴趣而不仅仅是它的副作用。

Promises

如果说futures是为了一个还没有存在的结果，而当成一种只读占位符的对象类型去创建，那么Promise就被认为是一个可写的，可以实现一个Future的单一赋值容器。这就是说，promise通过这种success方法可以成功去实现一个带有值的future。相反的，因为一个失败的promise通过failure方法就会实现一个带有异常的future。

Future和Promise的区别

Promise与Future的区别在于，Future是Promise的一个只读的视图，也就是说Future没有设置任务结果的方法，只能获取任务执行结果或者为Future添加回调函数。

1.3 Channel

1. Netty 网络通信的组件，能够用于执行网络 I/O 操作。

2. 通过Channel 可获得当前网络连接的通道的状态

3. 通过Channel 可获得 网络连接的配置参数 （例如接收缓冲区大小）

4. Channel 提供异步的网络 I/O 操作(如建立连接，读写，绑定端口)，异步调用意味着任何 I/O 调用都将立即返回，并且不保证在调用结束时所请求的 I/O 操作已完成

5. 调用立即返回一个 ChannelFuture 实例，通过注册监听器到 ChannelFuture 上，可以 I/O 操作成功、失败或取消时回调通知调用方

6. 支持关联 I/O 操作与对应的处理程序

7. 不同协议、不同的阻塞类型的连接都有不同的 Channel 类型与之对应，常用的 Channel 类型:

NioSocketChannel，异步的客户端 TCP Socket 连接。
NioServerSocketChannel，异步的服务器端 TCP Socket 连接。
NioDatagramChannel，异步的 UDP 连接。
NioSctpChannel，异步的客户端 Sctp 连接。
NioSctpServerChannel，异步的 Sctp 服务器端连接，这些通道涵盖了 UDP 和 TCP 网络 IO 以及文件 IO。

1.4 Selector

1. Netty 基于 Selector 对象实现 I/O 多路复用，通过 Selector 一个线程可以监听多个连接的 Channel 事件。

2. 当向一个 Selector 中注册 Channel 后，Selector 内部的机制就可以自动不断地查询(Select) 这些注册的 Channel 是否有已就绪的 I/O 事件（例如可读，可写，网络连接完成等），这样程序就可以很简单地使用一个线程高效地管理多个 Channel

1.5 ChannelHandler及其实现类

1. ChannelHandler 是一个接口，处理 I/O 事件或拦截 I/O 操作，并将其转发到其 ChannelPipeline(业务处理链)中的下一个处理程序。

2. ChannelHandler 本身并没有提供很多方法，因为这个接口有许多的方法需要实现，方便使用期间，可以继承它的子类

3. ChannelHandler及其实现类一览图

• ChannelInboundHandler 用于处理入站 I/O 事件。

• ChannelOutboundHandler 用于处理出站 I/O 操作。

​ //适配器

• ChannelInboundHandlerAdapter 用于处理入站 I/O 事件。

• ChannelOutboundHandlerAdapter 用于处理出站 I/O 操作。

• ChannelDuplexHandler 用于处理入站和出站事件。

4. 我们经常需要自定义一个 Handler 类去继承 ChannelInboundHandlerAdapter，然后通过重写相应方法实现业务逻辑.

1.6 Pipeline 和ChannelPipeline

image-20200331193956564.png

ChannelPipeline 是一个重点：

1. ChannelPipeline 是一个Handler 的集合，它负责处理和拦截inbound 或者outbound 的事件和操作，相当于
   一个贯穿Netty 的链。(也可以这样理解：ChannelPipeline 是保存ChannelHandler 的List，用于处理或拦截
   Channel 的入站事件和出站操作)
2. ChannelPipeline 实现了一种高级形式的拦截过滤器模式，使用户可以完全控制事件的处理方式，以及Channel
   中各个的ChannelHandler 如何相互交互
3. 在Netty 中每个Channel 都有且仅有一个ChannelPipeline 与之对应，它们的组成关系如下

一个 Channel 包含了一个 ChannelPipeline，而 ChannelPipeline 中又维护了一个由 ChannelHandlerContext 组成的双向链表，并且每个 ChannelHandlerContext 中又关联着一个 ChannelHandler

入站事件和出站事件在一个双向链表中，入站事件会从链表 head 往后传递到最后一个入站的 handler，出站事件会从链表 tail 往前传递到最前一个出站的 handler，两种类型的 handler 互不干扰

4. 常用方法

ChannelPipeline addFirst(ChannelHandler... handlers)，把一个业务处理类（handler）添加到链中的第一个位置
ChannelPipeline addLast(ChannelHandler... handlers)，把一个业务处理类（handler）添加到链中的最后一个位

1.7 ChannelHandlerContext

1. 保存 Channel 相关的所有上下文信息，同时关联一个 ChannelHandler 对象

2. 即ChannelHandlerContext 中 包 含 一 个 具 体 的 事 件 处 理 器 ChannelHandler ， 同 时ChannelHandlerContext 中也绑定了对应的 pipeline 和 Channel 的信息，方便对 ChannelHandler进行调用.

3. 常用方法

ChannelFuture close()，关闭通道
ChannelOutboundInvoker flush()，刷新
ChannelFuture writeAndFlush(Object msg) ， 将 数 据 写 到 ChannelPipeline 中 当 前
ChannelHandler 的下一个 ChannelHandler 开始处理（出站）

1.8 ChannelOption

1. Netty 在创建 Channel 实例后,一般都需要设置 ChannelOption 参数。

2. ChannelOption 参数如下:

ChannelOption.SO_BACKLOG
对应 TCP/IP 协议 listen 函数中的 backlog 参数，用来初始化服务器可连接队列大小。服
务端处理客户端连接请求是顺序处理的，所以同一时间只能处理一个客户端连接。多个客户
端来的时候，服务端将不能处理的客户端连接请求放在队列中等待处理，backlog 参数指定
了队列的大小。

ChannelOption.SO_KEEPALIVE
一直保持连接活动状态

1.9 EventLoopGroup 和其实现类NioEventLoopGroup

1. EventLoopGroup 是一组 EventLoop 的抽象，Netty 为了更好的利用多核 CPU 资源，一般会有多个 EventLoop 同时工作，每个 EventLoop 维护着一个 Selector 实例。

2. EventLoopGroup 提供 next 接口，可以从组里面按照一定规则获取其中一个 EventLoop来处理任务。在 Netty 服务器端编程中，我们一般都需要提供两个 EventLoopGroup，例如：BossEventLoopGroup 和 WorkerEventLoopGroup。

3. 通常一个服务端口即一个 ServerSocketChannel对应一个Selector 和一个EventLoop线程。BossEventLoop 负责接收客户端的连接并将 SocketChannel 交给 WorkerEventLoopGroup 来进行 IO 处理

BossEventLoopGroup 通常是一个单线程的 EventLoop，EventLoop 维护着一个注册了ServerSocketChannel 的 Selector 实例BossEventLoop 不断轮询 Selector 将连接事件分离出来

通常是 OP_ACCEPT 事件，然后将接收到的 SocketChannel 交给 WorkerEventLoopGroup
WorkerEventLoopGroup 会由 next 选择其中一个 EventLoop来将这个 SocketChannel 注册到其维护的 Selector 并对其后续的 IO 事件进行处理

4. 常用方法

public NioEventLoopGroup()，构造方法
public Future<?> shutdownGracefully()，断开连接，关闭线程

1.10 Unpooled 类

1. Netty 提供一个专门用来操作缓冲区(即Netty的数据容器)的工具类

2. 常用方法如下所示

//通过给定的数据和字符编码返回一个 ByteBuf 对象（类似于 NIO 中的 ByteBuffer 但有区别）

public static ByteBuf copiedBuffer(CharSequence string, Charset charset)

3. 举例说明Unpooled获取 Netty的数据容器ByteBuf 的基本使用

public class NettyByteBuf01 {
    public static void main(String[] args) {


        //创建一个ByteBuf
        //说明
        //1. 创建 对象，该对象包含一个数组arr , 是一个byte[10]
        //2. 在netty 的buffer中，不需要使用flip 进行反转
        //   底层维护了 readerIndex 和 writerIndex
        //3. 通过 readerIndex 和  writerIndex 和  capacity， 将buffer分成三个区域
        // 0---readerIndex 已经读取的区域
        // readerIndex---writerIndex ， 可读的区域
        // writerIndex -- capacity, 可写的区域
        ByteBuf buffer = Unpooled.buffer(10);

        for (int i = 0; i < 10; i++) {
            buffer.writeByte(i);
        }

        //10
        System.out.println("capacity=" + buffer.capacity());
        //输出
        for (int i = 0; i < buffer.capacity(); i++) {
            System.out.println(buffer.getByte(i));
        }
        for (int i = 0; i < buffer.capacity(); i++) {
            System.out.println(buffer.readByte());
        }
        System.out.println("执行完毕");
    }
}

public class NettyByteBuf02 {
    public static void main(String[] args) {

        //创建ByteBuf
        ByteBuf byteBuf = Unpooled.copiedBuffer("hello,world!", StandardCharsets.UTF_8);

        //使用相关的方法// true
        if (byteBuf.hasArray()) {

            byte[] content = byteBuf.array();

            //将 content 转成字符串
            System.out.println(new String(content, StandardCharsets.UTF_8));

            System.out.println("byteBuf=" + byteBuf);

            // 0
            System.out.println(byteBuf.arrayOffset());
            // 0
            System.out.println(byteBuf.readerIndex());
            // 12
            System.out.println(byteBuf.writerIndex());
            // 36
            System.out.println(byteBuf.capacity());

            //System.out.println(byteBuf.readByte());
            //104
            System.out.println(byteBuf.getByte(0));

            //可读的字节数  12
            int len = byteBuf.readableBytes();
            System.out.println("len=" + len);

            //使用for取出各个字节
            for (int i = 0; i < len; i++) {
                System.out.println((char) byteBuf.getByte(i));
            }

            //按照某个范围读取
            System.out.println(byteBuf.getCharSequence(0, 4, StandardCharsets.UTF_8));
            System.out.println(byteBuf.getCharSequence(4, 6, StandardCharsets.UTF_8));
        }
    }
}

1.11 Netty 应用实例-群聊系统

public class GroupChatServer {

    private int port;

    public GroupChatServer(int port) {
        this.port = port;
    }

    /**
     * 编写run方法，处理客户端的请求
     */
    public void run() throws Exception {

        //创建两个线程组
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        //8个NioEventLoop
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap b = new ServerBootstrap();

            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 128)
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    .childHandler(new ChannelInitializer<SocketChannel>() {

                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {

                            //获取到pipeline
                            ChannelPipeline pipeline = ch.pipeline();
                            //向pipeline加入解码器
                            pipeline.addLast("decoder", new StringDecoder());
                            //向pipeline加入编码器
                            pipeline.addLast("encoder", new StringEncoder());
                            //加入自己的业务处理handler
                            pipeline.addLast(new GroupChatServerHandler());

                        }
                    });

            System.out.println("netty 服务器启动");
            ChannelFuture channelFuture = b.bind(port).sync();

            //监听关闭
            channelFuture.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }

    }

    public static void main(String[] args) throws Exception {
        new GroupChatServer(7000).run();
    }
}

public class GroupChatServerHandler extends SimpleChannelInboundHandler<String> {

    //public static List<Channel> channels = new ArrayList<Channel>();

    //使用一个hashMap 管理
    //public static Map<String, Channel> channels = new HashMap<String,Channel>();

    /**
     * 定义一个channel 组，管理所有的channel
     * GlobalEventExecutor.INSTANCE) 是全局的事件执行器，是一个单例
     */
    private static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    public static final DateTimeFormatter DATE_TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSS");

    /**
     * handlerAdded 表示连接建立，一旦连接，第一个被执行
     * 将当前channel 加入到  channelGroup
     */
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        //将该客户加入聊天的信息推送给其它在线的客户端
        //该方法会将 channelGroup 中所有的channel 遍历，并发送 消息，我们不需要自己遍历
        channelGroup.writeAndFlush("[客户端]" + channel.remoteAddress() + " 加入聊天" + DATE_TIME_FORMATTER.format(LocalDateTime.now()) + " \n");
        channelGroup.add(channel);
    }

    /**
     * 断开连接, 将xx客户离开信息推送给当前在线的客户
     */
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        channelGroup.writeAndFlush("[客户端]" + channel.remoteAddress() + " 离开了\n");
        System.out.println("channelGroup size" + channelGroup.size());
    }

    /**
     * 表示channel 处于活动状态, 提示 xx上线
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println(ctx.channel().remoteAddress() + " 上线了~");
    }

    /**
     * 表示channel 处于不活动状态, 提示 xx离线了
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println(ctx.channel().remoteAddress() + " 离线了~");
    }

    /**
     * 读取数据
     */
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        //获取到当前channel
        Channel channel = ctx.channel();
        //这时我们遍历channelGroup, 根据不同的情况，回送不同的消息
        channelGroup.forEach(ch -> {
            //不是当前的channel,转发消息
            if (channel != ch) {
                ch.writeAndFlush("[客户]" + channel.remoteAddress() + " 发送了消息" + msg + "\n");
            } else {
                //回显自己发送的消息给自己
                ch.writeAndFlush("[自己]发送了消息" + msg + "\n");
            }
        });
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        //关闭通道
        ctx.close();
    }
}

public class GroupChatClient {

    private final String host;
    private final int port;

    public GroupChatClient(String host, int port) {
        this.host = host;
        this.port = port;
    }

    public void run() throws Exception {
        EventLoopGroup group = new NioEventLoopGroup();

        try {
            Bootstrap bootstrap = new Bootstrap()
                    .group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {

                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            //得到pipeline
                            ChannelPipeline pipeline = ch.pipeline();
                            //加入相关handler
                            pipeline.addLast("decoder", new StringDecoder());
                            pipeline.addLast("encoder", new StringEncoder());
                            //加入自定义的handler
                            pipeline.addLast(new GroupChatClientHandler());
                        }
                    });

            ChannelFuture channelFuture = bootstrap.connect(host, port).sync();
            //得到channel
            Channel channel = channelFuture.channel();
            System.out.println("-------" + channel.localAddress() + "--------");
            //客户端需要输入信息，创建一个扫描器
            Scanner scanner = new Scanner(System.in);
            while (scanner.hasNextLine()) {
                String msg = scanner.nextLine();
                //通过channel 发送到服务器端
                channel.writeAndFlush(msg + "\r\n");
            }
        } finally {
            group.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws Exception {
        new GroupChatClient("127.0.0.1", 7000).run();
    }
}

public class GroupChatClientHandler extends SimpleChannelInboundHandler<String> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        System.out.println(msg.trim());
    }
}

1.12 Netty 心跳检测机制案例

定时断线重连

客户端断线重连机制。

客户端数量多，且需要传递的数据量级较大。可以周期性的发送数据的时候，使用。要求对数据的即时性不高的时候，才可使用。

优点： 可以使用数据缓存。不是每条数据进行一次数据交互。可以定时回收资源，对资源利用率高。相对来说，即时性可以通过其他方式保证。如： 120秒自动断线。数据变化1000次请求服务器一次。300秒中自动发送不足1000次的变化数据。

对于长连接的程序断网重连几乎是程序的标配。断网重连具体可以分为两类:

CONNECT失败，需要重连；

实现ChannelFutureListener 用来启动时监测是否连接成功，不成功的话重试

实现ChannelFutureListener 用来启动时监测是否连接成功，不成功的话重试
private void doConnect() {
    Bootstrap b = ...;
    b.connect().addListener((ChannelFuture f) -> {
        if (!f.isSuccess()) {
            long nextRetryDelay = nextRetryDelay(...);
            f.channel().eventLoop().schedule(nextRetryDelay, ..., () -> {
                doConnect();
            }); // or you can give up at some point by just doing nothing.
        }
    });
}
或者
public class ConnectionListener implements ChannelFutureListener {
  private Client client;
  public ConnectionListener(Client client) {
    this.client = client;
  }
  @Override
  public void operationComplete(ChannelFuture channelFuture) throws Exception {
    if (!channelFuture.isSuccess()) {
      System.out.println("Reconnect");
      //因为是建立网络连接所以可以共用EventLoop
      final EventLoop loop = channelFuture.channel().eventLoop();
      loop.schedule(new Runnable() {
        @Override
        public void run() {
          client.createBootstrap(new Bootstrap(), loop);
        }
      }, 1L, TimeUnit.SECONDS);
    }
  }
}

程序运行过程中断网、远程强制关闭连接、收到错误包必须重连；

public Bootstrap createBootstrap(Bootstrap bootstrap, EventLoopGroup eventLoop) {
     if (bootstrap != null) {
       final MyInboundHandler handler = new MyInboundHandler(this);
       bootstrap.group(eventLoop);
       bootstrap.channel(NioSocketChannel.class);
       bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
       bootstrap.handler(new ChannelInitializer<SocketChannel>() {
         @Override
         protected void initChannel(SocketChannel socketChannel) throws Exception {
           socketChannel.pipeline().addLast(handler);
         }
       });
       bootstrap.remoteAddress("localhost", 8888);
       bootstrap.connect().addListener(new ConnectionListener(this));
     }
     return bootstrap;
   }

public class MyInboundHandler extends SimpleChannelInboundHandler {
   private Client client;
   public MyInboundHandler(Client client) {
     this.client = client;
   }
   @Override
   public void channelInactive(ChannelHandlerContext ctx) throws Exception {
     final EventLoop eventLoop = ctx.channel().eventLoop();
     eventLoop.schedule(new Runnable() {
       @Override
       public void run() {
         client.createBootstrap(new Bootstrap(), eventLoop);
       }
     }, 1L, TimeUnit.SECONDS);
     super.channelInactive(ctx);
   }
 }

心跳监测

使用定时发送消息的方式，实现硬件检测，达到心态检测的目的。

心跳监测是用于检测电脑硬件和软件信息的一种技术。如：CPU使用率，磁盘使用率，内存使用率，进程情况，线程情况等。

Netty提供的心跳检测机制分为三种：

• 读空闲，链路持续时间t没有读取到任何消息；
• 写空闲，链路持续时间t没有发送任何消息；
• 读写空闲，链路持续时间t没有接收或者发送任何消息。

心跳检测机制分为三个层面：

• TCP层面的心跳检测，即TCP的Keep-Alive机制，它的作用域是整个TCP协议栈；
• 协议层的心跳检测，主要存在于长连接协议中。例如SMPP协议；
• 应用层的心跳检测，它主要由各业务产品通过约定方式定时给对方发送心跳消息实现。

public class MyServer {
    public static void main(String[] args) throws Exception {
        //创建两个线程组
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        //8个NioEventLoop
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {

            ServerBootstrap serverBootstrap = new ServerBootstrap();

            serverBootstrap.group(bossGroup, workerGroup);
            serverBootstrap.channel(NioServerSocketChannel.class);
            serverBootstrap.handler(new LoggingHandler(LogLevel.INFO));
            serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {

                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    ChannelPipeline pipeline = ch.pipeline();
                    //加入一个netty 提供 IdleStateHandler
//                    说明
//                    1. IdleStateHandler 是netty 提供的处理空闲状态的处理器
//                    2. long readerIdleTime : 表示多长时间没有读, 就会发送一个心跳检测包检测是否连接
//                    3. long writerIdleTime : 表示多长时间没有写, 就会发送一个心跳检测包检测是否连接
//                    4. long allIdleTime : 表示多长时间没有读写, 就会发送一个心跳检测包检测是否连接
//
//                    5. 文档说明
//                    triggers an {@link IdleStateEvent} when a {@link Channel} has not performed
// * read, write, or both operation for a while.
// *                  6. 当 IdleStateEvent 触发后 , 就会传递给管道 的下一个handler去处理
// *                  通过调用(触发)下一个handler 的 userEventTiggered , 在该方法中去处理 IdleStateEvent(读空闲，写空闲，读写空闲)
                    pipeline.addLast(new IdleStateHandler(7000, 7000, 10, TimeUnit.SECONDS));
                    //加入一个对空闲检测进一步处理的handler(自定义)
                    pipeline.addLast(new MyServerHandler());
                }
            });

            //启动服务器
            ChannelFuture channelFuture = serverBootstrap.bind(7000).sync();
            channelFuture.channel().closeFuture().sync();

        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

public class MyServerHandler extends ChannelInboundHandlerAdapter {

    /**
     * @param ctx 上下文
     * @param evt 事件
     * @throws Exception
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {

        if (evt instanceof IdleStateEvent) {

            //将  evt 向下转型 IdleStateEvent
            IdleStateEvent event = (IdleStateEvent) evt;
            String eventType = null;
            switch (event.state()) {
                case READER_IDLE:
                    eventType = "读空闲";
                    break;
                case WRITER_IDLE:
                    eventType = "写空闲";
                    break;
                case ALL_IDLE:
                    eventType = "读写空闲";
                    break;
                default:
                    throw new IllegalStateException("Unexpected value: " + event.state());
            }
            System.out.println(ctx.channel().remoteAddress() + "--超时时间--" + eventType);
            System.out.println("服务器做相应处理..");

            //如果发生空闲，我们关闭通道
            // ctx.channel().close();
        }
    }
}

客户端和服务端之间连接断开机制

TCP连接的建立需要三个分节(三次握手)，终止则需要四个分节。

1. 某个应用进程首先调用close，称该端执行主动关闭(active close)。该端的TCP于是发送一个FIN分节，表示数据发送完毕。
2. 接收到这个FIN分节的对端执行被动关闭(passive close) 。这个FIN由TCP确认。他的接收也作为一个文件结束符传递给接收端应用程序，因为FIN的接收意味着接收端应用程序在相应连接上再无额外数据可以收取;
3. 一段时间后，接收到这个文件结束符的应用程序调用close管理他的套接字。这导致他的TCP也发送一个FIN。
4. 接收这个最终FIN的原发送端TCP(执行主动关闭的那一端)确认这个FIN。

既然每个方向都需要一个FIN和一个ACK，因此通常需要4个分节。但是某些情况下步骤1的FIN随数据一起发送；另外，步骤2和步骤3发送的分节都处在执行被动关闭的那一端，有可能合并成一个分节发送。

image

TCP关闭连接时对应的状态图如下:

image

对于大量短连接的情况下，经常出现卡在FIN_WAIT2和TIMEWAIT状态的连接，等待系统回收，但是操作系统底层回收的时间频率很长，导致SOCKET被耗尽。解决方案如下:

Linux平台:

#!/bin/sh

echo "Now,config system parameters..."

echo "#config for MWGATE" >> /etc/sysctl.conf
#当出现SYN等待队列溢出时，启用cookies来处理，可防范少量SYN攻击
echo 'net.ipv4.tcp_syncookies = 1' >> /etc/sysctl.conf
#允许将TIME-WAIT sockets重新用于新的TCP连接，默认为0，表示关闭；
echo 'net.ipv4.tcp_tw_reuse = 1' >> /etc/sysctl.conf
#TIM_WAIT_2也就是FIN_SYN_2的等待时间
echo 'net.ipv4.tcp_fin_timeout = 30' >> /etc/sysctl.conf
#表示开启TCP连接中TIME-WAIT sockets的快速回收，默认为0，表示关闭。
echo 'net.ipv4.tcp_tw_recycle = 1' >>/etc/sysctl.conf
FILEMAX=$(cat /proc/sys/fs/file-max)
PORTRANGE="net.ipv4.ip_local_port_range = 1024 $FILEMAX"
echo $PORTRANGE >> /etc/sysctl.conf
echo "#end config for MWGATE" >> /etc/sysctl.conf

echo '#config for MWGATE' >> /etc/security/limits.conf
SOFTLIMIT="* soft nofile $FILEMAX"
HARDLIMIT="* hard nofile $FILEMAX"
echo "$SOFTLIMIT" >> /etc/security/limits.conf
echo "$HARDLIMIT" >> /etc/security/limits.conf
echo '#end config for MWGATE' >> /etc/security/limits.conf

echo "#config for MWGATE" >> /etc/pam.d/login
echo 'session required /lib/security/pam_limits.so' >> /etc/pam.d/login
echo "#end config for MWGATE" >> /etc/pam.d/login

sysctl -p

echo "#config for MWGATE" >> /etc/profile
echo 'ulimit -S -c unlimited > /dev/null 2>&1' >> /etc/profile
echo "#end config for MWGATE" >> /etc/profile

echo 'core-%p-%t' >> /proc/sys/kernel/core_pattern

source /etc/profile

echo "Config is OK..."

Windows平台:

Windows Registry Editor Version 5.00
[HKEY_LOCAL_MACHINE\SYSTEM\CurrentControlSet\Services\Tcpip\Parameters]
"MaxUserPort"=dword:0000fffe
"TCPTimedWaitDelay"=dword:00000005

TCP状态图

TCP状态转化图

TCP/IP半关闭

从上述讲的TCP关闭的四个分节可以看出，被动关闭执行方，发送FIN分节的前提是TCP套接字对应应用程序调用close产生的。如果服务端有数据发送给客户端那么可能存在服务端在接受到FIN之后，需要将数据发送到客户端才能发送FIN字节。这种处于业务考虑的情形通常称为半关闭。

半关闭可能导致大量socket处于CLOSE_WAIT状态

谁负责关闭连接合理

连接关闭触发的条件通常分为如下几种:

1. 数据发送完成(发送到对端并且收到响应)，关闭连接；
2. 通信过程中产生异常；
3. 特殊指令强制要求关闭连接；

对于第一种，通常关闭时机是，数据发送完成方发起(客户端触发居多)；
对于第二种，异常产生方触发(例如残包、错误数据等)发起。但是此种情况可能也导致压根无法发送FIN。
对于第三种，通常是用于运维等。由命令发起方产生。

1.13 Netty 通过WebSocket 编程实现服务器和客户端长连接

public class MyServer {
    public static void main(String[] args) throws Exception {

        //创建两个线程组
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        //8个NioEventLoop
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {

            ServerBootstrap serverBootstrap = new ServerBootstrap();

            serverBootstrap.group(bossGroup, workerGroup);
            serverBootstrap.channel(NioServerSocketChannel.class);
            serverBootstrap.handler(new LoggingHandler(LogLevel.INFO));
            serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {

                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    ChannelPipeline pipeline = ch.pipeline();

                    //因为基于http协议，使用http的编码和解码器
                    pipeline.addLast(new HttpServerCodec());
                    //是以块方式写，添加ChunkedWriteHandler处理器
                    pipeline.addLast(new ChunkedWriteHandler());

                    /*
                    说明
                    1. http数据在传输过程中是分段, HttpObjectAggregator ，就是可以将多个段聚合
                    2. 这就就是为什么，当浏览器发送大量数据时，就会发出多次http请求
                     */
                    pipeline.addLast(new HttpObjectAggregator(8192));
                    /*
                    说明
                    1. 对应websocket ，它的数据是以 帧(frame) 形式传递
                    2. 可以看到WebSocketFrame 下面有六个子类
                    3. 浏览器请求时 ws://localhost:7000/hello 表示请求的uri
                    4. WebSocketServerProtocolHandler 核心功能是将 http协议升级为 ws协议 , 保持长连接
                    5. 是通过一个 状态码 101
                     */
                    pipeline.addLast(new WebSocketServerProtocolHandler("/hello2"));

                    //自定义的handler ，处理业务逻辑
                    pipeline.addLast(new MyTextWebSocketFrameHandler());
                }
            });

            //启动服务器
            ChannelFuture channelFuture = serverBootstrap.bind(7000).sync();
            channelFuture.channel().closeFuture().sync();

        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

/**
 * 这里 TextWebSocketFrame 类型，表示一个文本帧(frame)
 *
 * @author Administrator
 */
public class MyTextWebSocketFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
        System.out.println("服务器收到消息 " + msg.text());
        //回复消息
        ctx.channel().writeAndFlush(new TextWebSocketFrame("服务器时间" + LocalDateTime.now() + " " + msg.text()));
    }

    /**
     * 当web客户端连接后， 触发方法
     */
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        //id 表示唯一的值，LongText 是唯一的 ShortText 不是唯一
        System.out.println("handlerAdded 被调用" + ctx.channel().id().asLongText());
        System.out.println("handlerAdded 被调用" + ctx.channel().id().asShortText());
    }


    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        System.out.println("handlerRemoved 被调用" + ctx.channel().id().asLongText());
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println("异常发生 " + cause.getMessage());
        //关闭连接
        ctx.close();
    }
}

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>
</head>
<body>
<script>
    var socket;
    //判断当前浏览器是否支持websocket
    if(window.WebSocket) {
        //go on
        socket = new WebSocket("ws://localhost:7000/hello2");
        //相当于channelReado, ev 收到服务器端回送的消息
        socket.onmessage = function (ev) {
            var rt = document.getElementById("responseText");
            rt.value = rt.value + "\n" + ev.data;
        }

        //相当于连接开启(感知到连接开启)
        socket.onopen = function (ev) {
            var rt = document.getElementById("responseText");
            rt.value = "连接开启了.."
        }

        //相当于连接关闭(感知到连接关闭)
        socket.onclose = function (ev) {

            var rt = document.getElementById("responseText");
            rt.value = rt.value + "\n" + "连接关闭了.."
        }
    } else {
        alert("当前浏览器不支持websocket")
    }

    //发送消息到服务器
    function send(message) {
        if(!window.socket) { //先判断socket是否创建好
            return;
        }
        if(socket.readyState == WebSocket.OPEN) {
            //通过socket 发送消息
            socket.send(message)
        } else {
            alert("连接没有开启");
        }
    }
</script>
    <form onsubmit="return false">
        <textarea name="message" style="height: 300px; width: 300px"></textarea>
        <input type="button" value="发生消息" onclick="send(this.form.message.value)">
        <textarea id="responseText" style="height: 300px; width: 300px"></textarea>
        <input type="button" value="清空内容" onclick="document.getElementById('responseText').value=''">
    </form>
</body>
</html>