本博客 貓叔的博客金麸，轉(zhuǎn)載請申明出處
閱讀本文約 “4分鐘”
適讀人群：Java-Netty 初級
Netty自動重連機制
版本：netty 4.1.*
申明：本文旨在重新分享討論Netty官方相關(guān)案例，添加部分個人理解與要點解析矿瘦。
這個是InChat的案例地址篡悟，里面補充了詳細的注釋劫樟，比起官方會容易看一點也糊。
官方案例地址：https://netty.io/4.1/xref/io/netty/example/uptime/package-summary.html
正文
- UptimeClient（客戶端）
- UptimeClientHandler
- UptimeServer（服務端）
- UptimeServerHandler
要點介紹
一個對Channel尚未執(zhí)行讀、寫或兩次操作的觸發(fā)器
| 屬性 | 含義 |
|---|---|
| readerIdleTime | 在IdleStateEvent其狀態(tài)IdleState.READER_IDLE 時的指定時間段沒有執(zhí)行讀操作將被觸發(fā)逮壁。指定0禁用。 |
| writerIdleTime | 在IdleStateEvent其狀態(tài)IdleState.WRITER_IDLE 時的指定時間段沒有執(zhí)行寫操作將被觸發(fā)粮宛。指定0禁用窥淆。 |
| allIdleTime | 一個IdleStateEvent其狀態(tài)IdleState.ALL_IDLE 時的時間在規(guī)定的時間進行讀取和寫入都將被觸發(fā)。指定0禁用窟勃。 |
如下一個在沒有信息時發(fā)送ping消息祖乳，且30秒沒有入站信息則關(guān)閉連接
public class MyChannelInitializer extends ChannelInitializer<Channel> {
@Override
public void initChannel(Channel channel) {
channel.pipeline().addLast("idleStateHandler", new IdleStateHandler(60, 30, 0));
channel.pipeline().addLast("myHandler", new MyHandler());
}
}
// Handler should handle the IdleStateEvent triggered by IdleStateHandler.
public class MyHandler extends ChannelDuplexHandler {
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
if (evt instanceof IdleStateEvent) {
IdleStateEvent e = (IdleStateEvent) evt;
if (e.state() == IdleState.READER_IDLE) {
ctx.close();
} else if (e.state() == IdleState.WRITER_IDLE) {
ctx.writeAndFlush(new PingMessage());
}
}
}
}
項目源碼
- UptimeClient
/**
* Created by MySelf on 2019/8/27.
*/
public final class UptimeClient {
static final String HOST = System.getProperty("host", "127.0.0.1");
static final int PORT = Integer.parseInt(System.getProperty("port", "8080"));
// 重新連接前睡眠5秒
static final int RECONNECT_DELAY = Integer.parseInt(System.getProperty("reconnectDelay", "5"));
// 當服務器在 10 秒內(nèi)不發(fā)送任何內(nèi)容時重新連接。
private static final int READ_TIMEOUT = Integer.parseInt(System.getProperty("readTimeout", "10"));
private static final UptimeClientHandler handler = new UptimeClientHandler();
private static final Bootstrap bs = new Bootstrap();
public static void main(String[] args) throws Exception {
EventLoopGroup group = new NioEventLoopGroup();
bs.group(group)
.channel(NioSocketChannel.class)
.remoteAddress(HOST,PORT)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new IdleStateHandler(READ_TIMEOUT,0,0),handler);
}
});
bs.connect();
}
static void connect(){
bs.connect().addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
if (future.cause() != null){
handler.startTime = -1;
handler.println("Failed to connect:" + future.cause());
}
}
});
}
}
- UptimeClientHandler
/**
* Created by MySelf on 2019/8/27.
*/
@ChannelHandler.Sharable
public class UptimeClientHandler extends SimpleChannelInboundHandler<Object> {
long startTime = -1;
@Override
protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
//Discard received data
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
if (startTime < 0){
startTime = System.currentTimeMillis();
}
println("Connected to:" + ctx.channel().remoteAddress());
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
println("Disconnected from: " + ctx.channel().remoteAddress());
}
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
if (!(evt instanceof IdleStateEvent)){
return;
}
IdleStateEvent e = (IdleStateEvent)evt;
if (e.state() == IdleState.READER_IDLE){
// 連接正常秉氧，但是沒有讀信息眷昆，關(guān)閉連接
println("Disconnecting due to no inbound traffic");
ctx.close();
}
}
@Override
public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
// 睡眠5秒
println("Sleeping for:" + UptimeClient.RECONNECT_DELAY + 's');
// 啟動線程重新連接
ctx.channel().eventLoop().schedule(new Runnable() {
@Override
public void run() {
println("Reconnecting to:" + UptimeClient.HOST + ":" + UptimeClient.PORT);
UptimeClient.connect();
}
},UptimeClient.RECONNECT_DELAY, TimeUnit.SECONDS);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
void println(String msg){
if (startTime < 0){
System.err.format("[SERVER IS DOWN] %s%n",msg);
} else {
System.err.format("[UPTIME: %5ds] %s%n",(System.currentTimeMillis() - startTime)/1000,msg);
}
}
}
- UptimeServer
/**
* Created by MySelf on 2019/8/27.
*/
public final class UptimeServer {
private static final int PORT = Integer.parseInt(System.getProperty("port", "8080"));
private static final UptimeServerHandler handler = new UptimeServerHandler();
private UptimeServer(){}
public static void main(String[] args) throws Exception {
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup,workerGroup)
.channel(NioServerSocketChannel.class)
.handler(new LoggingHandler(LogLevel.INFO))
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(handler);
}
});
// Bind and start to accept incoming connections.
ChannelFuture f = b.bind(PORT).sync();
// Wait until the server socket is closed.
// In this example, this does not happen, but you can do that to gracefully
// shut down your server.
f.channel().closeFuture().sync();
}finally {
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}
}
}
- UptimeServerHandler
/**
* Created by MySelf on 2019/8/27.
*/
@ChannelHandler.Sharable
public class UptimeServerHandler extends SimpleChannelInboundHandler<Object> {
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
// Close the connection when an exception is raised.
cause.printStackTrace();
ctx.close();
}
@Override
protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
// discard
}
}
