Netty实战五：Netty客户端断线重连

• 1、断线重连
  断线时，触发channelInactive方法，然后调用run()方法实现重连；

• 2、client启动连接服务器时，连接失败，调用run()方法实现重连；
  run()：重连，如果没有连上服务端，则触发channelInactive方法，再次循环调用run()；如果连接上，则触发channelActive方法，把clientId和socketChannel存储起来

• 3、利用userEventTriggered实现心跳维护，具体代码如下

1、NettyClient

package com.xxx.monitor.netty.client;

import com.xxx.monitor.common.constants.NettyConstant;
import com.xxx.monitor.common.util.StringUtil;
import com.xxx.monitor.entity.master.DeviceParams;
import com.xxx.monitor.entity.vo.BaseDataVo;
import com.xxx.monitor.entity.vo.clientRequest.DeviceMonitoringVo;
import com.xxx.monitor.entity.vo.clientRequest.DeviceParamsVo;
import com.xxx.monitor.netty.util.MessageType;
import com.xxx.monitor.netty.util.NettyMap;
import com.xxx.monitor.netty.vo.Message;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.bytes.ByteArrayDecoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.HashedWheelTimer;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@Component
public class NettyClient  {
    private static final Logger logger = LoggerFactory.getLogger(NettyClient.class);

    private int port;
    private String host;
    private String clientId;
    public static SocketChannel socketChannel;
    protected final HashedWheelTimer timer = new HashedWheelTimer();

    public void start() {
        logger.info("客户端正在启动---------------");

        EventLoopGroup eventLoopGroup = new NioEventLoopGroup();

        Bootstrap bootstrap = new Bootstrap();
        bootstrap.channel(NioSocketChannel.class);
        bootstrap.handler(new LoggingHandler(LogLevel.INFO));
        bootstrap.option(ChannelOption.SO_KEEPALIVE,true);
        bootstrap.group(eventLoopGroup);

       host = "127.0.0.1";
        port = 8765;
        clientId = "7105000000";

        final ConnectionWatchdog watchdog = new ConnectionWatchdog(bootstrap, timer, port,host, true) {
            public ChannelHandler[] handlers() {
                return new ChannelHandler[] {
                        this,
                        new IdleStateHandler(
                                NettyConstant.CLENT_READ_IDEL_TIME_OUT,
                                NettyConstant.CLENT_WRITE_IDEL_TIME_OUT,
                                NettyConstant.CLENT_ALL_IDEL_TIME_OUT,
                                TimeUnit.SECONDS),
                        new ByteArrayEncoder(),
                        new ByteArrayDecoder(),
                        new NettyClientHandler(clientId)
                };
            }
        };

        ChannelFuture future=null;
        //进行连接
        try {
            synchronized (bootstrap) {
                bootstrap.remoteAddress(host,port);
                bootstrap.handler(new ChannelInitializer<Channel>() {
                    //初始化channel
                    @Override
                    protected void initChannel(Channel ch) throws Exception {
                        ch.pipeline().addLast(watchdog.handlers());
                    }
                });
                future =  bootstrap.connect(host,port);
            }

            // 以下代码在synchronized同步块外面是安全的
            future.sync();

            if(future.isSuccess()) {
                socketChannel = (SocketChannel) future.channel();
                logger.info("客户端完成启动-------------,ip为{},端口为{}",host,port);

                Message message = new Message();
                DeviceParamsVo deviceParams = new DeviceParamsVo();
                deviceParams.setClientId("7105000000");
                deviceParams.setSimIccid("1233456678789");
                message.setMsgType(MessageType.DEV_CLIENT_PARAMS);
                message.setData(deviceParams);
                sendMessage(message);
            }
        } catch (Throwable t) {
            logger.info("客户端连接失败------------,ip为{},端口为{}",host,port);
           if (null != future) {
try {
   timer.newTimeout(watchdog,60, TimeUnit.SECONDS);
} catch (Exception e) {
   e.printStackTrace();
}
        }
        //netty优雅退出机制，这里会影响到上面定时任务的执行，所以不退出
       //eventLoopGroup.shutdownGracefully();
        }
    }



    public static void sendMessage(Message message) {

        if(StringUtils.isEmpty(message.getData().getClientId())){
            logger.error("factoryDevNo 不能为空");
            return;
        }
        SocketChannel socketChannel = NettyMap.getChannel(message.getData().getClientId());
        if (socketChannel!=null&& socketChannel.isOpen()) {
            if(StringUtil.isEmpty(message.getMsgType())){
                logger.error("消息类型不能为空");
                return;
            }
            socketChannel.writeAndFlush(message);
        }
        else{
            logger.error("客户端未连接服务器,发送消息失败!");
        }
    }

    public int getPort() {
        return port;
    }
    public void setPort(int port) {
        this.port = port;
    }
    public String getHost() {
        return host;
    }
    public void setHost(String host) {
        this.host = host;
    }
    public String getClientId() {
        return clientId;
    }
    public void setClientId(String clientId) {
        this.clientId = clientId;
    }
}

2、ConnectionWatchdog 实现重连机制

package com.xxx.monitor.netty.client;

import com.xxx.monitor.common.constants.NettyConstant;
import com.xxx.monitor.entity.vo.clientRequest.DeviceParamsVo;
import com.xxx.monitor.netty.util.MessageType;
import com.xxx.monitor.netty.util.NettyMap;
import com.xxx.monitor.netty.vo.Message;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.socket.SocketChannel;
import io.netty.util.Timeout;
import io.netty.util.Timer;
import io.netty.util.TimerTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 *
 * 重连检测狗，当发现当前的链路不稳定关闭之后，进行12次重连
 */
@Sharable
public abstract class ConnectionWatchdog extends ChannelInboundHandlerAdapter implements TimerTask，ChannelHandlerHolder{
    private static final Logger logger = LoggerFactory.getLogger(ConnectionWatchdog.class);

    private final Bootstrap bootstrap;
    private final Timer timer;
    private final int port;

    private final String host;

    private volatile boolean reconnect;
    private int attempts = 0;

    public ConnectionWatchdog(Bootstrap bootstrap, Timer timer, int port, String host, boolean reconnect) {
        this.bootstrap = bootstrap;
        this.timer = timer;
        this.port = port;
        this.host = host;
        this.reconnect = reconnect;
    }

    /**
     * channel链路每次active的时候，将其连接的次数重新☞ 0
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        logger.info("-------客户端上线----------");
        NettyClient.socketChannel = (SocketChannel) ctx.channel();


        logger.info("当前链路已经**了，重连尝试次数重新置为0");
        attempts = 0;
        ctx.fireChannelActive();
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        logger.info("-------客户端下线----------");
        boolean isChannelActive = true;
        if(null == NettyClient.socketChannel || !NettyClient.socketChannel.isActive()){
            isChannelActive = false;
        }
        if(reconnect && false){
            logger.info("链接关闭，将进行重连");

            if (attempts < NettyConstant.RET_CONNECT_TIME) {
                 int timeout = 60;
                timer.newTimeout(this, timeout,TimeUnit.SECONDS);
            }
        }
        else
        {
            logger.info("链接关闭");
        }
    }


public void run(Timeout timeout) throws Exception {
        
        ChannelFuture future;
        //bootstrap已经初始化好了，只需要将handler填入就可以了
        synchronized (bootstrap) {
            bootstrap.handler(new ChannelInitializer<Channel>() {

                @Override
                protected void initChannel(Channel ch) throws Exception {
                    
                    ch.pipeline().addLast(handlers());
                }
            });
            future = bootstrap.connect(host,port);
        }

        //future对象监听
        future.addListener(new ChannelFutureListener() {
            public void operationComplete(ChannelFuture f) throws Exception {
                boolean succeed = f.isSuccess();
                Channel channel=f.channel();
                if (!succeed) {
                    logger.info("重连失败");
                    f.channel().pipeline().fireChannelInactive();
                }else{
                    logger.info("重连成功");
                }
            }
        });
    }
}

3、ChannelHandlerHolder.java

package com.xxx.monitor.netty.client;

import io.netty.channel.ChannelHandler;

/**
 *
 * 客户端的ChannelHandler集合，由子类实现，这样做的好处：
 * 继承这个接口的所有子类可以很方便地获取ChannelPipeline中的Handlers
 * 获取到handlers之后方便ChannelPipeline中的handler的初始化和在重连的时候也能很方便
 * 地获取所有的handlers
 */
public interface ChannelHandlerHolder {

    ChannelHandler[] handlers();
}

4、NettyClientHandler.java
注：在channelActive设备上线时，通过HashMap把clientId和SocketChannel存储起来，在channelInactive设备下线时删除，这样只要连接上，任何地方都可以调用socketChannel发送消息了。

package com.xxx.monitor.netty.client;

import com.xxx.monitor.common.constants.NettyConstant;
import com.xxx.monitor.entity.vo.BaseDataVo;
import com.xxx.monitor.entity.vo.clientRequest.DeviceHeartBeatVo;
import com.xxx.monitor.netty.util.MessageType;
import com.xxx.monitor.netty.util.NettyMap;
import com.xxx.monitor.netty.vo.Message;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.timeout.IdleStateEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/*
1）客户端连接服务端
2）在客户端的的ChannelPipeline中加入一个比较特殊的IdleStateHandler，设置一下客户端的写空闲时间，例如5s
3）当客户端的所有ChannelHandler中4s内没有write事件，则会触发userEventTriggered方法（上文介绍过）
4）我们在客户端的userEventTriggered中对应的触发事件下发送一个心跳包给服务端，检测服务端是否还存活，防止服务端已经宕机，客户端还不知道
5）同样，服务端要对心跳包做出响应，其实给客户端最好的回复就是“不回复”，这样可以服务端的压力，假如有10w个空闲Idle的连接，那么服务端光发送心跳回复，则也是费事的事情，那么怎么才能告诉客户端它还活着呢，其实很简单，因为5s服务端都会收到来自客户端的心跳信息，那么如果10秒内收不到，服务端可以认为客户端挂了，可以close链路
6）加入服务端因为什么因素导致宕机的话，就会关闭所有的链路链接，所以作为客户端要做的事情就是断线重连
 */
@ChannelHandler.Sharable
public class NettyClientHandler extends SimpleChannelInboundHandler<byte[]> {
    private static final Logger logger = LoggerFactory.getLogger(NettyClientHandler.class);

    private int pingTime=0;//心跳次数
    Message heartBeatMsg=null;//心跳数据
    String clientId="";
    public NettyClientHandler(String clientId) {
        this.clientId=clientId;
        DeviceHeartBeatVo heartBeat = new DeviceHeartBeatVo();
        heartBeat.setClientId(clientId);
        heartBeat.setTimeSp(System.currentTimeMillis());
        heartBeatMsg=new Message(MessageType.DEV_CLIENT_HEART_BEAT,heartBeat);
    }

    //利用写空闲发送心跳检测消息
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent e = (IdleStateEvent) evt;
            switch (e.state()) {
                case WRITER_IDLE:
                    //发送心跳到服务器
                    //一分钟发送一次心跳
                    pingTime++;

                    //一天重置一次
                    if(pingTime==1024){
                        pingTime=0;
                    }
                    ctx.writeAndFlush(heartBeatMsg);
                    logger.info("客户端发送心跳----------");
                    break;
                default:
                    break;
            }
        }
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        NettyConstant.CLIENT_NETWORK_STATUS=1;
        logger.info("-------客户端上线----------{}",clientId);
        NettyMap.putChannel(clientId,(SocketChannel)ctx.channel());
        //发送登录消息
        ctx.fireChannelActive();
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        NettyConstant.CLIENT_NETWORK_STATUS=0;
        NettyMap.removeChannel(clientId);
        logger.info("-------客户端下线----------");
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, byte[] msg) throws Exception {
        //客户端通道
        Message message = new Message(msg);
        String msgType = message.getMsgType();
        int dataLength = message.getDataLength();
        if(null == message.getData()){
            return;
        }
        String facDevNo = message.getData().getClientId();

        this.producerRun(ctx,msgType,message.getData(),facDevNo);
    }


    private void producerRun(ChannelHandlerContext ctx, String msgType, BaseDataVo vo, String facDevNo) {

        switch (msgType) {
            //心跳 接收应答
            case MessageType.DEV_CLIENT_HEART_BEAT:

                break;

        }
    }

}