Netty自动重连机制的Demo

jiezi

5 年前

本博客猫叔的博客，转载请申明出处

阅读本文约“4 分钟”

适读人群：Java-Netty 初级

版本：netty 4.1.*

申明：本文旨在重新分享讨论 Netty 官方相关案例，添加部分个人理解与要点解析。

这个是 InChat 的案例地址，里面补充了详细的注释，比起官方会容易看一点。

官方案例地址：https://netty.io/4.1/xref/io/…

UptimeClient（客户端）
UptimeClientHandler
UptimeServer（服务端）
UptimeServerHandler

IdleStateHandler https://netty.io/4.1/api/io/n…

一个对 Channel 尚未执行读、写或两次操作的触发器

属性	含义
readerIdleTime	在 IdleStateEvent 其状态 IdleState.READER_IDLE 时的指定时间段没有执行读操作将被触发。指定 0 禁用。
writerIdleTime	在 IdleStateEvent 其状态 IdleState.WRITER_IDLE 时的指定时间段没有执行写操作将被触发。指定 0 禁用。
allIdleTime	一个 IdleStateEvent 其状态 IdleState.ALL_IDLE 时的时间在规定的时间进行读取和写入都将被触发。指定 0 禁用。

如下一个在没有信息时发送 ping 消息，且 30 秒没有入站信息则关闭连接

public class MyChannelInitializer extends ChannelInitializer<Channel> {
      @Override
     public void initChannel(Channel channel) {channel.pipeline().addLast("idleStateHandler", new IdleStateHandler(60, 30, 0));
         channel.pipeline().addLast("myHandler", new MyHandler());
     }
 }

 // Handler should handle the IdleStateEvent triggered by IdleStateHandler.
 public class MyHandler extends ChannelDuplexHandler {
      @Override
     public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {if (evt instanceof IdleStateEvent) {IdleStateEvent e = (IdleStateEvent) evt;
             if (e.state() == IdleState.READER_IDLE) {ctx.close();
             } else if (e.state() == IdleState.WRITER_IDLE) {ctx.writeAndFlush(new PingMessage());
             }
         }
     }
 }

UptimeClient

/**
 * Created by MySelf on 2019/8/27.
 */
public final class UptimeClient {static final String HOST = System.getProperty("host", "127.0.0.1");
    static final int PORT = Integer.parseInt(System.getProperty("port", "8080"));
    // 重新连接前睡眠 5 秒
    static final int RECONNECT_DELAY = Integer.parseInt(System.getProperty("reconnectDelay", "5"));
    // 当服务器在 10 秒内不发送任何内容时重新连接。private static final int READ_TIMEOUT = Integer.parseInt(System.getProperty("readTimeout", "10"));

    private static final UptimeClientHandler handler = new UptimeClientHandler();
    private static final Bootstrap bs = new Bootstrap();

    public static void main(String[] args) throws Exception {EventLoopGroup group = new NioEventLoopGroup();
        bs.group(group)
                .channel(NioSocketChannel.class)
                .remoteAddress(HOST,PORT)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(new IdleStateHandler(READ_TIMEOUT,0,0),handler);
                    }
                });
        bs.connect();}

    static void connect(){bs.connect().addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {if (future.cause() != null){
                    handler.startTime = -1;
                    handler.println("Failed to connect:" + future.cause());
                }
            }
        });
    }

}

UptimeClientHandler

/**
 * Created by MySelf on 2019/8/27.
 */
@ChannelHandler.Sharable
public class UptimeClientHandler extends SimpleChannelInboundHandler<Object> {

    long startTime = -1;

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {//Discard received data}

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {if (startTime < 0){startTime = System.currentTimeMillis();
        }
        println("Connected to:" + ctx.channel().remoteAddress());
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {println("Disconnected from:" + ctx.channel().remoteAddress());
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {if (!(evt instanceof IdleStateEvent)){return;}
        IdleStateEvent e = (IdleStateEvent)evt;
        if (e.state() == IdleState.READER_IDLE){
            // 连接正常，但是没有读信息，关闭连接
            println("Disconnecting due to no inbound traffic");
            ctx.close();}
    }

    @Override
    public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
        // 睡眠 5 秒
        println("Sleeping for:" + UptimeClient.RECONNECT_DELAY + 's');
        // 启动线程重新连接
        ctx.channel().eventLoop().schedule(new Runnable() {
            @Override
            public void run() {println("Reconnecting to:" + UptimeClient.HOST + ":" + UptimeClient.PORT);
                UptimeClient.connect();}
        },UptimeClient.RECONNECT_DELAY, TimeUnit.SECONDS);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {cause.printStackTrace();
        ctx.close();}

    void println(String msg){if (startTime < 0){System.err.format("[SERVER IS DOWN] %s%n",msg);
        } else {System.err.format("[UPTIME: %5ds] %s%n",(System.currentTimeMillis() - startTime)/1000,msg);
        }
    }
}

UptimeServer

/**
 * Created by MySelf on 2019/8/27.
 */
public final class UptimeServer {private static final int PORT = Integer.parseInt(System.getProperty("port", "8080"));
    private static final UptimeServerHandler handler = new UptimeServerHandler();

    private UptimeServer(){}

    public static void main(String[] args) throws Exception {EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup,workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(handler);
                        }
                    });
            // Bind and start to accept incoming connections.
            ChannelFuture f = b.bind(PORT).sync();

            // Wait until the server socket is closed.
            // In this example, this does not happen, but you can do that to gracefully
            // shut down your server.
            f.channel().closeFuture().sync();}finally {workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();}
    }
}

UptimeServerHandler

/**
 * Created by MySelf on 2019/8/27.
 */
@ChannelHandler.Sharable
public class UptimeServerHandler extends SimpleChannelInboundHandler<Object> {

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        // Close the connection when an exception is raised.
        cause.printStackTrace();
        ctx.close();}

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {// discard}
}

Netty自动重连机制的Demo

Netty 自动重连机制

正文

要点介绍

项目源码