netty对keepalive和idle的支持

骑猪看日落 2022-12-26 10:14 203阅读 0赞

# netty对keepalive和idle的支持 #

## 为什么需要keepalive ##

keepalive就是心跳，一个人的心跳证明人还活着，那么在网络通信的双方如何证明对端还活着着，两个服务之间使用心跳来检测对方是否还活着。

为什么要检测对方是否还活着呢？假如客户端因为某种原因（停电宕机、终止运行）没有发送关闭连接的数据包，那么服务器就会一直维持着连接，占用服务器资源，后面需要使用连接的时候还会报错。有了心跳，服务器就能及时释放资源。

## TCP中的keepalive ##

TCP keepalive核心参数如下：

$ sysctl -a| grep tcp_keepalive
    net.ipv4.tcp_keepalive_intvl = 75
    net.ipv4.tcp_keepalive_probes = 9
    net.ipv4.tcp_keepalive_time = 7200

TCP在连接没有数据通过后的7200s(tcp\_keepalive\_time)后会发送keepalive消息，当消息没有被确认后，按75s（tcp\_keepalive\_intvl）的频率重新发送，一直发送9（tcp\_keepalive\_probes）个探测包都没有被确认，就认定这个连接失效了。

## 由了TCP的keepalive，为什么还需要应用层的keepalive ##

*  TCP中的keepalive默认是关闭，因为探测包可能在传递过程中会丢失（例如用了代理）。
 *  默认的超时时间太长，默认是7200+9\*75秒，也就是2个多小时。
 *  TCP是一个传输层的协议，传输层的数据畅通并不一定操作系统进程所对应的服务畅通。

HTTP中的Keep-Alive是指在HTTP的请求头部携带参数`Connection: Keep-Alive`，这样浏览器与服务器端就会保持一个长连接，HTTP1.1协议默认是长连接，可以不用携带这个参数。

## Idle检测 ##

Idle是空闲的意思，也就是当客户端不向服务器端发送数据了，不会立马发送心跳包，会等待一段时间，判断这个连接空闲时才会发送。

keepalive的设计思路：

1.  开启一个定时任务，不管客户端和服务器端有没有数据的传输，定时发送心跳包。
2.  在连接通道中有数据传送的时候不发送心跳包，无数据传送超过一定时间判定为空闲时再发送。

netty中使用的是第二种。

## netty中开启keepalive和idle ##

server端开启keepalive：

.childOption(ChannelOption.SO_KEEPALIVE, true)
    .childOption(NioChannelOption.SO_KEEPALIVE, true)

> option(ChannelOption.SO\_KEEPALIVE, true)存在但是设置无效。

server端开启idle：

ch.pipeline().addLast(new IdleStateHandler(0, 20, 0, TimeUnit.SECONDS));

IdleStateHandler参数说明：

*  readerIdleTime：读空闲时间，超过指定时间未读取数据就会触发`IdleState.READER_IDLE`事件。
 *  writerIdleTime：写空闲时间，超过指定时间未发送数据就会触发`IdleState.WRITER_IDLE`事件。
 *  allIdleTime：读或写空闲时间，超过指定时间未读取或者发送数据就会触发`IdleState.ALL_IDLE`事件。
 *  unit：时间单位。

## idle的使用 ##

服务器端的代码实现：

package com.morris.netty.keepalive;
    
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.*;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioChannelOption;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    import io.netty.handler.codec.LineBasedFrameDecoder;
    import io.netty.handler.codec.string.StringDecoder;
    import io.netty.handler.codec.string.StringEncoder;
    import io.netty.handler.logging.LoggingHandler;
    import io.netty.handler.timeout.IdleState;
    import io.netty.handler.timeout.IdleStateEvent;
    import io.netty.handler.timeout.IdleStateHandler;
    import lombok.extern.slf4j.Slf4j;
    
    import java.net.StandardSocketOptions;
    import java.util.concurrent.TimeUnit;
    
    @Slf4j
    public class Server { 
    
        public static final int PORT = 8899;
    
        public static void main(String[] args) throws InterruptedException { 
    
            EventLoopGroup bossGroup = new NioEventLoopGroup();
            EventLoopGroup workerGroup = new NioEventLoopGroup();
    
            try { 
                ServerBootstrap b = new ServerBootstrap();
                b.group(bossGroup, workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .childOption(ChannelOption.SO_KEEPALIVE, true)
                        .childOption(NioChannelOption.of(StandardSocketOptions.SO_KEEPALIVE), true)
                        .childHandler(new ChannelInitializer<SocketChannel>() { 
                            @Override
                            public void initChannel(SocketChannel ch) throws Exception { 
                                ch.pipeline().addLast(new LoggingHandler());
                                ch.pipeline().addLast(new IdleStateHandler(60, 0, 0, TimeUnit.SECONDS));
                                ch.pipeline().addLast(new LineBasedFrameDecoder(1 << 10));
                                ch.pipeline().addLast(new StringDecoder());
                                ch.pipeline().addLast(new StringEncoder());
                                ch.pipeline().addLast(new SimpleChannelInboundHandler<String>() { 
                                    @Override
                                    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception { 
                                        log.info("receive from client: {}", msg);
                                        ctx.writeAndFlush("ok\n");
                                    }
    
                                    @Override
                                    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { 
                                        if(evt instanceof IdleStateEvent) { 
                                            IdleStateEvent idleStateEvent = (IdleStateEvent) evt;
                                            if(idleStateEvent.state() == IdleState.READER_IDLE) { 
                                                // 60s未收到数据就会关闭连接
                                                log.warn("timeout: {}", ctx.channel().remoteAddress());
                                                ctx.channel().close();
                                            }
                                        } else { 
                                            super.userEventTriggered(ctx, evt);
                                        }
                                    }
    
                                    @Override
                                    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { 
                                        cause.printStackTrace();
                                        ctx.close();
                                    }
                                });
                            }
                        });
    
                // 启动 server.
                ChannelFuture f = b.bind(PORT).sync();
    
                System.out.println("server is start on port: " + PORT);
    
                // 等待socket关闭
                f.channel().closeFuture().sync();
            } finally { 
                workerGroup.shutdownGracefully();
                bossGroup.shutdownGracefully();
            }
        }
    
    }

服务器端开启读空闲监测，60s未收到数据就会关闭连接。

客户端代码的实现：

package com.morris.netty.keepalive;
    
    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.LineBasedFrameDecoder;
    import io.netty.handler.codec.string.StringDecoder;
    import io.netty.handler.codec.string.StringEncoder;
    import io.netty.handler.logging.LoggingHandler;
    import io.netty.handler.timeout.IdleState;
    import io.netty.handler.timeout.IdleStateEvent;
    import io.netty.handler.timeout.IdleStateHandler;
    import lombok.extern.slf4j.Slf4j;
    
    import java.sql.Time;
    import java.util.concurrent.TimeUnit;
    
    @Slf4j
    public class Client { 
    
        public static void main(String[] args) throws InterruptedException { 
    
            EventLoopGroup workerGroup = new NioEventLoopGroup();
    
            try { 
                Bootstrap b = new Bootstrap();
                b.group(workerGroup)
                        .channel(NioSocketChannel.class)
                        .handler(new ChannelInitializer<SocketChannel>() { 
                            @Override
                            public void initChannel(SocketChannel ch) throws Exception { 
                                ch.pipeline().addLast(new LoggingHandler());
                                ch.pipeline().addLast(new IdleStateHandler(0, 30, 0, TimeUnit.SECONDS));
                                ch.pipeline().addLast(new LineBasedFrameDecoder(1 << 10));
                                ch.pipeline().addLast(new StringEncoder());
                                ch.pipeline().addLast(new StringDecoder());
                                ch.pipeline().addLast(new SimpleChannelInboundHandler<String>() { 
                                    @Override
                                    public void channelActive(ChannelHandlerContext ctx) throws Exception { 
                                        ctx.writeAndFlush("hello\n");
                                    }
    
                                    @Override
                                    public void channelRead0(ChannelHandlerContext ctx, String msg) { 
                                        log.info("receive from server: {}", msg);
                                    }
    
                                    @Override
                                    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { 
                                        if(evt instanceof IdleStateEvent) { 
                                            IdleStateEvent idleStateEvent = (IdleStateEvent) evt;
                                            if(idleStateEvent.state() == IdleState.WRITER_IDLE) { 
                                                // 30s未写数据就会发送心跳
                                                ctx.writeAndFlush("hi\n");
                                            }
                                        } else { 
                                            super.userEventTriggered(ctx, evt);
                                        }
                                    }
    
                                    @Override
                                    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { 
                                        cause.printStackTrace();
                                        ctx.close();
                                    }
                                });
                            }
                        });
    
                // 启动 server.
                ChannelFuture f = b.connect("127.0.0.1", 8899).sync();
    
                // 等待socket关闭
                f.channel().closeFuture().sync();
            } finally { 
                workerGroup.shutdownGracefully();
            }
        }
    
    }

客户端开启写空闲监测，30s未写数据就会发送心跳。

当然正常情况下本地测试上面的代码，服务器端是不会触发读空闲的事件，即使强制关闭了客户端，客户端也会发送关闭连接的请求给服务器，然后服务器端将连接关闭。

服务器端要想触发读空闲的事件，可以在使用两台机器或者虚拟机来测试，客户端启动后直接把网线拔出，如果是在linux下可以使用下面的命令关闭网络来测试：

# ifconfig ens32 down // 关闭网卡
    # ifconfig ens32 up // 开启网卡