Netty 断线重连解决方案

本篇文章是Netty专题的第七篇,前面六篇文章如下:

用Netty实现长连接服务,当发生下面的情况时,会发生断线的情况。

  • 网络问题
  • 客户端启动时服务端挂掉了,连接不上服务端
  • 客户端已经连接服务端,服务端突然挂掉了
  • 其它问题等…

如何解决上面的问题?

1.心跳机制检测连接存活

长连接是指建立的连接长期保持,不管有无数据包的发送都要保持连接通畅。心跳是用来检测一个系统是否存活或者网络链路是否通畅的一种方式,一般的做法是客户端定时向服务端发送心跳包,服务端收到心跳包后进行回复,客户端收到回复说明服务端存活。

通过心跳检测机制,可以检测客户端与服务的长连接是否保持,当客户端发送的心跳包没有收到服务端的响应式,可以认为服务端已经出故障了,这个时候可以重新连接或者选择其他的可用的服务进行连接。

在Netty中提供了一个IdleStateHandler类用于心跳检测,用法如下:

ch.pipeline().addLast("ping", new IdleStateHandler(60, 20, 60 * 10, TimeUnit.SECONDS));
  • 第一个参数 60 表示读操作空闲时间
  • 第二个参数 20 表示写操作空闲时间
  • 第三个参数 60*10 表示读写操作空闲时间
  • 第四个参数 单位/秒

在处理数据的ClientPoHandlerProto中增加userEventTriggered用来接收心跳检测结果,event.state()的状态分别对应上面三个参数的时间设置,当满足某个时间的条件时会触发事件。

public class ClientPoHandlerProto extends ChannelInboundHandlerAdapter {
    private ImConnection imConnection = new ImConnection();
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        MessageProto.Message message = (MessageProto.Message) msg;
        System.out.println("client:" + message.getContent());
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        super.userEventTriggered(ctx, evt);
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            if (event.state().equals(IdleState.READER_IDLE)) {
                System.out.println("长期没收到服务器推送数据");
                //可以选择重新连接
            } else if (event.state().equals(IdleState.WRITER_IDLE)) {
                System.out.println("长期未向服务器发送数据");
                //发送心跳包
                ctx.writeAndFlush(MessageProto.Message.newBuilder().setType(1));
            } else if (event.state().equals(IdleState.ALL_IDLE)) {
                System.out.println("ALL");
            }
        }
    }
}

服务端收到客户端发送的心跳消息后,回复一条信息

public class ServerPoHandlerProto extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        MessageProto.Message message = (MessageProto.Message) msg;
        if (ConnectionPool.getChannel(message.getId()) == null) {
            ConnectionPool.putChannel(message.getId(), ctx);
        }
        System.err.println("server:" + message.getId());
        // ping
        if (message.getType() == 1) {
            ctx.writeAndFlush(message);
        }
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}

当客户端20秒没往服务端发送过数据,就会触发IdleState.WRITER_IDLE事件,这个时候我们就像服务端发送一条心跳数据,跟业务无关,只是心跳。服务端收到心跳之后就会回复一条消息,表示已经收到了心跳的消息,只要收到了服务端回复的消息,那么就不会触发IdleState.READER_IDLE事件,如果触发了IdleState.READER_IDLE事件就说明服务端没有给客户端响应,这个时候可以选择重新连接。

2.启动时连接重试

在Netty中实现重连的操作比较简单,Netty已经封装好了,我们只需要稍微扩展一下即可。

连接的操作是客户端这边执行的,重连的逻辑也得加在客户端,首先我们来看启动时要是连接不上怎么去重试

增加一个负责重试逻辑的监听器,代码如下:

import java.util.concurrent.TimeUnit;
import com.netty.im.client.ImClientApp;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.EventLoop;
/**
 * 负责监听启动时连接失败,重新连接功能
 * @author yinjihuan
 *
 */
public class ConnectionListener implements ChannelFutureListener {
    private ImConnection imConnection = new ImConnection();
    @Override
    public void operationComplete(ChannelFuture channelFuture) throws Exception {
        if (!channelFuture.isSuccess()) {
            final EventLoop loop = channelFuture.channel().eventLoop();
            loop.schedule(new Runnable() {
                @Override
                public void run() {
                    System.err.println("服务端链接不上,开始重连操作...");
                    imConnection.connect(ImClientApp.HOST, ImClientApp.PORT);
                }
            }, 1L, TimeUnit.SECONDS);
        } else {
            System.err.println("服务端链接成功...");
        }
    }
}

通过channelFuture.isSuccess()可以知道在连接的时候是成功了还是失败了,如果失败了我们就启动一个单独的线程来执行重新连接的操作。

只需要在ConnectionListener添加到ChannelFuture中去即可使用

public class ImConnection {
    private Channel channel;
    public Channel connect(String host, int port) {
        doConnect(host, port);
        return this.channel;
    }
    private void doConnect(String host, int port) {
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(workerGroup);
            b.channel(NioSocketChannel.class);
            b.option(ChannelOption.SO_KEEPALIVE, true);
            b.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch) throws Exception {
                    // 实体类传输数据,protobuf序列化
                    ch.pipeline().addLast("decoder",  
                            new ProtobufDecoder(MessageProto.Message.getDefaultInstance()));  
                    ch.pipeline().addLast("encoder",  
                            new ProtobufEncoder());  
                    ch.pipeline().addLast(new ClientPoHandlerProto());
                }
            });
            ChannelFuture f = b.connect(host, port);
            f.addListener(new ConnectionListener());
            channel = f.channel();
        } catch(Exception e) {
            e.printStackTrace();
        }
    }
}

可以按照如下步骤进行测试:

  • 直接启动客户端,不启动服务端
  • 当连接失败的时候会进入ConnectionListener中的operationComplete方法执行我们的重连逻辑

3.运行中连接断开时重试

使用的过程中服务端突然挂了,就得用另一种方式来重连了,可以在处理数据的Handler中进行处理。

public class ClientPoHandlerProto extends ChannelInboundHandlerAdapter {
    private ImConnection imConnection = new ImConnection();
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        MessageProto.Message message = (MessageProto.Message) msg;
        System.out.println("client:" + message.getContent());
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.err.println("掉线了...");
        //使用过程中断线重连
        final EventLoop eventLoop = ctx.channel().eventLoop();
        eventLoop.schedule(new Runnable() {
            @Override
            public void run() {
                imConnection.connect(ImClientApp.HOST, ImClientApp.PORT);
            }
        }, 1L, TimeUnit.SECONDS);
        super.channelInactive(ctx);
    }
}

在连接断开时都会触发 channelInactive 方法, 处理重连的逻辑跟上面的一样。

可以按照如下步骤进行测试:

  • 启动服务端
  • 启动客户端,连接成功
  • 停掉服务端就会触发channelInactive进行重连操作

源码参考:https://github.com/yinjihuan/netty-im

原文发布于微信公众号 - 猿天地(cxytiandi)

原文发表时间:2018-03-16

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏JetpropelledSnake

Python入门之logging模块

本章目录:     一、logging模块简介     二、logging模块的使用     三、通过JSON或者YMAL文件配置logging模块 ====...

379120
来自专栏用户2442861的专栏

25个经典的Spring面试问答

本人收集了一些在大家在面试时被经常问及的关于Spring的主要问题,这些问题有可能在你下次面试时就会被问到。对于本文中未提及的Spring其他模块,我会单独分...

11020
来自专栏java一日一条

Spring面试问答

Spring框架是一个为Java应用程序的开发提供了综合、广泛的基础性支持的Java平台。Spring帮助开发者解决了开发中基础性的问题,使得开发人员可以专注于...

12110
来自专栏吴伟祥

Settings -> Plugins 原

Free Mybatis Plugins    (*mapper.java-- *mapper.xml)

8220
来自专栏精讲JAVA

Spring MVC 到底是如何工作的?

这篇文章将深入探讨Spring框架的一部分——Spring Web MVC的强大功能及其内部工作原理。

10230
来自专栏高性能服务器开发

(五)如何编写高性能日志

一、服务器端日志与客户端日志的区别 在正式讲解之前,我们先来看一个日志类的实现方法,这个日志类也是代表着大多数客户端日志的主流写法: /** *@desc:...

44080
来自专栏安富莱嵌入式技术分享

【RL-TCPnet网络教程】第26章 RL-TCPnet之DHCP应用

本章节为大家讲解RL-TCPnet的DHCP应用,学习本章节前,务必要优先学习第25章的DHCP基础知识。有了这些基础知识之后,再搞本章节会有事半功倍的效果。

10910
来自专栏钟绍威的专栏

锁原理:偏向锁、轻量锁、重量锁1.加锁2.撤销偏向锁1.加锁2.解锁3.膨胀为重量级锁

 java中每个对象都可作为锁,锁有四种级别,按照量级从轻到重分为:无锁、偏向锁、轻量级锁、重量级锁。每个对象一开始都是无锁的,随着线程间争夺锁,越激烈,锁的级...

1.1K50
来自专栏Android 研究

Android跨进程通信IPC之7——Binder相关结构体简介

binder_node 代表的是Binder实体对象,每一个service组件或者ServiceManager在Binder驱动程序中的描述,Binder驱动通...

13120
来自专栏潇涧技术专栏

Android Ndk and Opencv Development 2

本节主要介绍的内容是Android NDK开发的核心内容和开发总结(包括很多常见问题的解决方案)。

11220

扫码关注云+社区

领取腾讯云代金券