Netty系列(6) Netty中的讀寫檢測&斷線重連機制

2.示例實現(xiàn)

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 為了調(diào)試, 這里不要通過線程工廠設置 daemon 線程, 如果后續(xù) 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 這里一直監(jiān)聽 client 連接; 也可以用死循環(huán)阻塞線程如 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() ? 二者區(qū)別?
                    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>
 * http://www.itdecent.cn/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 為了調(diào)試, 這里不要通過線程工廠設置 daemon 線程, 如果后續(xù) 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 這行代碼的意思是, 當監(jiān)聽到 channel 被關閉時, 關閉自身的連接, 僅做調(diào)試用, 不可用于生產(chǎn)
            // 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());
        }
    }

    /**
     * 請一定要加上次注解, 關于該注解的作用, 在下文已講過
     * http://www.itdecent.cn/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 + "次連接失效了");
            // 這里可以根據(jù) 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
最后編輯于 ：
?著作權歸作者所有,轉載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務。

相關閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容