我是靠谱客的博主 爱撒娇野狼,最近开发中收集的这篇文章主要介绍Netty 如何实现心跳机制与断线重连?,觉得挺不错的,现在分享给大家,希望可以做个参考。

概述

640?wx_fmt=jpeg

作者:sprinkle_liz

www.jianshu.com/p/1a28e48edd92


心跳机制


何为心跳


所谓心跳, 即在 TCP 长连接中, 客户端和服务器之间定期发送的一种特殊的数据包, 通知对方自己还在线, 以确保 TCP 连接的有效性.


注:心跳包还有另一个作用,经常被忽略,即:一个连接如果长时间不用,防火墙或者路由器就会断开该连接。


如何实现


核心Handler —— IdleStateHandler


Netty 中, 实现心跳机制的关键是 IdleStateHandler, 那么这个 Handler 如何使用呢? 先看下它的构造器:


public IdleStateHandler(int readerIdleTimeSeconds, int writerIdleTimeSeconds, int allIdleTimeSeconds) {    this((long)readerIdleTimeSeconds, (long)writerIdleTimeSeconds, (long)allIdleTimeSeconds, TimeUnit.SECONDS);}


这里解释下三个参数的含义:


  • readerIdleTimeSeconds: 读超时. 即当在指定的时间间隔内没有从 Channel 读取到数据时, 会触发一个 READER_IDLEIdleStateEvent 事件.

  • writerIdleTimeSeconds: 写超时. 即当在指定的时间间隔内没有数据写入到 Channel 时, 会触发一个 WRITER_IDLEIdleStateEvent 事件.

  • allIdleTimeSeconds: 读/写超时. 即当在指定的时间间隔内没有读或写操作时, 会触发一个 ALL_IDLEIdleStateEvent 事件.


注:这三个参数默认的时间单位是秒。若需要指定其他时间单位,可以使用另一个构造方法:IdleStateHandler(boolean observeOutput, long readerIdleTime, long writerIdleTime, long allIdleTime, TimeUnit unit)


在看下面的实现之前,建议先了解一下IdleStateHandler的实现原理。


下面直接上代码,需要注意的地方,会在代码中通过注释进行说明。


使用IdleStateHandler实现心跳


下面将使用IdleStateHandler来实现心跳,Client端连接到Server端后,会循环执行一个任务:随机等待几秒然后ping一下Server即发送一个心跳包。当等待的时间超过规定时间,将会发送失败,以为Server端在此之前已经主动断开连接了。代码如下:


Client端


ClientIdleStateTrigger —— 心跳触发器


ClientIdleStateTrigger也是一个Handler,只是重写了userEventTriggered方法,用于捕获IdleState.WRITER_IDLE事件(未在指定时间内向服务器发送数据),然后向Server端发送一个心跳包。


/** * <p> *  用于捕获{@link IdleState#WRITER_IDLE}事件(未在指定时间内向服务器发送数据),然后向<code>Server</code>端发送一个心跳包。 * </p> */public class ClientIdleStateTrigger extends ChannelInboundHandlerAdapter {    public static final String HEART_BEAT = "heart beat!";    @Override    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {        if (evt instanceof IdleStateEvent) {            IdleState state = ((IdleStateEvent) evt).state();            if (state == IdleState.WRITER_IDLE) {                // write heartbeat to server                ctx.writeAndFlush(HEART_BEAT);            }        } else {            super.userEventTriggered(ctx, evt);        }    }}


Pinger —— 心跳发射器


/** * <p>客户端连接到服务器端后,会循环执行一个任务:随机等待几秒,然后ping一下Server端,即发送一个心跳包。</p> */public class Pinger extends ChannelInboundHandlerAdapter {    private Random random = new Random();    private int baseRandom = 8;    private Channel channel;    @Override    public void channelActive(ChannelHandlerContext ctx) throws Exception {        super.channelActive(ctx);        this.channel = ctx.channel();        ping(ctx.channel());    }    private void ping(Channel channel) {        int second = Math.max(1, random.nextInt(baseRandom));        System.out.println("next heart beat will send after " + second + "s.");        ScheduledFuture<?> future = channel.eventLoop().schedule(new Runnable() {            @Override            public void run() {                if (channel.isActive()) {                    System.out.println("sending heart beat to the server...");                    channel.writeAndFlush(ClientIdleStateTrigger.HEART_BEAT);                } else {                    System.err.println("The connection had broken, cancel the task that will send a heart beat.");                    channel.closeFuture();                    throw new RuntimeException();                }            }        }, second, TimeUnit.SECONDS);        future.addListener(new GenericFutureListener() {            @Override            public void operationComplete(Future future) throws Exception {                if (future.isSuccess()) {                    ping(channel);                }            }        });    }    @Override    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {        // 当Channel已经断开的情况下, 仍然发送数据, 会抛异常, 该方法会被调用.        cause.printStackTrace();        ctx.close();    }}


ClientHandlersInitializer —— 客户端处理器集合的初始化类

public class ClientHandlersInitializer extends ChannelInitializer<SocketChannel> {    private ReconnectHandler reconnectHandler;    private EchoHandler echoHandler;    public ClientHandlersInitializer(TcpClient tcpClient) {        Assert.notNull(tcpClient, "TcpClient can not be null.");        this.reconnectHandler = new ReconnectHandler(tcpClient);        this.echoHandler = new EchoHandler();    }    @Override    protected void initChannel(SocketChannel ch) throws Exception {        ChannelPipeline pipeline = ch.pipeline();        pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));        pipeline.addLast(new LengthFieldPrepender(4));        pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));        pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));        pipeline.addLast(new Pinger());    }}


注: 上面的Handler集合,除了Pinger,其他都是编解码器和解决粘包,可以忽略。


TcpClient —— TCP连接的客户端


public class TcpClient {    private String host;    private int port;    private Bootstrap bootstrap;    /** 将<code>Channel</code>保存起来, 可用于在其他非handler的地方发送数据 */    private Channel channel;    public TcpClient(String host, int port) {        this(host, port, new ExponentialBackOffRetry(1000, Integer.MAX_VALUE, 60 * 1000));    }    public TcpClient(String host, int port, RetryPolicy retryPolicy) {        this.host = host;        this.port = port;        init();    }    /**     * 向远程TCP服务器请求连接     */    public void connect() {        synchronized (bootstrap) {            ChannelFuture future = bootstrap.connect(host, port);            this.channel = future.channel();        }    }    private void init() {        EventLoopGroup group = new NioEventLoopGroup();        // bootstrap 可重用, 只需在TcpClient实例化的时候初始化即可.        bootstrap = new Bootstrap();        bootstrap.group(group)                .channel(NioSocketChannel.class)                .handler(new ClientHandlersInitializer(TcpClient.this));    }    public static void main(String[] args) {        TcpClient tcpClient = new TcpClient("localhost", 2222);        tcpClient.connect();    }}


Server端


ServerIdleStateTrigger —— 断连触发器

/** * <p>在规定时间内未收到客户端的任何数据包, 将主动断开该连接</p> */public class ServerIdleStateTrigger extends ChannelInboundHandlerAdapter {    @Override    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {        if (evt instanceof IdleStateEvent) {            IdleState state = ((IdleStateEvent) evt).state();            if (state == IdleState.READER_IDLE) {                // 在规定时间内没有收到客户端的上行数据, 主动断开连接                ctx.disconnect();            }        } else {            super.userEventTriggered(ctx, evt);        }    }}



ServerBizHandler —— 服务器端的业务处理器


/** * <p>收到来自客户端的数据包后, 直接在控制台打印出来.</p> */@ChannelHandler.Sharablepublic class ServerBizHandler extends SimpleChannelInboundHandler<String> {    private final String REC_HEART_BEAT = "I had received the heart beat!";    @Override    protected void channelRead0(ChannelHandlerContext ctx, String data) throws Exception {        try {            System.out.println("receive data: " + data);//            ctx.writeAndFlush(REC_HEART_BEAT);        } catch (Exception e) {            e.printStackTrace();        }    }    @Override    public void channelActive(ChannelHandlerContext ctx) throws Exception {        System.out.println("Established connection with the remote client.");        // do something        ctx.fireChannelActive();    }    @Override    public void channelInactive(ChannelHandlerContext ctx) throws Exception {        System.out.println("Disconnected with the remote client.");        // do something        ctx.fireChannelInactive();    }    @Override    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {        cause.printStackTrace();        ctx.close();    }}


ServerHandlerInitializer —— 服务器端处理器集合的初始化类

/** * <p>用于初始化服务器端涉及到的所有<code>Handler</code></p> */public class ServerHandlerInitializer extends ChannelInitializer<SocketChannel> {    protected void initChannel(SocketChannel ch) throws Exception {        ch.pipeline().addLast("idleStateHandler", new IdleStateHandler(5, 0, 0));        ch.pipeline().addLast("idleStateTrigger", new ServerIdleStateTrigger());        ch.pipeline().addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));        ch.pipeline().addLast("frameEncoder", new LengthFieldPrepender(4));        ch.pipeline().addLast("decoder", new StringDecoder());        ch.pipeline().addLast("encoder", new StringEncoder());        ch.pipeline().addLast("bizHandler", new ServerBizHandler());    }}


注:new IdleStateHandler(5, 0, 0)handler代表如果在5秒内没有收到来自客户端的任何数据包(包括但不限于心跳包),将会主动断开与该客户端的连接。


TcpServer —— 服务器端


public class TcpServer {    private int port;    private ServerHandlerInitializer serverHandlerInitializer;    public TcpServer(int port) {        this.port = port;        this.serverHandlerInitializer = new ServerHandlerInitializer();    }    public void start() {        EventLoopGroup bossGroup = new NioEventLoopGroup(1);        EventLoopGroup workerGroup = new NioEventLoopGroup();        try {            ServerBootstrap bootstrap = new ServerBootstrap();            bootstrap.group(bossGroup, workerGroup)                    .channel(NioServerSocketChannel.class)                    .childHandler(this.serverHandlerInitializer);            // 绑定端口,开始接收进来的连接            ChannelFuture future = bootstrap.bind(port).sync();            System.out.println("Server start listen at " + port);            future.channel().closeFuture().sync();        } catch (Exception e) {            bossGroup.shutdownGracefully();            workerGroup.shutdownGracefully();            e.printStackTrace();        }    }    public static void main(String[] args) throws Exception {        int port = 2222;        new TcpServer(port).start();    }}


至此,所有代码已经编写完毕。


测试


首先启动客户端,再启动服务器端。启动完成后,在客户端的控制台上,可以看到打印如下类似日志:


640?wx_fmt=jpeg

客户端控制台输出的日志


在服务器端可以看到控制台输出了类似如下的日志:


640?wx_fmt=jpeg

服务器端控制台输出的日志


可以看到,客户端在发送4个心跳包后,第5个包因为等待时间较长,等到真正发送的时候,发现连接已断开了;而服务器端收到客户端的4个心跳数据包后,迟迟等不到下一个数据包,所以果断断开该连接。


异常情况


在测试过程中,有可能会出现如下情况:


640?wx_fmt=jpeg

异常情况


出现这种情况的原因是:在连接已断开的情况下,仍然向服务器端发送心跳包。虽然在发送心跳包之前会使用channel.isActive()判断连接是否可用,但也有可能上一刻判断结果为可用,但下一刻发送数据包之前,连接就断了。


目前尚未找到优雅处理这种情况的方案,各位看官如果有好的解决方案,还望不吝赐教。拜谢!!!


断线重连


断线重连这里就不过多介绍,相信各位都知道是怎么回事。这里只说大致思路,然后直接上代码。


实现思路


客户端在监测到与服务器端的连接断开后,或者一开始就无法连接的情况下,使用指定的重连策略进行重连操作,直到重新建立连接或重试次数耗尽。


对于如何监测连接是否断开,则是通过重写ChannelInboundHandler#channelInactive来实现,但连接不可用,该方法会被触发,所以只需要在该方法做好重连工作即可。


代码实现


注:以下代码都是在上面心跳机制的基础上修改/添加的。


因为断线重连是客户端的工作,所以只需对客户端代码进行修改。


重试策略


RetryPolicy —— 重试策略接口


public interface RetryPolicy {    /**     * Called when an operation has failed for some reason. This method should return     * true to make another attempt.     *     * @param retryCount the number of times retried so far (0 the first time)     * @return true/false     */    boolean allowRetry(int retryCount);    /**     * get sleep time in ms of current retry count.     *     * @param retryCount current retry count     * @return the time to sleep     */    long getSleepTimeMs(int retryCount);}


ExponentialBackOffRetry —— 重连策略的默认实现


/** * <p>Retry policy that retries a set number of times with increasing sleep time between retries</p> */public class ExponentialBackOffRetry implements RetryPolicy {    private static final int MAX_RETRIES_LIMIT = 29;    private static final int DEFAULT_MAX_SLEEP_MS = Integer.MAX_VALUE;    private final Random random = new Random();    private final long baseSleepTimeMs;    private final int maxRetries;    private final int maxSleepMs;    public ExponentialBackOffRetry(int baseSleepTimeMs, int maxRetries) {        this(baseSleepTimeMs, maxRetries, DEFAULT_MAX_SLEEP_MS);    }    public ExponentialBackOffRetry(int baseSleepTimeMs, int maxRetries, int maxSleepMs) {        this.maxRetries = maxRetries;        this.baseSleepTimeMs = baseSleepTimeMs;        this.maxSleepMs = maxSleepMs;    }    @Override    public boolean allowRetry(int retryCount) {        if (retryCount < maxRetries) {            return true;        }        return false;    }    @Override    public long getSleepTimeMs(int retryCount) {        if (retryCount < 0) {            throw new IllegalArgumentException("retries count must greater than 0.");        }        if (retryCount > MAX_RETRIES_LIMIT) {            System.out.println(String.format("maxRetries too large (%d). Pinning to %d", maxRetries, MAX_RETRIES_LIMIT));            retryCount = MAX_RETRIES_LIMIT;        }        long sleepMs = baseSleepTimeMs * Math.max(1, random.nextInt(1 << retryCount));        if (sleepMs > maxSleepMs) {            System.out.println(String.format("Sleep extension too large (%d). Pinning to %d", sleepMs, maxSleepMs));            sleepMs = maxSleepMs;        }        return sleepMs;    }}


ReconnectHandler—— 重连处理器


@ChannelHandler.Sharablepublic class ReconnectHandler extends ChannelInboundHandlerAdapter {    private int retries = 0;    private RetryPolicy retryPolicy;    private TcpClient tcpClient;    public ReconnectHandler(TcpClient tcpClient) {        this.tcpClient = tcpClient;    }    @Override    public void channelActive(ChannelHandlerContext ctx) throws Exception {        System.out.println("Successfully established a connection to the server.");        retries = 0;        ctx.fireChannelActive();    }    @Override    public void channelInactive(ChannelHandlerContext ctx) throws Exception {        if (retries == 0) {            System.err.println("Lost the TCP connection with the server.");            ctx.close();        }        boolean allowRetry = getRetryPolicy().allowRetry(retries);        if (allowRetry) {            long sleepTimeMs = getRetryPolicy().getSleepTimeMs(retries);            System.out.println(String.format("Try to reconnect to the server after %dms. Retry count: %d.", sleepTimeMs, ++retries));            final EventLoop eventLoop = ctx.channel().eventLoop();            eventLoop.schedule(() -> {                System.out.println("Reconnecting ...");                tcpClient.connect();            }, sleepTimeMs, TimeUnit.MILLISECONDS);        }        ctx.fireChannelInactive();    }    private RetryPolicy getRetryPolicy() {        if (this.retryPolicy == null) {            this.retryPolicy = tcpClient.getRetryPolicy();        }        return this.retryPolicy;    }}


ClientHandlersInitializer


在之前的基础上,添加了重连处理器ReconnectHandler


public class ClientHandlersInitializer extends ChannelInitializer<SocketChannel> {    private ReconnectHandler reconnectHandler;    private EchoHandler echoHandler;    public ClientHandlersInitializer(TcpClient tcpClient) {        Assert.notNull(tcpClient, "TcpClient can not be null.");        this.reconnectHandler = new ReconnectHandler(tcpClient);        this.echoHandler = new EchoHandler();    }    @Override    protected void initChannel(SocketChannel ch) throws Exception {        ChannelPipeline pipeline = ch.pipeline();        pipeline.addLast(this.reconnectHandler);        pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));        pipeline.addLast(new LengthFieldPrepender(4));        pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));        pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));        pipeline.addLast(new Pinger());    }}


TcpClient


在之前的基础上添加重连、重连策略的支持。


public class TcpClient {    private String host;    private int port;    private Bootstrap bootstrap;    /** 重连策略 */    private RetryPolicy retryPolicy;    /** 将<code>Channel</code>保存起来, 可用于在其他非handler的地方发送数据 */    private Channel channel;    public TcpClient(String host, int port) {        this(host, port, new ExponentialBackOffRetry(1000, Integer.MAX_VALUE, 60 * 1000));    }    public TcpClient(String host, int port, RetryPolicy retryPolicy) {        this.host = host;        this.port = port;        this.retryPolicy = retryPolicy;        init();    }    /**     * 向远程TCP服务器请求连接     */    public void connect() {        synchronized (bootstrap) {            ChannelFuture future = bootstrap.connect(host, port);            future.addListener(getConnectionListener());            this.channel = future.channel();        }    }    public RetryPolicy getRetryPolicy() {        return retryPolicy;    }    private void init() {        EventLoopGroup group = new NioEventLoopGroup();        // bootstrap 可重用, 只需在TcpClient实例化的时候初始化即可.        bootstrap = new Bootstrap();        bootstrap.group(group)                .channel(NioSocketChannel.class)                .handler(new ClientHandlersInitializer(TcpClient.this));    }    private ChannelFutureListener getConnectionListener() {        return new ChannelFutureListener() {            @Override            public void operationComplete(ChannelFuture future) throws Exception {                if (!future.isSuccess()) {                    future.channel().pipeline().fireChannelInactive();                }            }        };    }    public static void main(String[] args) {        TcpClient tcpClient = new TcpClient("localhost", 2222);        tcpClient.connect();    }}


测试


在测试之前,为了避开 Connection reset by peer 异常,可以稍微修改Pingerping()方法,添加if (second == 5)的条件判断。如下:


private void ping(Channel channel) {        int second = Math.max(1, random.nextInt(baseRandom));        if (second == 5) {            second = 6;        }        System.out.println("next heart beat will send after " + second + "s.");        ScheduledFuture<?> future = channel.eventLoop().schedule(new Runnable() {            @Override            public void run() {                if (channel.isActive()) {                    System.out.println("sending heart beat to the server...");                    channel.writeAndFlush(ClientIdleStateTrigger.HEART_BEAT);                } else {                    System.err.println("The connection had broken, cancel the task that will send a heart beat.");                    channel.closeFuture();                    throw new RuntimeException();                }            }        }, second, TimeUnit.SECONDS);        future.addListener(new GenericFutureListener() {            @Override            public void operationComplete(Future future) throws Exception {                if (future.isSuccess()) {                    ping(channel);                }            }        });    }


启动客户端


先只启动客户端,观察控制台输出,可以看到类似如下日志:


640?wx_fmt=jpeg

断线重连测试——客户端控制台输出


可以看到,当客户端发现无法连接到服务器端,所以一直尝试重连。随着重试次数增加,重试时间间隔越大,但又不想无限增大下去,所以需要定一个阈值,比如60s。如上图所示,当下一次重试时间超过60s时,会打印Sleep extension too large(*). Pinning to 60000,单位为ms。出现这句话的意思是,计算出来的时间超过阈值(60s),所以把真正睡眠的时间重置为阈值(60s)。


启动服务器端


接着启动服务器端,然后继续观察客户端控制台输出。


640?wx_fmt=jpeg

断线重连测试——服务器端启动后客户端控制台输出


可以看到,在第9次重试失败后,第10次重试之前,启动的服务器,所以第10次重连的结果为Successfully established a connection to the server.,即成功连接到服务器。接下来因为还是不定时ping服务器,所以出现断线重连、断线重连的循环。


扩展


在不同环境,可能会有不同的重连需求。有不同的重连需求的,只需自己实现RetryPolicy接口,然后在创建TcpClient的时候覆盖默认的重连策略即可。


完!!!

关注Java技术栈微信公众号,在后台回复关键字:Java,可以获取一份栈长整理的 Java 最新技术干货。

最近干货分享

面试问我 Java 逃逸分析,瞬间被秒杀了。。

到底什么是重入锁,拜托,一次搞清楚!

图解 Java 垃圾回收机制,写得非常好!

如何写出让同事无法维护的代码?

分享一份Java架构师学习资料

640

点击「阅读原文」一起搞技术,爽~

最后

以上就是爱撒娇野狼为你收集整理的Netty 如何实现心跳机制与断线重连?的全部内容,希望文章能够帮你解决Netty 如何实现心跳机制与断线重连?所遇到的程序开发问题。

如果觉得靠谱客网站的内容还不错,欢迎将靠谱客网站推荐给程序员好友。

本图文内容来源于网友提供,作为学习参考使用,或来自网络收集整理,版权属于原作者所有。
点赞(90)

评论列表共有 0 条评论

立即
投稿
返回
顶部