netty 实现心跳检查--断开重连--通俗易懂

时间：2019-12-01 22:55:30 阅读：178 评论：0 收藏：0 [点我收藏+]

标签：obj 表示 dex style dha bootstra package oid uml类图

一.心跳介绍

　　网络中的接收和发送数据都是使用操作系统中的SOCKET进行实现。但是如果此套接字已经断开，那发送数据和接收数据的时候就一定会有问题。

1.心跳机制：

　　是服务端和客户端定时的发送一个心跳包(自定义的数据结构体)，让对方知道自己还活着，处于在线状态，以确保连接真实有效的一种机制。

2.心跳检查：

　　心跳检查是查看服务端和客户端是否定时的在正常的发送心跳包。

　　在java的定时线程任务中，我们也可以去实现定时的一些轮询任务，但是netty给我们提供了一些自身封装实现好的一些心跳检查机制，我们可以利用netty来实现高效的心跳检查机制。

二.netty 提供的心跳

　　netty4.x中为我们提供了IdleStateHandler来检查服务端和客户端的心跳。

IdleStateHandler 类中是这样描述的：
triggers an {@link IdleStateEvent} when a {@link Channel} has not performed read, write, or both operation for a while.
    解释：在一段时间内，如果有读、写、读写空闲时发生时，会触发这个这个事件
        IdleStateHandler会记录IdleStateEvent事件(读空闲、写空闲、读写空闲)交给下一个handler处理
IdleStateHandler(long readerIdleTime, long writerIdleTime, long allIdleTime,TimeUnit unit) 
        参数说明：
            1. long readerIdleTime : 表示多长时间没有读, 就会发送一个心跳检测包检测是否连接
            2. long writerIdleTime : 表示多长时间没有写, 就会发送一个心跳检测包检测是否连接
            3. long allIdleTime : 表示多长时间没有读写, 就会发送一个心跳检测包检测是否连接
            4.TimeUnit unit：时间大小

三.自定义心跳实现

下面我们利用netty的IdleStateHandler来实现一个断开重连的心跳检查机制

1.心跳实现思路：

服务端：
      服务端正常配置启动，并利用IdleStateHandler中的IdleStateEvent事件，在发生5秒后没有读事件发生时，就会触发userEventTrigger事件，如果服务端在5秒内没有发生读的事件，说明客户端已经断开。
      服务端正常编写，只不过是多了一个IdleStateHandler事件处理的handler而已。
         
客户端：
      客户端需要考虑2件事，第1是怎么定时的去向服务端发送数据，第2是如果失败时怎样去尝试再次连接。好在netty的handler都已提供了相应的处理机制和方法。
      1.定时发送数据问题：
            客户端利用IdleStateHandler的事件特性在发生IdleStateEvent后，会记录下触发的事件，然后交给下一下handler处理，我们可以通过ChannelInboundHandlerAdapter的userEventTriggered方法来向服务端写数据，也就是说如果4秒内没有发生写事件，就会触发userEventTrigger方法，我们可以在该方法中向服务端写数据。
      2.重连问题：
            当服务端发生异常断开时，我们可以利用ChannelInboundHandlerAdapter的channelInactive方法进行重连。在这里需要注意，由于netty每次进行重连时会使用的Bootstrap是不共享的，因此需要通过设置@Sharable标签让bootstrap数据共享，这样当每次尝试重连时就可以把之前设置的一些绑定信息可以共享使用。

2 .UML类图

技术图片

3.实现代码：

3.1 服务端代码实现

服务端代码实现没什么难度，一共是3个类组成：

HeartBeatServer :            服务端绑定启动项参数配置　　　　　　
HeartBeatServerInitHandler : 服务端创建时加载netty的channelhandler
HeartBeatServerHandler :     服务端创建时加载自定义的channelhandler

3.2客户端代码实现

　客户端代码稍微复杂一点，但其本质上和普通的客户端都一样

HeartBeatClient　　　　　　　　　  : 客户端绑定启动项参数配置
ClientUserEventTriggeredHandler : 客户端心跳事件发生时触发此类中的方法
ClientReconnectHandler　　　　　　: 客户端断开连接后，尝试重连的自定义handler,该类是个抽象类，需要在调用时传入相应的参数，具体情况在该类上有解释
FireChannelHandlers　　　　　　　 : 客户端在尝试重连时，需要透传的参数

HeartBeatServer

package com.zpb.netty.heartbeat.server;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;

/**
 * @Desc: com.zpb.netty.heare1
 * @Date: 2019/11/30
 * @Auther: pengbo.zhao
 * @version: 1.0
 */
public class HeartBeatServer {

    private int port;
    public HeartBeatServer(int port) {
        this.port = port;
    }
    public void start(){
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {

            ServerBootstrap serverBootstrap = new ServerBootstrap().group(bossGroup, workerGroup);
            serverBootstrap.channel(NioServerSocketChannel.class);
            serverBootstrap.handler(new LoggingHandler(LogLevel.INFO));
            serverBootstrap.option(ChannelOption.SO_BACKLOG, 128);
            serverBootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);
            serverBootstrap.childHandler(new ServerInitHandler());
           
            ChannelFuture channelFuture = serverBootstrap.bind(port).sync();
            System.out.println("Server start listen at... " + port);
            
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
    public static void main(String[] args) {
        new HeartBeatServer(8888).start();
    }
}

HeartBeatServerInitHandler

package com.zpb.netty.heartbeat.server;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.timeout.IdleStateHandler;

import java.util.concurrent.TimeUnit;

/**
 * @Desc: com.zpb.netty.demo
 * @Date: 2019/11/30
 * @Auther: pengbo.zhao
 * @version: 1.0
 */
public class HeartBeatServerInitHandler extends ChannelInitializer<SocketChannel> {

    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        ChannelPipeline pipeline = socketChannel.pipeline();
        //添加心跳检查包
        pipeline.addLast(new IdleStateHandler(5,0,0,TimeUnit.SECONDS));
        pipeline.addLast("decoder", new StringDecoder());
        pipeline.addLast("encoder", new StringEncoder());
        pipeline.addLast(new HeartBeatServerHandler());
    }
}

HeartBeatServerHandler

package com.zpb.netty.heartbeat.server;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;

/**
 * @Desc: com.zpb.netty.demo
 * @Date: 2019/11/30
 * @Auther: pengbo.zhao
 * @version: 1.0
 */
public class HeartBeatServerHandler extends ChannelInboundHandlerAdapter {

    //当服务器5秒内没有发生读的事件时，会触发这个事件
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleState state = ((IdleStateEvent) evt).state();
            if (state == IdleState.READER_IDLE) { //当事件为读事件触发时发生异常，或者中断
                throw new Exception("idle exception");//将通道进行关闭
            }
        }else {
            super.userEventTriggered(ctx, evt);
        }
    }
    //当通道有读事件时
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("server channelRead..");
        System.out.println(ctx.channel().remoteAddress() + "->Server :" + msg.toString());
    }

    //当通道发生异常时
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println("server happend exception ,server close channel :"+cause.getMessage());
        ctx.close();
    }
}

HeartBeatClient

package com.zpb.netty.heartbeat.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * @Desc: com.zpb.netty.demo.client
 * @Date: 2019/11/30
 * @Auther: pengbo.zhao
 * @version: 1.0
 */
public class HeartBeatClient {

    public void start(String host,int port){
        EventLoopGroup group = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();

        bootstrap.group(group);                                 //设置线程组
        bootstrap.channel(NioSocketChannel.class);              //设置管道

        final ClientReconnectHandler clientReconnectHandler = new ClientReconnectHandler(bootstrap, host, port) {
            @Override
            public ChannelHandler[] channelHandlers() {
                return new ChannelHandler[]{
                        this,　　　　　　　　　　　　　　　　　　　　　　　　　　　//重连的handler
                        new LoggingHandler(LogLevel.INFO),                //日志handler
                        new StringDecoder(),                             //编码handler
                        new StringEncoder(),                             //解码handler
                        new IdleStateHandler(0, 4, 0, TimeUnit.SECONDS), //心跳检查handler
                        new ClientUserEventTriggeredHandler()            //心跳检查失败handler
                };                                                     
            }
        };
        
        System.err.println("client is ready......");
        ChannelFuture channelFuture = null;
        try {
            synchronized (bootstrap) {
                bootstrap.handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline().addLast(clientReconnectHandler.channelHandlers());//正常情况时的连接绑定
                    }
                });
                channelFuture = bootstrap.connect(host,port).sync();
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    public static void main(String[] args) {
        ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
        new HeartBeatClient().start("127.0.0.1",8888);
        executorService.scheduleAtFixedRate(()->{
            System.out.println("客户端获取服务端是否在线的状态："+ClientReconnectHandler.CONNECTION_STATE);
        },800,800,TimeUnit.MILLISECONDS);
    }
}

ClientUserEventTriggeredHandler

package com.zpb.netty.heartbeat.client;

import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.CharsetUtil;

/**
 * 客户端的写事件
 * @Desc: com.zpb.netty.demo.client
 * @Date: 2019/11/30
 * @Auther: pengbo.zhao
 * @version: 1.0
 */
public class ClientUserEventTriggeredHandler extends ChannelInboundHandlerAdapter{

    //当超过n秒没有写时会触发该事件
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleState state = ((IdleStateEvent) evt).state();
            if (state == IdleState.WRITER_IDLE) {
                ctx.writeAndFlush(Unpooled.copiedBuffer("ping",CharsetUtil.UTF_8));
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }
}

ClientReconnectHandler

package com.zpb.netty.heartbeat.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;

import java.util.concurrent.TimeUnit;

/**
 * 该类继承了ChannelInboundHandlerAdapter方法，目的是为了重写channelActive 和channelInactive 2个方法
 *      channelActive   方法是： 在通道建立时，可以知道此时的客户端和服务端已经建立了连接
 *      channelInactive 方法是： 在通道断开后，可以知道此时的客户端已经和服务断断开了连接，需要在这个方法中设置重连客户端方法
 * 
 * 该类实现了netty的接口TimerTask,目的是为了重写run()方法
 *      run(TimeOut timeout) 方法是：写具体的重连方案
 *
 * 该类实现了RireChannelHandlers 这个接口，目的是为了重写channelHandlers()方法
 *      channelHandlers() 方法是： 获得所有的通道配置处理的channelHandler,包括netty提供的和自定义的实现的，重点是该类并没有实现这个接口，因为关于客户端的一些启动项配置参数，我们在这里是并不知道客户端要怎样配置的，所以这才是把该类定义抽象类的关键
 *                                让子类去实现这个方法更为合理。
 *                    
 * @Sharabel 标签
 * 　　　　该注解的目的是在每次重连时，可以让此类中的的channelhandler可以共享，多次使用
 *
 * @Date: 2019/11/30
 * @Auther: pengbo.zhao
 * @version: 1.0
 */
@Sharable 
public abstract class ClientReconnectHandler extends ChannelInboundHandlerAdapter implements TimerTask,FireChannelHandlers {

    public static volatile boolean CONNECTION_STATE = false;//对外提供连接标志
    protected final HashedWheelTimer timer = new HashedWheelTimer();
    private int reconnectCount;
    private final Bootstrap bootstrap;
    private final String host;
    private final int port;

    public ClientReconnectHandler(Bootstrap bootstrap, String host, int port) {
        this.bootstrap = bootstrap;
        this.host = host;
        this.port = port;
    }

    //当通道建立时
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("当前链路已经激活了，重连尝试次数重新置为0");
        reconnectCount = 0;
        CONNECTION_STATE = true;
        ctx.fireChannelActive();
    }
    //通道关闭时启动重连
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("通道关闭,将再次进行重连");
        CONNECTION_STATE = false;
        if (reconnectCount < 12) {
            reconnectCount++;
            System.out.println("重连第"+reconnectCount+"次");
            int timeout = 2 << reconnectCount;
            timer.newTimeout(this, timeout, TimeUnit.MILLISECONDS);
        }
        ctx.fireChannelInactive();
    }

    @Override
    public void run(Timeout timeout) throws Exception {
        ChannelFuture channelFuture;
        synchronized (bootstrap) {
            bootstrap.handler(new ChannelInitializer<Channel>() {
                @Override
                protected void initChannel(Channel ch) throws Exception {
                    ch.pipeline().addLast(channelHandlers());
                }
            });
            channelFuture = bootstrap.connect(host,port);
        }
        
        //添加重连监听
        channelFuture.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture channelFuture) throws Exception {
                boolean success = channelFuture.isSuccess();
                if(!success){
                    System.out.println("重连失败");
                    channelFuture.channel().pipeline().fireChannelInactive();
                }else{
                     CONNECTION_STATE = true;
                    System.out.println("重连成功");
                }
            }
        });
    }
}

FireChannelHandlers

package com.zpb.netty.heartbeat.client;

import io.netty.channel.ChannelHandler;

/**
 * 透传handler列表
 * @Desc: com.zpb.netty.demo.client
 * @Date: 2019/12/1
 * @Auther: pengbo.zhao
 * @version: 1.0
 */
public interface FireChannelHandlers {
    ChannelHandler [] channelHandlers();
}

服务端启动：

技术图片