专栏名称: 狗厂
目录
相关文章推荐
51好读  ›  专栏  ›  狗厂

Netty(一) SpringBoot 整合长连接心跳机制

狗厂  · 掘金  ·  · 2018-05-29 03:35

正文

photo-1522204657746-fccce0824cfd.jpeg

前言

Netty 是一个高性能的 NIO 网络框架，本文基于 SpringBoot 以常见的心跳机制来认识 Netty。

最终能达到的效果：

  • 客户端每隔 N 秒检测是否需要发送心跳。
  • 服务端也每隔 N 秒检测是否需要发送心跳。
  • 服务端可以主动 push 消息到客户端。
  • 基于 SpringBoot 监控，可以查看实时连接以及各种应用信息。

效果如下：

show

IdleStateHandler

Netty 可以使用 IdleStateHandler 来实现连接管理，当连接空闲时间太长（没有发送、接收消息）时则会触发一个事件，我们便可在该事件中实现心跳机制。

客户端心跳

当客户端空闲了 N 秒没有给服务端发送消息时会自动发送一个心跳来维持连接。

核心代码代码如下： 

public class EchoClientHandle extends SimpleChannelInboundHandler<ByteBuf> {
    private final static Logger LOGGER = LoggerFactory.getLogger(EchoClientHandle.class);
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent){
            IdleStateEvent idleStateEvent = (IdleStateEvent) evt ;
            if (idleStateEvent.state() == IdleState.WRITER_IDLE){
                LOGGER.info("已经 10 秒没有发送信息！");
                //向服务端发送消息
                CustomProtocol heartBeat = SpringBeanFactory.getBean("heartBeat", CustomProtocol.class);
                ctx.writeAndFlush(heartBeat).addListener(ChannelFutureListener.CLOSE_ON_FAILURE) ;
            }
        }
        super.userEventTriggered(ctx, evt);
    }
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, ByteBuf in) throws Exception {
        //从服务端收到消息时被调用
        LOGGER.info("客户端收到消息={}",in.toString(CharsetUtil.UTF_8)) ;
    }
}

实现非常简单，只需要在事件回调中发送一个消息即可。

由于整合了 SpringBoot ，所以发送的心跳信息是一个单例的 Bean。 

@Configuration
public class HeartBeatConfig {
    @Value("${channel.id}")
    private long id ;
    @Bean(value = "heartBeat")
    public CustomProtocol heartBeat(){
        return new CustomProtocol(id,"ping") ;
    }
}

这里涉及到了自定义协议的内容，请继续查看下文。

当然少不了启动引导： 

@Component
public class HeartbeatClient {
    private final static Logger LOGGER = LoggerFactory.getLogger(HeartbeatClient.class);
    private EventLoopGroup group = new NioEventLoopGroup();
    @Value("${netty.server.port}")
    private int nettyPort;
    @Value("${netty.server.host}")
    private String host;
    private SocketChannel channel;
    @PostConstruct
    public void start() throws InterruptedException {
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .handler(new CustomerHandleInitializer())
        ;
        ChannelFuture future = bootstrap.connect(host, nettyPort).sync();
        if (future.isSuccess()) {
            LOGGER.info("启动 Netty 成功");
        }
        channel = (SocketChannel) future.channel();
    }
    
}
public class CustomerHandleInitializer extends ChannelInitializer<Channel> {
    @Override
    protected void initChannel(Channel ch) throws Exception {
        ch.pipeline()
                //10 秒没发送消息 将IdleStateHandler 添加到 ChannelPipeline 中
                .addLast(new IdleStateHandler(0, 10, 0))
                .addLast(new HeartbeatEncode())
                .addLast(new EchoClientHandle())
        ;
    }
}

所以当应用启动每隔 10 秒会检测是否发送过消息，不然就会发送心跳信息。

服务端心跳

服务器端的心跳其实也是类似，也需要在 ChannelPipeline 中添加一个 IdleStateHandler 。 

public class HeartBeatSimpleHandle extends SimpleChannelInboundHandler<CustomProtocol> {
    private final static Logger LOGGER = LoggerFactory.getLogger(HeartBeatSimpleHandle.class);
    private static final ByteBuf HEART_BEAT =  Unpooled.unreleasableBuffer(Unpooled.copiedBuffer(new CustomProtocol(123456L,"pong").toString(),CharsetUtil.UTF_8));
    /**
     * 取消绑定
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        NettySocketHolder.remove((NioSocketChannel) ctx.channel());
    }
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent){
            IdleStateEvent idleStateEvent = (IdleStateEvent) evt ;
            if (idleStateEvent.state() == IdleState.READER_IDLE){
                LOGGER.info("已经5秒没有收到信息！");
                //向客户端发送消息
                ctx.writeAndFlush(HEART_BEAT).addListener(ChannelFutureListener.CLOSE_ON_FAILURE) ;
            }
        }
        super.userEventTriggered(ctx, evt);
    }
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, CustomProtocol customProtocol) throws Exception {
        LOGGER.info("收到customProtocol={}", customProtocol);
        //保存客户端与 Channel 之间的关系
        NettySocketHolder.put(customProtocol.getId(),(NioSocketChannel)ctx.channel()) ;
    }
}

这里有点需要注意 ：

当有多个客户端连上来时，服务端需要区分开，不然响应消息就会发生混乱。

所以每当有个连接上来的时候，我们都将当前的 Channel 与连上的客户端 ID 进行关联（ 因此每个连上的客户端 ID 都必须唯一 ）。

这里采用了一个 Map 来保存这个关系，并且在断开连接时自动取消这个关联。 

public class NettySocketHolder {
    private static final Map<Long, NioSocketChannel> MAP = new ConcurrentHashMap<>(16);
    public static void put(Long id, NioSocketChannel socketChannel) {
        MAP.put(id, socketChannel);
    }
    public static NioSocketChannel get(Long id) {
        return MAP.get(id);
    }
    public static Map<Long, NioSocketChannel> getMAP() {
        return MAP;
    }
    public static void remove(NioSocketChannel nioSocketChannel) {
        MAP.entrySet().stream().filter(entry -> entry.getValue() == nioSocketChannel).forEach(entry -> MAP.remove(entry.getKey()));
    }
}

启动引导程序： 

Component
public class HeartBeatServer {
    private final static Logger LOGGER = LoggerFactory.getLogger(HeartBeatServer.class);
    private EventLoopGroup boss = new NioEventLoopGroup();
    private EventLoopGroup work = new NioEventLoopGroup();
    @Value("${netty.server.port}")
    private int nettyPort;
    /**
     * 启动 Netty
     *
     * @return
     * @throws InterruptedException
     */
    @PostConstruct
    public void start() throws InterruptedException {
        ServerBootstrap bootstrap = new ServerBootstrap()
                .group(boss, work)
                .channel(NioServerSocketChannel.class)
                .localAddress(new InetSocketAddress(nettyPort))
                //保持长连接
                .childOption(ChannelOption.SO_KEEPALIVE, true)
                .childHandler(new HeartbeatInitializer());
        ChannelFuture future = bootstrap.bind().sync();
        if (future.isSuccess()) {
            LOGGER.info("启动 Netty 成功");
        }
    }
    /**
     * 销毁
     */
    @PreDestroy
    public void destroy() {
        boss.shutdownGracefully().syncUninterruptibly();
        work.shutdownGracefully().syncUninterruptibly();
        LOGGER.info("关闭 Netty 成功");
    }
}    
public class HeartbeatInitializer extends ChannelInitializer<Channel> {
    @Override
    protected void initChannel(Channel ch) throws Exception {
        ch.pipeline()
                //五秒没有收到消息 将IdleStateHandler 添加到 ChannelPipeline 中
                .addLast(new IdleStateHandler(5, 0, 0))
                .addLast(new HeartbeatDecoder())
                .addLast(new HeartBeatSimpleHandle());
    }
}

也是同样将IdleStateHandler 添加到 ChannelPipeline 中，也会有一个定时任务，每5秒校验一次是否有收到消息，否则就主动发送一次请求。

因为测试是有两个客户端连上所以有两个日志。

自定义协议

上文其实都看到了：服务端与客户端采用的是自定义的 POJO 进行通讯的。

所以需要在客户端进行编码，服务端进行解码，也都只需要各自实现一个编解码器即可。

CustomProtocol： 

public class CustomProtocol implements Serializable{
    private static final long serialVersionUID = 4671171056588401542L;
    private long id ;
    private String content ;
    //省略 getter/setter
}

客户端的编码器： 

public class HeartbeatEncode extends MessageToByteEncoder<CustomProtocol> {
    @Override
    protected void encode(ChannelHandlerContext ctx, CustomProtocol msg, ByteBuf out) throws Exception {
        out.writeLong(msg.getId()) ;
        out.writeBytes(msg.getContent().getBytes()) ;
    }
}

也就是说消息的前八个字节为 header，剩余的全是 content。

服务端的解码器： 

public class HeartbeatDecoder extends ByteToMessageDecoder {
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        long id = in.readLong() ;
        byte[] bytes = new byte[in.readableBytes()] ;
        in.readBytes(bytes) ;
        String content = new String(bytes) ;
        CustomProtocol customProtocol = new CustomProtocol() ;
        customProtocol.setId(id);
        customProtocol.setContent(content) ;
        out.add(customProtocol) ;
    }
}

只需要按照刚才的规则进行解码即可。

实现原理

其实联想到 IdleStateHandler 的功能，自然也能想到它实现的原理：

应该会存在一个定时任务的线程去处理这些消息。

来看看它的源码：

首先是构造函数: 

public IdleStateHandler(
        int readerIdleTimeSeconds,
        int writerIdleTimeSeconds,
        int allIdleTimeSeconds) {
    this(readerIdleTimeSeconds, writerIdleTimeSeconds, allIdleTimeSeconds,
         TimeUnit.SECONDS);
}

其实就是初始化了几个数据：

  • readerIdleTimeSeconds：一段时间内没有数据读取
  • writerIdleTimeSeconds：一段时间内没有数据发送
  • allIdleTimeSeconds：以上两种满足其中一个即可

因为 IdleStateHandler 也是一种 ChannelHandler，所以会在 channelActive 中初始化任务：







请到「今天看啥」查看全文


推荐文章
奥斯CAR  ·  3月SUV销量榜出炉！H2被干掉，H6还能坚守吗？
7 年前
DearAuto  ·  全新Jeep指南者3月销量再次力压翼虎 第一季度复合月增长率同级第一
7 年前
生意社  ·  大宗商品数据每日播报（2017年6月6日）
7 年前
植物前沿  ·  200款·绝美露台小花园
7 年前
禅茶一味  ·  【茶语馆】买和省，都是态度
7 年前
Sov5搜索   ·   小百科   ·   今天看啥   ·   移动版
51好读 - 好文章就要读起来!