netty心跳过程中发送消息失败_Netty4.0实现心跳检测和断线重连

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

这个处理器&＃xff0c;它的作用就是用来检测客户端的读取超时的&＃xff0c;该类的第一个参数是指定读操作空闲秒数&＃xff0c;第二个参数是指定写操作的空闲秒数&＃xff0c;第三个参数是指定读写空闲秒数&＃xff0c;当有操作操作超出指定空闲秒数时&＃xff0c;便会触发UserEventTriggered事件。所以我们只需要在自己的handler中截获该事件&＃xff0c;然后发起相应的操作即可(比如说发起心跳操作)。以下是我们自定义的handler中的代码&＃xff1a;

/**

* 一段时间未进行读写操作 回调

*/

&＃64;Override

public void userEventTriggered(ChannelHandlerContext ctx, Object evt)

throws Exception {

// TODO Auto-generated method stub

super.userEventTriggered(ctx, evt);

if (evt instanceof IdleStateEvent) {

IdleStateEvent event &＃61; (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内未进行读写操作&＃xff0c;就会向客户端发送心跳包&＃xff0c;客户端收到心跳包后立即回复心跳包给服务端&＃xff0c;此时服务端就进行了读操作&＃xff0c;也就不会触发IdleState.READER_IDLE(未读操作状态)&＃xff0c;若客户端异常掉线了&＃xff0c;并不能响应服务端发来的心跳包&＃xff0c;在25s后就会触发IdleState.READER_IDLE(未读操作状态)&＃xff0c;此时服务器就会将通道关闭

客户端代码略

二 客户端实现断线重连

原理当客户端连接服务器时

bootstrap.connect(new InetSocketAddress(

serverIP, port));

会返回一个ChannelFuture的对象&＃xff0c;我们对这个对象进行监听

代码如下&＃xff1a;

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 &＃61; null;

public MyClient() {

}

// 初始化客户端

public static void initClient() {

NioEventLoopGroup group &＃61; new NioEventLoopGroup();

// Client服务启动器 3.x的ClientBootstrap

// 改为Bootstrap&＃xff0c;且构造函数变化很大&＃xff0c;这里用无参构造。

bootstrap &＃61; new Bootstrap();

// 指定EventLoopGroup

bootstrap.group(group);

// 指定channel类型

bootstrap.channel(NioSocketChannel.class);

// 指定Handler

bootstrap.handler(new ChannelInitializer() {

&＃64;Override

protected void initChannel(SocketChannel ch) throws Exception {

// 创建分隔符缓冲对象

ByteBuf delimiter &＃61; 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 &＃61; new ChannelFutureListener() {

public void operationComplete(ChannelFuture f) throws Exception {

// Log.d(Config.TAG, "isDone:" &＃43; f.isDone() &＃43; " isSuccess:" &＃43; f.isSuccess() &＃43;

// " cause" &＃43; f.cause() &＃43; " isCancelled" &＃43; f.isCancelled());

if (f.isSuccess()) {

Log.d(Config.TAG, "重新连接服务器成功");

} else {

Log.d(Config.TAG, "重新连接服务器失败");

// 3秒后重新连接

f.channel().eventLoop().schedule(new Runnable() {

&＃64;Override

public void run() {

doConnect();

}

}, 3, TimeUnit.SECONDS);

}

}

};

}

// 连接到服务端

public static void doConnect() {

Log.d(TAG, "doConnect");

ChannelFuture future &＃61; null;

try {

future &＃61; bootstrap.connect(new InetSocketAddress(

serverIP, port));

future.addListener(channelFutureListener);

} catch (Exception e) {

e.printStackTrace();

//future.addListener(channelFutureListener);

Log.d(TAG, "关闭连接");

}

}

}

监听到连接服务器失败时&＃xff0c;会在3秒后重新连接(执行doConnect方法)

这还不够&＃xff0c;当客户端掉线时要进行重新连接

在我们自己定义逻辑处理的Handler中

&＃64;Override

public void channelInactive(ChannelHandlerContext ctx) throws Exception {

Log.d(Config.TAG, "与服务器断开连接服务器");

super.channelInactive(ctx);

MsgHandle.getInstance().channel &＃61; null;

//重新连接服务器

ctx.channel().eventLoop().schedule(new Runnable() {

&＃64;Override

public void run() {

MyClient.doConnect();

}

}, 2, TimeUnit.SECONDS);

ctx.close();

}

3

顶

1

踩

分享到&＃xff1a;

2016-02-17 21:42

浏览 30989

评论

3 楼

VIP庚

2016-12-23

sdtzyb 写道

MsgHandle 这个里面写的是什么为什么需要把        MsgHandle.getInstance().channel &＃61; null;

弄成空。

能不能把提懂点详细代码

channel是客户端与服务器之间通信的通道&＃xff0c;和socket类似&＃xff0c;当客户端与服务器连接成功后&＃xff0c;会将chanel保存。这里是客户端与服务器断开连接后进行重连&＃xff0c;所以要将原先的通道变成空

2 楼

sdtzyb

2016-12-21

MsgHandle 这个里面写的是什么为什么需要把        MsgHandle.getInstance().channel &＃61; null;

弄成空。

能不能把提懂点详细代码

1 楼

newboy2004

2016-04-21

对netty应用理解的更深入了一点