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

时间  2019-11-29
标签 netty springboot 整合 链接 心跳 机制 栏目 Netty 繁體版
原文   原文链接

前言

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

最终能达到的效果：java

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

效果以下：git

show

IdleStateHandler

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

客户端心跳

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

核心代码代码以下：bootstrap

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)) ;

    }
}    
复制代码

实现很是简单，只须要在事件回调中发送一个消息便可。bash

因为整合了 SpringBoot ，因此发送的心跳信息是一个单例的 Bean。服务器

@Configuration
public class HeartBeatConfig {

    @Value("${channel.id}")
    private long id ;


    @Bean(value = "heartBeat")
    public CustomProtocol heartBeat(){
        return new CustomProtocol(id,"ping") ;
    }
}
复制代码

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

固然少不了启动引导：app

@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 中初始化任务：

@Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        // This method will be invoked only if this handler was added
        // before channelActive() event is fired. If a user adds this handler
        // after the channelActive() event, initialize() will be called by beforeAdd().
        initialize(ctx);
        super.channelActive(ctx);
    }
    
    private void initialize(ChannelHandlerContext ctx) {
        // Avoid the case where destroy() is called before scheduling timeouts.
        // See: https://github.com/netty/netty/issues/143
        switch (state) {
        case 1:
        case 2:
            return;
        }

        state = 1;
        initOutputChanged(ctx);

        lastReadTime = lastWriteTime = ticksInNanos();
        if (readerIdleTimeNanos > 0) {
            readerIdleTimeout = schedule(ctx, new ReaderIdleTimeoutTask(ctx),
                    readerIdleTimeNanos, TimeUnit.NANOSECONDS);
        }
        if (writerIdleTimeNanos > 0) {
            writerIdleTimeout = schedule(ctx, new WriterIdleTimeoutTask(ctx),
                    writerIdleTimeNanos, TimeUnit.NANOSECONDS);
        }
        if (allIdleTimeNanos > 0) {
            allIdleTimeout = schedule(ctx, new AllIdleTimeoutTask(ctx),
                    allIdleTimeNanos, TimeUnit.NANOSECONDS);
        }
    }    
复制代码

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

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

private final class ReaderIdleTimeoutTask extends AbstractIdleTask {

        ReaderIdleTimeoutTask(ChannelHandlerContext ctx) {
            super(ctx);
        }

        @Override
        protected void run(ChannelHandlerContext ctx) {
            long nextDelay = readerIdleTimeNanos;
            if (!reading) {
                nextDelay -= ticksInNanos() - lastReadTime;
            }

            if (nextDelay <= 0) {
                // Reader is idle - set a new timeout and notify the callback.
                readerIdleTimeout = schedule(ctx, this, readerIdleTimeNanos, TimeUnit.NANOSECONDS);

                boolean first = firstReaderIdleEvent;
                firstReaderIdleEvent = false;

                try {
                    IdleStateEvent event = newIdleStateEvent(IdleState.READER_IDLE, first);
                    channelIdle(ctx, event);
                } catch (Throwable t) {
                    ctx.fireExceptionCaught(t);
                }
            } else {
                // Read occurred before the timeout - set a new timeout with shorter delay.
                readerIdleTimeout = schedule(ctx, this, nextDelay, TimeUnit.NANOSECONDS);
            }
        }
    }
复制代码

若是知足条件则会生成一个 IdleStateEvent 事件。

SpringBoot 监控

因为整合了 SpringBoot 以后不但能够利用 Spring 帮咱们管理对象，也能够利用它来作应用监控。

actuator 监控

当咱们为引入了:

<dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>
复制代码

就开启了 SpringBoot 的 actuator 监控功能，他能够暴露出不少监控端点供咱们使用。

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

存在的 Beans：

更多信息请查看：docs.spring.io/spring-boot…

可是若是我想监控如今个人服务端有多少客户端连上来了，分别的 ID 是多少？

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

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

自定义端点

暴露的方式也很简单：

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

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


    /** * 监控端点的 访问地址 * @param id */
    public CustomEndpoint(String id) {
        //false 表示不是敏感端点
        super(id, false);
    }

    @Override
    public Map<Long, NioSocketChannel> invoke() {
        return NettySocketHolder.getMAP();
    }
}
复制代码

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

再配置一个该类型的 Bean 便可：

@Configuration
public class EndPointConfig {


    @Value("${monitor.channel.map.key}")
    private String channelMap;

    @Bean
    public CustomEndpoint buildEndPoint(){
        CustomEndpoint customEndpoint = new CustomEndpoint(channelMap) ;
        return customEndpoint ;
    }
}
复制代码

这样咱们就能够经过配置文件中的 monitor.channel.map.key 来访问了：

一个客户端链接时：

两个客户端链接时：

整合 SBA

这样其实监控功能已经能够知足了，但能不能展现的更美观、而且多个应用也能够方便查看呢？

有这样的开源工具帮咱们作到了：

github.com/codecentric…

简单来讲咱们能够利用该工具将 actuator 暴露出来的接口可视化并聚合的展现在页面中：

接入也很简单，首先须要引入依赖：

<dependency>
            <groupId>de.codecentric</groupId>
            <artifactId>spring-boot-admin-starter-client</artifactId>
        </dependency>        

复制代码

并在配置文件中加入：

# 关闭健康检查权限
management.security.enabled=false
# SpringAdmin 地址
spring.boot.admin.url=http://127.0.0.1:8888
复制代码

在启动应用以前先讲 SpringBootAdmin 部署好：

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

@SpringBootApplication
@Configuration
@EnableAutoConfiguration
@EnableAdminServer
public class AdminApplication {

	public static void main(String[] args) {
		SpringApplication.run(AdminApplication.class, args);
	}
}
复制代码

引入：

<dependency>
			<groupId>de.codecentric</groupId>
			<artifactId>spring-boot-admin-starter-server</artifactId>
			<version>1.5.7</version>
		</dependency>
		<dependency>
			<groupId>de.codecentric</groupId>
			<artifactId>spring-boot-admin-server-ui</artifactId>
			<version>1.5.6</version>
		</dependency>
复制代码

以后直接启动就好了。

这样咱们在 SpringBootAdmin 的页面中就能够查看不少应用信息了。

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

codecentric.github.io/spring-boot…

自定义监控数据

其实咱们彻底能够借助 actuator 以及这个可视化页面帮咱们监控一些简单的度量信息。

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

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

客户端：

@Controller
@RequestMapping("/")
public class IndexController {

    /** * 统计 service */
    @Autowired
    private CounterService counterService;

    @Autowired
    private HeartbeatClient heartbeatClient ;

    /** * 向服务端发消息 * @param sendMsgReqVO * @return */
    @ApiOperation("客户端发送消息")
    @RequestMapping("sendMsg")
    @ResponseBody
    public BaseResponse<SendMsgResVO> sendMsg(@RequestBody SendMsgReqVO sendMsgReqVO){
        BaseResponse<SendMsgResVO> res = new BaseResponse();
        heartbeatClient.sendMsg(new CustomProtocol(sendMsgReqVO.getId(),sendMsgReqVO.getMsg())) ;

        // 利用 actuator 来自增
        counterService.increment(Constants.COUNTER_CLIENT_PUSH_COUNT);

        SendMsgResVO sendMsgResVO = new SendMsgResVO() ;
        sendMsgResVO.setMsg("OK") ;
        res.setCode(StatusEnum.SUCCESS.getCode()) ;
        res.setMessage(StatusEnum.SUCCESS.getMessage()) ;
        res.setDataBody(sendMsgResVO) ;
        return res ;
    }
}
复制代码

只要咱们引入了 actuator 的包，那就能够直接注入 counterService ，利用它来帮咱们记录数据。

当咱们调用该接口时：

在监控页面中能够查询刚才的调用状况：

服务端主动 push 消息也是相似，只是须要在发送时候根据客户端的 ID 查询到具体的 Channel 发送：

总结

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

本文全部代码：

github.com/crossoverJi…

号外

最近在总结一些 Java 相关的知识点，感兴趣的朋友能够一块儿维护。

地址: github.com/crossoverJi…

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程