Netty 实现心跳发送机制

Netty 实现心跳发送机制

处理空闲连接是一项常见的任务，为了能够及时的将资源释放出来，需要在服务端检测空闲连接和超时，常见的方法是通过发送信息来测试一个不活跃的链接，通常被称为“心跳”，然后在远端确认它是否还活着，Netty 提供了 IdleStateHandler 处理这个过程；
以下是一个示例的实现控制器，在服务端每间隔 5s（设置为 5s 仅仅只是为了测试方便，实际开发中可以设置一个更长的事件间隔）向客户端发送一个心跳包，同时添加一个监听器，如果客户端响应该心跳包，证明该客户端连接是有效的，如果该客户端没有响应，则说明该客户端连接已经失效了，此时监听器获取该事件，并关服务端连接；
完整的示例地址（包含示例的客户端）：https://gitee.com/assad/netty-test-sample/tree/master/netty-test-sample/src/main/java/heartbeatSample
以下为服务端心跳包控制器的实现，IdleStateHandlerInitializer
 

public class IdleStateHandlerInitializer extends ChannelInitializer<Channel>{

   @Override

   protected void initChannel(Channel ch) throws Exception {

       ChannelPipeline pipeline = ch.pipeline();

       //添加空闲状态处理器，当超过 5s 没有收到远端连接的信息时，该处理器会传播一个 IdleStateHandler

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

       //添加 IdleStateHandler 传播的 IdleStateEvent 的处理器

       pipeline.addLast(new HeartbeatHandler());

       //添加普通入站处理器

       pipeline.addLast(new EchoServerHandler());

   }

​

​

   //IdleEvent 事件处理器

   public static final class HeartbeatHandler extends ChannelInboundHandlerAdapter{

​

       private static final ByteBuf HEARTBEAT_SEQUENCE = Unpooled.unreleasableBuffer(

               Unpooled.copiedBuffer("HEARTBEAT", Charset.forName("UTF-8")) );

​

       @Override

       public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {

           if(evt instanceof IdleStateEvent){

               //当捕获到 IdleStateEvent，发送心跳到远端，并添加一个监听器，如果发送失败就关闭服务端连接

               ctx.writeAndFlush(HEARTBEAT_SEQUENCE.duplicate()).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);

           }else{

               //如果捕获到的时间不是一个 IdleStateEvent，就将该事件传递到下一个处理器

               super.userEventTriggered(ctx,evt);

           }

       }

   }

​

   @Sharable

   //普通入站处理器:打印可客户端发送的信息

   public class EchoServerHandler extends ChannelInboundHandlerAdapter{

       @Override

       public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

           ByteBuf in = (ByteBuf) msg;

           System.out.println("Server received: " + in.toString(CharsetUtil.UTF_8));

       }

       @Override

       public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {

           cause.printStackTrace();

           ctx.close();

       }

   }

​

}