『Netty核心』Netty断线重连

发表于

「这是我参与11月更文挑战的第29天，活动详情查看：2021最后一次更文挑战」。

点赞再看，养成习惯👏👏

一、前言

由于在通信层的网络连接的不可靠性，比如：网络闪断，网络抖动等，经常会出现连接断开。这样对于使用长连接的应用而言，当突然高流量冲击势必会造成进行网络连接，从而产生网络堵塞，应用响应速度下降，延迟上升，用户体验较差。

在通信层的高可用设计中，需要保活长连接的网络，保证通信能够正常。一般有两种设计方式：

  1. 利用TCP提供的连接保活特性
  2. 应用层做连接保活

二、TCP连接保活性的局限

TCP协议层面提供了KeepAlive的机制保证连接的活跃，但是其有很多劣势：

  • 该保活机制非TCP协议的标准，默认是关闭
  • 该机制依赖操作系统，需要进行系统级配置，不够灵活方便
  • 当应用底层传输协议变更时，将无法适用

由于以上的原因，绝大多数的框架、应用处理连接的保活性都是在应用层处理。目前的主流方案是心跳检测，断线重连。

在上一篇文章《『Netty核心』Netty心跳机制》已经介绍Netty是如何实现心跳检测机制的。

三、断线重连

断线重连是指由于网络波动造成用户间歇性的断开与服务器的连接，待网络恢复之后服务器尝试将用户连接到上次断开时的状态和数据。

当心跳检测发现连接断开后，为了保证通信层的可用性，仍然需要重新连接，保证通信的可靠。对于短线重连一般有两种设计方式比较常见：

  1. 通过额外的线程定时轮循所有的连接的活跃性，如果发现其中有死连接，则执行重连
  2. 监听连接上发送的断开事件，如果发送则执行重连操作

四、Netty断线自动重连实现

  1. 客户端启动连接服务端时，如果网络或服务端有问题，客户端连接失败，可以重连，重连的逻辑加在客户端。
  2. 系统运行过程中网络故障或服务端故障，导致客户端与服务端断开连接了也需要重连，可以在客户端处理数据的Handler的 channelInactive 方法中进行重连。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
java复制代码public class NettyClient {

    private String host;
    private int port;
    private Bootstrap bootstrap;
    private EventLoopGroup group;

    public static void main(String[] args) throws Exception {
        NettyClient nettyClient = new NettyClient("localhost", 9000);
        nettyClient.connect();
    }

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

    private void init() {
        //客户端需要一个事件循环组
        group = new NioEventLoopGroup();
        //创建客户端启动对象
        // bootstrap 可重用, 只需在NettyClient实例化的时候初始化即可.
        bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        //加入处理器
                        ch.pipeline().addLast(new NettyClientHandler(NettyClient.this));
                    }
                });
    }

    public void connect() throws Exception {
        System.out.println("netty client start。。");
        //启动客户端去连接服务器端
        ChannelFuture cf = bootstrap.connect(host, port);
        cf.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (!future.isSuccess()) {
                    //重连交给后端线程执行
                    future.channel().eventLoop().schedule(() -> {
                        System.err.println("重连服务端...");
                        try {
                            connect();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }, 3000, TimeUnit.MILLISECONDS);
                } else {
                    System.out.println("服务端连接成功...");
                }
            }
        });
        //对通道关闭进行监听
        cf.channel().closeFuture().sync();
    }
}

实现自定义的ChannelInboundHandlerAdapter

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
java复制代码public class NettyClientHandler extends ChannelInboundHandlerAdapter {
    private NettyClient nettyClient;

    public NettyClientHandler(NettyClient nettyClient) {
        this.nettyClient = nettyClient;
    }

    /**
     * 当客户端连接服务器完成就会触发该方法
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ByteBuf buf = Unpooled.copiedBuffer("HelloServer".getBytes(CharsetUtil.UTF_8));
        ctx.writeAndFlush(buf);
    }

    //当通道有读取事件时会触发，即服务端发送数据给客户端
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf buf = (ByteBuf) msg;
        System.out.println("收到服务端的消息:" + buf.toString(CharsetUtil.UTF_8));
        System.out.println("服务端的地址： " + ctx.channel().remoteAddress());
    }

    // channel 处于不活动状态时调用
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.err.println("运行中断开重连。。。");
        nettyClient.connect();
    }

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

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

0%