阅读本文约 "4 分钟"
适读人群: Java-Netty 初级
Netty 自动重连机制
版本: netty 4.1.*
申明: 本文旨在重新分享讨论 Netty 官方相关案例, 添加部分个人理解与要点解析.
这个是 InChat 的案例 地址 , 里面补充了详细的注释, 比起官方会容易看一点.
官方案例地址: https://netty.io/4.1/xref/io/netty/example/uptime/package-summary.html
正文
- UptimeClient(客户端)
- UptimeClientHandler
- UptimeServer(服务端)
- UptimeServerHandler
要点介绍
IdleStateHandler https://netty.io/4.1/api/io/netty/handler/timeout/IdleStateHandler.html
一个对 Channel 尚未执行读, 写或两次操作的触发器
属性 | 含义 |
---|---|
readerIdleTime | 在 IdleStateEvent 其状态 IdleState.READER_IDLE 时的指定时间段没有执行读操作将被触发。指定 0 禁用。 |
writerIdleTime | 在 IdleStateEvent 其状态 IdleState.WRITER_IDLE 时的指定时间段没有执行写操作将被触发。指定 0 禁用。 |
allIdleTime | 一个 IdleStateEvent 其状态 IdleState.ALL_IDLE 时的时间在规定的时间进行读取和写入都将被触发。指定 0 禁用。 |
如下一个在没有信息时发送 ping 消息, 且 30 秒没有入站信息则关闭连接
- 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 秒内不发送任何内容时重新连接.
- 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){
- // 连接正常, 但是没有读信息, 关闭连接
- 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: ]s] %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
- }
- }
- Previous
基于 InChat 的 SpringBoot 版本通讯聊天数据存储 Demo, 附带详细流程说明 [InChat1.1.4]
来源: http://www.tuicool.com/articles/VbEJnqf