2.示例实现

2.1 server

package com.zy.netty.netty06;

import com.zy.netty.common.Constants;
import io.netty.bootstrap.ServerBootstrap;
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.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.DefaultThreadFactory;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

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

@Slf4j
@Getter
public class Server06 {
    private final ServerBootstrap server;
    private final NioEventLoopGroup acceptorGroup;
    private final NioEventLoopGroup ioGroup;
    private final int acceptorThreads;
    private final int ioThreads;
    private final String host;
    private final int port;

    public Server06(int acceptorThreads, int ioThreads, String host, int port) {
        this.acceptorThreads = acceptorThreads;
        this.ioThreads = ioThreads;
        this.host = host;
        this.port = port;
        this.server = new ServerBootstrap();
        // FIXME 为了调试, 这里不要通过线程工厂设置 daemon 线程, 如果后续 client 运行在容器(如 tomcat) 中, 这里可以设置为 daemon
        // FIXME 说明下, 如果这里全设置为 daemon, 而下述 bind 端口后又未阻塞主线程, 则所有 daemon 将被销毁, 进行停止. 关于 daemon 线程, 可自行研究
        this.acceptorGroup = new NioEventLoopGroup(acceptorThreads, new DefaultThreadFactory("server-acceptor-group", false));
        this.ioGroup = new NioEventLoopGroup(ioThreads, new DefaultThreadFactory("server-io-group", true));
        this.start();
    }

    public Server06(String host, int port) {
        this(1, Runtime.getRuntime().availableProcessors(), host, port);
    }

    private void start() {
        try {
            server.group(acceptorGroup, ioGroup)
                    .channel(NioServerSocketChannel.class)
                    .childOption(ChannelOption.TCP_NODELAY, true)
                    .childHandler(new ChannelInitializer<NioSocketChannel>() {
                        @Override
                        protected void initChannel(NioSocketChannel ch) throws Exception {
                            ch.pipeline()
                                    .addLast(new DelimiterBasedFrameDecoder(4096, Unpooled.copiedBuffer(Constants.CUSTOM_DELIMITER_01.getBytes(StandardCharsets.UTF_8))))
                                    .addLast(new StringDecoder())
                                    .addLast(new StringEncoder())
                                    .addLast(new IdleStateHandler(180L, 0L, 0L, TimeUnit.SECONDS))
                                    .addLast(new BusinessServerHandler());
                        }
                    });

            server.bind(host, port).syncUninterruptibly();
            // FIXME 这里一直监听 client 连接; 也可以用死循环阻塞线程如 while(true){}; 如果 boss 线程都不为 daemon, 则此处也不用再设置
            // channelFuture.channel().closeFuture().sync();
        } catch (Throwable e) {
            System.out.println("server error.");
            e.printStackTrace();
        }
    }

    public void close() {
        if (Objects.nonNull(acceptorGroup)) {
            acceptorGroup.shutdownGracefully();
        }
        if (Objects.nonNull(ioGroup)) {
            ioGroup.shutdownGracefully();
        }
    }

    @ChannelHandler.Sharable
    private class BusinessServerHandler extends SimpleChannelInboundHandler<Object> {
        private LongAdder counter = new LongAdder();
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
            counter.increment();
            System.out.println("server receive msg: " + msg + "; the counter is: " + counter.doubleValue());
            // FIXME 这里是 ctx.channel()?
            ctx.writeAndFlush(LocalDateTime.now().toString() + Constants.CUSTOM_DELIMITER_01);
        }

        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            if (evt instanceof IdleStateEvent) {
                IdleStateEvent event = (IdleStateEvent) evt;
                if (event == IdleStateEvent.READER_IDLE_STATE_EVENT) {
                    System.out.println(String.format("there is a long time to receive msg from client [%s]", ctx.channel().remoteAddress().toString()));
                    // FIXME 这里是用 ctx.close() 还是 ctx.disconnect() ? 二者区别?
                    ctx.close();
                }
            } else {
                super.userEventTriggered(ctx, evt);
            }
        }
    }
}

2.2 client

package com.zy.netty.netty06;

import com.zy.netty.common.Constants;
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.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;
import io.netty.util.concurrent.DefaultThreadFactory;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.StandardCharsets;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.LongAdder;

/**
 * https://www.cnblogs.com/MrRightZhao/p/11967982.html
 * <p>
 * https://www.jianshu.com/p/7052761fa480
 */
@Slf4j
@Getter
public class Client06 {
    private final Bootstrap client;
    private final NioEventLoopGroup nioEventLoopGroup;
    private final int nioThreads;
    private final String host;
    private final int port;
    private final int connectTimeoutMills;
    private final int reconnectTimeoutMills;
    // volatile, please copy reference to use
    private volatile Channel channel;

    public Client06(int nioThreads, String host, int port, int connectTimeoutMills, int reconnectTimeoutMills) {
        this.nioThreads = nioThreads;
        this.host = host;
        this.port = port;
        this.connectTimeoutMills = connectTimeoutMills;
        this.reconnectTimeoutMills = reconnectTimeoutMills;
        this.client = new Bootstrap();
        // FIXME 为了调试, 这里不要通过线程工厂设置 daemon 线程, 如果后续 client 运行在容器(如 tomcat) 中, 这里可以设置为 daemon
        // FIXME 说明下, 如果这里全设置为 daemon, 而下述 bind 端口后又未阻塞主线程, 则所有 daemon 将被销毁, 进行停止. 关于 daemon 线程, 可自行研究
        this.nioEventLoopGroup = new NioEventLoopGroup(nioThreads, new DefaultThreadFactory("client-group", false));
        this.init();
    }

    public Client06(String host, int port) {
        this(Runtime.getRuntime().availableProcessors(), host, port, Constants.CONNECT_TIMEOUT_MILLS, Constants.RECONNECT_TIMEOUT_MILLS);
    }

    private void init() {
        try {
            final WatchdogClientHandler watchdogClientHandler = new WatchdogClientHandler(reconnectTimeoutMills);
            client.group(nioEventLoopGroup)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .option(ChannelOption.SO_KEEPALIVE, true)
                    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeoutMills)
                    .handler(new ChannelInitializer<NioSocketChannel>() {
                        @Override
                        protected void initChannel(NioSocketChannel ch) throws Exception {
                            ch.pipeline()
                                    .addLast(new DelimiterBasedFrameDecoder(4096, Unpooled.copiedBuffer(Constants.CUSTOM_DELIMITER_01.getBytes(StandardCharsets.UTF_8))))
                                    .addLast(new StringDecoder())
                                    .addLast(new StringEncoder())
                                    .addLast(new BusinessClientHandler())
                                    .addLast(new IdleStateHandler(0L, 90L, 0L, TimeUnit.SECONDS))
                                    .addLast(watchdogClientHandler);
                        }
                    });

            doConnect();
            // FIXME 这行代码的意思是, 当监听到 channel 被关闭时, 关闭自身的连接, 仅做调试用, 不可用于生产
            // channelFuture.channel().closeFuture().sync();
        } catch (Throwable e) {
            System.out.println("client error");
            e.printStackTrace();
        }
    }

    public void doConnect() {
        synchronized (client) {
            try {
                ChannelFuture channelFuture = client.connect(host, port);
                channelFuture.addListener((ChannelFutureListener) future -> {
                    if (!future.isSuccess()) {
                        future.channel().pipeline().fireChannelInactive();
                    } else {
                        Channel newChannel = channelFuture.channel();
                        // copy volatile reference
                        Channel oldChannel = Client06.this.channel;
                        if (Objects.nonNull(oldChannel)) {
                            oldChannel.close();
                        }
                        Client06.this.channel = newChannel;
                    }
                });
            } catch (Exception e) {
                System.out.println("client error");
                e.printStackTrace();
            }
        }
    }

    public Channel getChannel() {
        Channel ch = channel;
        if (Objects.isNull(ch) || !ch.isActive()) {
            return null;
        }
        return ch;
    }

    public void close() {
        if (Objects.nonNull(nioEventLoopGroup)) {
            nioEventLoopGroup.shutdownGracefully();
        }
    }

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

        @Override
        protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
            counter.increment();
            System.out.println("client receive msg: " + msg + "; the counter is: " + counter.doubleValue());
        }
    }

    /**
     * 请一定要加上次注解, 关于该注解的作用, 在下文已讲过
     * https://www.jianshu.com/p/051e48410a9d
     */
    @ChannelHandler.Sharable
    private class WatchdogClientHandler extends ChannelInboundHandlerAdapter implements TimerTask {
        private HashedWheelTimer timer;
        private final int timeInternalMills;
        private int retryTimes;

        WatchdogClientHandler(int timeInternalMills) {
            this.timer = new HashedWheelTimer();
            this.timeInternalMills = timeInternalMills;
            this.retryTimes = 0;
        }

        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            if (evt instanceof IdleStateEvent) {
                IdleStateEvent event = (IdleStateEvent) evt;
                if (event == IdleStateEvent.WRITER_IDLE_STATE_EVENT) {
                    // FIXME 注释掉这里, 测试服务端 close channel 后, 是否重连
                    ctx.writeAndFlush("ping" + Constants.CUSTOM_DELIMITER_01);
                }
            } else {
                super.userEventTriggered(ctx, evt);
            }
        }

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            System.out.println("连接成功 -- 重置 retryTimes=0");
            retryTimes = 0;
            ctx.fireChannelActive();
        }

        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            retryTimes++;
            System.out.println("第" + retryTimes + "次连接失效了");
            // 这里可以根据 retryTimes 大小判断是否无限重连
            timer.newTimeout(this, timeInternalMills, TimeUnit.MILLISECONDS);
            ctx.fireChannelInactive();
        }

        @Override
        public void run(Timeout timeout) throws Exception {
            System.out.println("开始重连");
            doConnect();
        }
    }
}

2.3 common

package com.zy.netty.common;

public interface Constants {
    String CUSTOM_DELIMITER_01 = "$_";
    Integer CONNECT_TIMEOUT_MILLS = 3000;
    Integer RECONNECT_TIMEOUT_MILLS = 180;
}

服务端读空闲, 主动关闭了客户端连接.png 服务端异常关闭,客户端重连.png