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;
}
}
}
