Netty 4.0 实现心跳检测和断线重连

时间 2019-11-19

标签 netty 4.0 实现心跳检测断线栏目 Netty 繁體版

原文原文链接

一实现心跳检测
原理：当服务端每隔一段时间就会向客户端发送心跳包，客户端收到心跳包后一样也会回一个心跳包给服务端
通常状况下，客户端与服务端在指定时间内没有任何读写请求，就会认为链接是idle（空闲的）的。此时，客户端须要向服务端发送心跳消息，来维持服务端与客户端的连接。那么怎么判断客户端在指定时间里没有任何读写请求呢？netty中为咱们提供一个特别好用的IdleStateHandler来干这个苦差事！

在服务端工做线程中添加：
java

Java代码 android

arg0.pipeline().addLast("ping", new IdleStateHandler(25, 15, 10,TimeUnit.SECONDS)); bootstrap

这个处理器，它的做用就是用来检测客户端的读取超时的，该类的第一个参数是指定读操做空闲秒数，第二个参数是指定写操做的空闲秒数，第三个参数是指定读写空闲秒数，当有操做操做超出指定空闲秒数时，便会触发UserEventTriggered事件。因此咱们只须要在本身的handler中截获该事件，而后发起相应的操做便可（好比说发起心跳操做）。如下是咱们自定义的handler中的代码：
服务器

Java代码 socket

/** ide
* 一段时间未进行读写操做回调函数
*/ oop
@Override spa
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) .net
throws Exception {
// TODO Auto-generated method stub
super.userEventTriggered(ctx, evt);
if (evt instanceof IdleStateEvent) {
IdleStateEvent event = (IdleStateEvent) evt;
if (event.state().equals(IdleState.READER_IDLE)) {
//未进行读操做
System.out.println("READER_IDLE");
// 超时关闭channel
ctx.close();
} else if (event.state().equals(IdleState.WRITER_IDLE)) {
} else if (event.state().equals(IdleState.ALL_IDLE)) {
//未进行读写
System.out.println("ALL_IDLE");
// 发送心跳消息
MsgHandleService.getInstance().sendMsgUtil.sendHeartMessage(ctx);
}
}
}

也就是说服务端在10s内未进行读写操做，就会向客户端发送心跳包，客户端收到心跳包后当即回复心跳包给服务端，此时服务端就进行了读操做，也就不会触发IdleState.READER_IDLE（未读操做状态），若客户端异常掉线了，并不能响应服务端发来的心跳包，在25s后就会触发IdleState.READER_IDLE（未读操做状态），此时服务器就会将通道关闭

客户端代码略

二客户端实现断线重连
原理当客户端链接服务器时

Java代码

bootstrap.connect(new InetSocketAddress(
serverIP, port));

会返回一个ChannelFuture的对象，咱们对这个对象进行监听
代码以下：

Java代码

import android.os.Handler;
import android.os.HandlerThread;
import android.os.Message;
import android.util.Log;
import com.ld.qmwj.Config;
import com.ld.qmwj.MyApplication;
import java.net.InetSocketAddress;
import java.nio.charset.Charset;
import java.util.concurrent.TimeUnit;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
/**
* Created by zsg on 2015/11/21.
*/
public class MyClient implements Config {
private static Bootstrap bootstrap;
private static ChannelFutureListener channelFutureListener = null;
public MyClient() {
}
// 初始化客户端
public static void initClient() {
NioEventLoopGroup group = new NioEventLoopGroup();
// Client服务启动器 3.x的ClientBootstrap
// 改成Bootstrap，且构造函数变化很大，这里用无参构造。
bootstrap = new Bootstrap();
// 指定EventLoopGroup
bootstrap.group(group);
// 指定channel类型
bootstrap.channel(NioSocketChannel.class);
// 指定Handler
bootstrap.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
// 建立分隔符缓冲对象
ByteBuf delimiter = Unpooled.copiedBuffer("#"
.getBytes());
// 当达到最大长度仍没找到分隔符就抛出异常
ch.pipeline().addLast(
new DelimiterBasedFrameDecoder(10000, true, false, delimiter));
// 将消息转化成字符串对象下面的到的消息就不用转化了
//解码
ch.pipeline().addLast(new StringEncoder(Charset.forName("UTF-8")));
ch.pipeline().addLast(new StringDecoder(Charset.forName("GBK")));
ch.pipeline().addLast(new MyClientHandler());
}
});
//设置TCP协议的属性
bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
bootstrap.option(ChannelOption.TCP_NODELAY, true);
bootstrap.option(ChannelOption.SO_TIMEOUT, 5000);
channelFutureListener = new ChannelFutureListener() {
public void operationComplete(ChannelFuture f) throws Exception {
// Log.d(Config.TAG, "isDone:" + f.isDone() + " isSuccess:" + f.isSuccess() +
// " cause" + f.cause() + " isCancelled" + f.isCancelled());
if (f.isSuccess()) {
Log.d(Config.TAG, "从新链接服务器成功");
} else {
Log.d(Config.TAG, "从新链接服务器失败");
// 3秒后从新链接
f.channel().eventLoop().schedule(new Runnable() {
@Override
public void run() {
doConnect();
}
}, 3, TimeUnit.SECONDS);
}
}
};
}
// 链接到服务端
public static void doConnect() {
Log.d(TAG, "doConnect");
ChannelFuture future = null;
try {
future = bootstrap.connect(new InetSocketAddress(
serverIP, port));
future.addListener(channelFutureListener);
} catch (Exception e) {
e.printStackTrace();
//future.addListener(channelFutureListener);
Log.d(TAG, "关闭链接");
}
}
}

监听到链接服务器失败时，会在3秒后从新链接（执行doConnect方法）

这还不够，当客户端掉线时要进行从新链接
在咱们本身定义逻辑处理的Handler中

Java代码

@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
Log.d(Config.TAG, "与服务器断开链接服务器");
super.channelInactive(ctx);
MsgHandle.getInstance().channel = null;
//从新链接服务器
ctx.channel().eventLoop().schedule(new Runnable() {
@Override
public void run() {
MyClient.doConnect();
}
}, 2, TimeUnit.SECONDS);
ctx.close();
}