查看原文
其他

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

crossoverJie crossoverJie 2019-03-01

前言

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

最终能达到的效果:

  • 客户端每隔 N 秒检测是否需要发送心跳。

  • 服务端也每隔 N 秒检测是否需要发送心跳。

  • 服务端可以主动 push 消息到客户端。

  • 基于 SpringBoot 监控,可以查看实时连接以及各种应用信息。

效果如下:

IdleStateHandler

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

客户端心跳

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

核心代码代码如下:

  1. public class EchoClientHandle extends SimpleChannelInboundHandler<ByteBuf> {

  2.    private final static Logger LOGGER = LoggerFactory.getLogger(EchoClientHandle.class);

  3.    @Override

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

  5.        if (evt instanceof IdleStateEvent){

  6.            IdleStateEvent idleStateEvent = (IdleStateEvent) evt ;

  7.            if (idleStateEvent.state() == IdleState.WRITER_IDLE){

  8.                LOGGER.info("已经 10 秒没有发送信息!");

  9.                //向服务端发送消息

  10.                CustomProtocol heartBeat = SpringBeanFactory.getBean("heartBeat", CustomProtocol.class);

  11.                ctx.writeAndFlush(heartBeat).addListener(ChannelFutureListener.CLOSE_ON_FAILURE) ;

  12.            }

  13.        }

  14.        super.userEventTriggered(ctx, evt);

  15.    }

  16.    @Override

  17.    protected void channelRead0(ChannelHandlerContext channelHandlerContext, ByteBuf in) throws Exception {

  18.        //从服务端收到消息时被调用

  19.        LOGGER.info("客户端收到消息={}",in.toString(CharsetUtil.UTF_8)) ;

  20.    }

  21. }    

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

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

  1. @Configuration

  2. public class HeartBeatConfig {

  3.    @Value("${channel.id}")

  4.    private long id ;

  5.    @Bean(value = "heartBeat")

  6.    public CustomProtocol heartBeat(){

  7.        return new CustomProtocol(id,"ping") ;

  8.    }

  9. }

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

当然少不了启动引导:

  1. @Component

  2. public class HeartbeatClient {

  3.    private final static Logger LOGGER = LoggerFactory.getLogger(HeartbeatClient.class);

  4.    private EventLoopGroup group = new NioEventLoopGroup();

  5.    @Value("${netty.server.port}")

  6.    private int nettyPort;

  7.    @Value("${netty.server.host}")

  8.    private String host;

  9.    private SocketChannel channel;

  10.    @PostConstruct

  11.    public void start() throws InterruptedException {

  12.        Bootstrap bootstrap = new Bootstrap();

  13.        bootstrap.group(group)

  14.                .channel(NioSocketChannel.class)

  15.                .handler(new CustomerHandleInitializer())

  16.        ;

  17.        ChannelFuture future = bootstrap.connect(host, nettyPort).sync();

  18.        if (future.isSuccess()) {

  19.            LOGGER.info("启动 Netty 成功");

  20.        }

  21.        channel = (SocketChannel) future.channel();

  22.    }

  23. }

  24. public class CustomerHandleInitializer extends ChannelInitializer<Channel> {

  25.    @Override

  26.    protected void initChannel(Channel ch) throws Exception {

  27.        ch.pipeline()

  28.                //10 秒没发送消息 将IdleStateHandler 添加到 ChannelPipeline 中

  29.                .addLast(new IdleStateHandler(0, 10, 0))

  30.                .addLast(new HeartbeatEncode())

  31.                .addLast(new EchoClientHandle())

  32.        ;

  33.    }

  34. }    

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

服务端心跳

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

  1. public class HeartBeatSimpleHandle extends SimpleChannelInboundHandler<CustomProtocol> {

  2.    private final static Logger LOGGER = LoggerFactory.getLogger(HeartBeatSimpleHandle.class);

  3.    private static final ByteBuf HEART_BEAT =  Unpooled.unreleasableBuffer(Unpooled.copiedBuffer(new CustomProtocol(123456L,"pong").toString(),CharsetUtil.UTF_8));

  4.    /**

  5.     * 取消绑定

  6.     * @param ctx

  7.     * @throws Exception

  8.     */

  9.    @Override

  10.    public void channelInactive(ChannelHandlerContext ctx) throws Exception {

  11.        NettySocketHolder.remove((NioSocketChannel) ctx.channel());

  12.    }

  13.    @Override

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

  15.        if (evt instanceof IdleStateEvent){

  16.            IdleStateEvent idleStateEvent = (IdleStateEvent) evt ;

  17.            if (idleStateEvent.state() == IdleState.READER_IDLE){

  18.                LOGGER.info("已经5秒没有收到信息!");

  19.                //向客户端发送消息

  20.                ctx.writeAndFlush(HEART_BEAT).addListener(ChannelFutureListener.CLOSE_ON_FAILURE) ;

  21.            }

  22.        }

  23.        super.userEventTriggered(ctx, evt);

  24.    }

  25.    @Override

  26.    protected void channelRead0(ChannelHandlerContext ctx, CustomProtocol customProtocol) throws Exception {

  27.        LOGGER.info("收到customProtocol={}", customProtocol);

  28.        //保存客户端与 Channel 之间的关系

  29.        NettySocketHolder.put(customProtocol.getId(),(NioSocketChannel)ctx.channel()) ;

  30.    }

  31. }

这里有点需要注意

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

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

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

  1. public class NettySocketHolder {

  2.    private static final Map<Long, NioSocketChannel> MAP = new ConcurrentHashMap<>(16);

  3.    public static void put(Long id, NioSocketChannel socketChannel) {

  4.        MAP.put(id, socketChannel);

  5.    }

  6.    public static NioSocketChannel get(Long id) {

  7.        return MAP.get(id);

  8.    }

  9.    public static Map<Long, NioSocketChannel> getMAP() {

  10.        return MAP;

  11.    }

  12.    public static void remove(NioSocketChannel nioSocketChannel) {

  13.        MAP.entrySet().stream().filter(entry -> entry.getValue() == nioSocketChannel).forEach(entry -> MAP.remove(entry.getKey()));

  14.    }

  15. }

启动引导程序:

  1. Component

  2. public class HeartBeatServer {

  3.    private final static Logger LOGGER = LoggerFactory.getLogger(HeartBeatServer.class);

  4.    private EventLoopGroup boss = new NioEventLoopGroup();

  5.    private EventLoopGroup work = new NioEventLoopGroup();

  6.    @Value("${netty.server.port}")

  7.    private int nettyPort;

  8.    /**

  9.     * 启动 Netty

  10.     *

  11.     * @return

  12.     * @throws InterruptedException

  13.     */

  14.    @PostConstruct

  15.    public void start() throws InterruptedException {

  16.        ServerBootstrap bootstrap = new ServerBootstrap()

  17.                .group(boss, work)

  18.                .channel(NioServerSocketChannel.class)

  19.                .localAddress(new InetSocketAddress(nettyPort))

  20.                //保持长连接

  21.                .childOption(ChannelOption.SO_KEEPALIVE, true)

  22.                .childHandler(new HeartbeatInitializer());

  23.        ChannelFuture future = bootstrap.bind().sync();

  24.        if (future.isSuccess()) {

  25.            LOGGER.info("启动 Netty 成功");

  26.        }

  27.    }

  28.    /**

  29.     * 销毁

  30.     */

  31.    @PreDestroy

  32.    public void destroy() {

  33.        boss.shutdownGracefully().syncUninterruptibly();

  34.        work.shutdownGracefully().syncUninterruptibly();

  35.        LOGGER.info("关闭 Netty 成功");

  36.    }

  37. }    

  38. public class HeartbeatInitializer extends ChannelInitializer<Channel> {

  39.    @Override

  40.    protected void initChannel(Channel ch) throws Exception {

  41.        ch.pipeline()

  42.                //五秒没有收到消息 将IdleStateHandler 添加到 ChannelPipeline 中

  43.                .addLast(new IdleStateHandler(5, 0, 0))

  44.                .addLast(new HeartbeatDecoder())

  45.                .addLast(new HeartBeatSimpleHandle());

  46.    }

  47. }

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

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

自定义协议

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

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

CustomProtocol:

  1. public class CustomProtocol implements Serializable{

  2.    private static final long serialVersionUID = 4671171056588401542L;

  3.    private long id ;

  4.    private String content ;

  5.    //省略 getter/setter

  6. }

客户端的编码器:

  1. public class HeartbeatEncode extends MessageToByteEncoder<CustomProtocol> {

  2.    @Override

  3.    protected void encode(ChannelHandlerContext ctx, CustomProtocol msg, ByteBuf out) throws Exception {

  4.        out.writeLong(msg.getId()) ;

  5.        out.writeBytes(msg.getContent().getBytes()) ;

  6.    }

  7. }

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

服务端的解码器:

  1. public class HeartbeatDecoder extends ByteToMessageDecoder {

  2.    @Override

  3.    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {

  4.        long id = in.readLong() ;

  5.        byte[] bytes = new byte[in.readableBytes()] ;

  6.        in.readBytes(bytes) ;

  7.        String content = new String(bytes) ;

  8.        CustomProtocol customProtocol = new CustomProtocol() ;

  9.        customProtocol.setId(id);

  10.        customProtocol.setContent(content) ;

  11.        out.add(customProtocol) ;

  12.    }

  13. }

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

实现原理

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

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

来看看它的源码:

首先是构造函数:

  1.    public IdleStateHandler(

  2.            int readerIdleTimeSeconds,

  3.            int writerIdleTimeSeconds,

  4.            int allIdleTimeSeconds) {

  5.        this(readerIdleTimeSeconds, writerIdleTimeSeconds, allIdleTimeSeconds,

  6.             TimeUnit.SECONDS);

  7.    }

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

  • readerIdleTimeSeconds:一段时间内没有数据读取

  • writerIdleTimeSeconds:一段时间内没有数据发送

  • allIdleTimeSeconds:以上两种满足其中一个即可

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

  1.    @Override

  2.    public void channelActive(ChannelHandlerContext ctx) throws Exception {

  3.        // This method will be invoked only if this handler was added

  4.        // before channelActive() event is fired.  If a user adds this handler

  5.        // after the channelActive() event, initialize() will be called by beforeAdd().

  6.        initialize(ctx);

  7.        super.channelActive(ctx);

  8.    }

  9.    private void initialize(ChannelHandlerContext ctx) {

  10.        // Avoid the case where destroy() is called before scheduling timeouts.

  11.        // See: https://github.com/netty/netty/issues/143

  12.        switch (state) {

  13.        case 1:

  14.        case 2:

  15.            return;

  16.        }

  17.        state = 1;

  18.        initOutputChanged(ctx);

  19.        lastReadTime = lastWriteTime = ticksInNanos();

  20.        if (readerIdleTimeNanos > 0) {

  21.            readerIdleTimeout = schedule(ctx, new ReaderIdleTimeoutTask(ctx),

  22.                    readerIdleTimeNanos, TimeUnit.NANOSECONDS);

  23.        }

  24.        if (writerIdleTimeNanos > 0) {

  25.            writerIdleTimeout = schedule(ctx, new WriterIdleTimeoutTask(ctx),

  26.                    writerIdleTimeNanos, TimeUnit.NANOSECONDS);

  27.        }

  28.        if (allIdleTimeNanos > 0) {

  29.            allIdleTimeout = schedule(ctx, new AllIdleTimeoutTask(ctx),

  30.                    allIdleTimeNanos, TimeUnit.NANOSECONDS);

  31.        }

  32.    }    

也就是会按照我们给定的时间初始化出定时任务。

接着在任务真正执行时进行判断:

  1.    private final class ReaderIdleTimeoutTask extends AbstractIdleTask {

  2.        ReaderIdleTimeoutTask(ChannelHandlerContext ctx) {

  3.            super(ctx);

  4.        }

  5.        @Override

  6.        protected void run(ChannelHandlerContext ctx) {

  7.            long nextDelay = readerIdleTimeNanos;

  8.            if (!reading) {

  9.                nextDelay -= ticksInNanos() - lastReadTime;

  10.            }

  11.            if (nextDelay <= 0) {

  12.                // Reader is idle - set a new timeout and notify the callback.

  13.                readerIdleTimeout = schedule(ctx, this, readerIdleTimeNanos, TimeUnit.NANOSECONDS);

  14.                boolean first = firstReaderIdleEvent;

  15.                firstReaderIdleEvent = false;

  16.                try {

  17.                    IdleStateEvent event = newIdleStateEvent(IdleState.READER_IDLE, first);

  18.                    channelIdle(ctx, event);

  19.                } catch (Throwable t) {

  20.                    ctx.fireExceptionCaught(t);

  21.                }

  22.            } else {

  23.                // Read occurred before the timeout - set a new timeout with shorter delay.

  24.                readerIdleTimeout = schedule(ctx, this, nextDelay, TimeUnit.NANOSECONDS);

  25.            }

  26.        }

  27.    }

如果满足条件则会生成一个 IdleStateEvent 事件。

SpringBoot 监控

由于整合了 SpringBoot 之后不但可以利用 Spring 帮我们管理对象,也可以利用它来做应用监控。

actuator 监控

当我们为引入了:

  1.        <dependency>

  2.            <groupId>org.springframework.boot</groupId>

  3.            <artifactId>spring-boot-starter-actuator</artifactId>

  4.        </dependency>

就开启了 SpringBoot 的 actuator 监控功能,他可以暴露出很多监控端点供我们使用。

如一些应用中的一些统计数据:

存在的 Beans:

更多信息请查看:https://docs.spring.io/spring-boot/docs/current/reference/html/production-ready-endpoints.html

但是如果我想监控现在我的服务端有多少客户端连上来了,分别的 ID 是多少?

其实就是实时查看我内部定义的那个关联关系的 Map。

这就需要暴露自定义端点了。

自定义端点

暴露的方式也很简单:

继承 AbstractEndpoint 并复写其中的 invoke 函数:

  1. public class CustomEndpoint extends AbstractEndpoint<Map<Long,NioSocketChannel>> {

  2.    /**

  3.     * 监控端点的 访问地址

  4.     * @param id

  5.     */

  6.    public CustomEndpoint(String id) {

  7.        //false 表示不是敏感端点

  8.        super(id, false);

  9.    }

  10.    @Override

  11.    public Map<Long, NioSocketChannel> invoke() {

  12.        return NettySocketHolder.getMAP();

  13.    }

  14. }

其实就是返回了 Map 中的数据。

再配置一个该类型的 Bean 即可:

  1. @Configuration

  2. public class EndPointConfig {

  3.    @Value("${monitor.channel.map.key}")

  4.    private String channelMap;

  5.    @Bean

  6.    public CustomEndpoint buildEndPoint(){

  7.        CustomEndpoint customEndpoint = new CustomEndpoint(channelMap) ;

  8.        return customEndpoint ;

  9.    }

  10. }

这样我们就可以通过配置文件中的 monitor.channel.map.key 来访问了:

一个客户端连接时:

两个客户端连接时:

整合 SBA

这样其实监控功能已经可以满足了,但能不能展示的更美观、并且多个应用也可以方便查看呢?

有这样的开源工具帮我们做到了:

https://github.com/codecentric/spring-boot-admin

简单来说我们可以利用该工具将 actuator 暴露出来的接口可视化并聚合的展示在页面中:

接入也很简单,首先需要引入依赖:

  1.        <dependency>

  2.            <groupId>de.codecentric</groupId>

  3.            <artifactId>spring-boot-admin-starter-client</artifactId>

  4.        </dependency>        

并在配置文件中加入:

  1. # 关闭健康检查权限

  2. management.security.enabled=false

  3. # SpringAdmin 地址

  4. spring.boot.admin.url=http://127.0.0.1:8888

在启动应用之前先讲 SpringBootAdmin 部署好:

这个应用就是一个纯粹的 SpringBoot ,只需要在主函数上加入 @EnableAdminServer 注解。

  1. @SpringBootApplication

  2. @Configuration

  3. @EnableAutoConfiguration

  4. @EnableAdminServer

  5. public class AdminApplication {

  6.    public static void main(String[] args) {

  7.        SpringApplication.run(AdminApplication.class, args);

  8.    }

  9. }

引入:

  1.        <dependency>

  2.            <groupId>de.codecentric</groupId>

  3.            <artifactId>spring-boot-admin-starter-server</artifactId>

  4.            <version>1.5.7</version>

  5.        </dependency>

  6.        <dependency>

  7.            <groupId>de.codecentric</groupId>

  8.            <artifactId>spring-boot-admin-server-ui</artifactId>

  9.            <version>1.5.6</version>

  10.        </dependency>

之后直接启动就行了。

这样我们在 SpringBootAdmin 的页面中就可以查看很多应用信息了。

更多内容请参考官方指南:

http://codecentric.github.io/spring-boot-admin/1.5.6/

自定义监控数据

其实我们完全可以借助 actuator 以及这个可视化页面帮我们监控一些简单的度量信息。

比如我在客户端和服务端中写了两个 Rest 接口用于向对方发送消息。

只是想要记录分别发送了多少次:

客户端:

  1. @Controller

  2. @RequestMapping("/")

  3. public class IndexController {

  4.    /**

  5.     * 统计 service

  6.     */

  7.    @Autowired

  8.    private CounterService counterService;

  9.    @Autowired

  10.    private HeartbeatClient heartbeatClient ;

  11.    /**

  12.     * 向服务端发消息

  13.     * @param sendMsgReqVO

  14.     * @return

  15.     */

  16.    @ApiOperation("客户端发送消息")

  17.    @RequestMapping("sendMsg")

  18.    @ResponseBody

  19.    public BaseResponse<SendMsgResVO> sendMsg(@RequestBody SendMsgReqVO sendMsgReqVO){

  20.        BaseResponse<SendMsgResVO> res = new BaseResponse();

  21.        heartbeatClient.sendMsg(new CustomProtocol(sendMsgReqVO.getId(),sendMsgReqVO.getMsg())) ;

  22.        // 利用 actuator 来自增

  23.        counterService.increment(Constants.COUNTER_CLIENT_PUSH_COUNT);

  24.        SendMsgResVO sendMsgResVO = new SendMsgResVO() ;

  25.        sendMsgResVO.setMsg("OK") ;

  26.        res.setCode(StatusEnum.SUCCESS.getCode()) ;

  27.        res.setMessage(StatusEnum.SUCCESS.getMessage()) ;

  28.        res.setDataBody(sendMsgResVO) ;

  29.        return res ;

  30.    }

  31. }

只要我们引入了 actuator 的包,那就可以直接注入 counterService ,利用它来帮我们记录数据。

当我们调用该接口时:

在监控页面中可以查询刚才的调用情况:

服务端主动 push 消息也是类似,只是需要在发送时候根据客户端的 ID 查询到具体的 Channel 发送:

总结

以上就是一个简单 Netty 心跳示例,并演示了 SpringBoot 的监控,之后会继续更新 Netty 相关内容,欢迎关注及指正。

本文所有代码:

https://github.com/crossoverJie/netty-action

号外

最近在总结一些 Java 相关的知识点,感兴趣的朋友可以一起维护。

地址: https://github.com/crossoverJie/Java-Interview


    您可能也对以下帖子感兴趣

    文章有问题?点此查看未经处理的缓存