SpringBoot如何整合Netty心跳機(jī)制

這篇文章將為大家詳細(xì)講解有關(guān)SpringBoot如何整合Netty心跳機(jī)制，小編覺得挺實用的，因此分享給大家做個參考，希望大家閱讀完這篇文章后可以有所收獲。

創(chuàng)新互聯(lián)建站云計算的互聯(lián)網(wǎng)服務(wù)提供商,擁有超過13年的服務(wù)器租用、服務(wù)器托管、云服務(wù)器、雅安服務(wù)器托管、網(wǎng)站系統(tǒng)開發(fā)經(jīng)驗，已先后獲得國家工業(yè)和信息化部頒發(fā)的互聯(lián)網(wǎng)數(shù)據(jù)中心業(yè)務(wù)許可證。專業(yè)提供云主機(jī)、雅安服務(wù)器托管、域名注冊、VPS主機(jī)、云服務(wù)器、香港云服務(wù)器、免備案服務(wù)器等。

前言

Netty 是一個高性能的 NIO 網(wǎng)絡(luò)框架，本文基于 SpringBoot 以常見的心跳機(jī)制來認(rèn)識 Netty。

最終能達(dá)到的效果：

  • 客戶端每隔 N 秒檢測是否需要發(fā)送心跳。

  • 服務(wù)端也每隔 N 秒檢測是否需要發(fā)送心跳。

  • 服務(wù)端可以主動 push 消息到客戶端。

  • 基于 SpringBoot 監(jiān)控，可以查看實時連接以及各種應(yīng)用信息。

IdleStateHandler

Netty 可以使用 IdleStateHandler 來實現(xiàn)連接管理，當(dāng)連接空閑時間太長（沒有發(fā)送、接收消息）時則會觸發(fā)一個事件，我們便可在該事件中實現(xiàn)心跳機(jī)制。

客戶端心跳

當(dāng)客戶端空閑了 N 秒沒有給服務(wù)端發(fā)送消息時會自動發(fā)送一個心跳來維持連接。

核心代碼代碼如下：

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("已經(jīng) 10 秒沒有發(fā)送信息！");
        //向服務(wù)端發(fā)送消息
        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 {

    //從服務(wù)端收到消息時被調(diào)用
    LOGGER.info("客戶端收到消息={}",in.toString(CharsetUtil.UTF_8)) ;
  }
}

實現(xiàn)非常簡單，只需要在事件回調(diào)中發(fā)送一個消息即可。

由于整合了 SpringBoot ，所以發(fā)送的心跳信息是一個單例的 Bean。

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

這里涉及到了自定義協(xié)議的內(nèi)容，請繼續(xù)查看下文。

當(dāng)然少不了啟動引導(dǎo)：

@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 秒沒發(fā)送消息 將IdleStateHandler 添加到 ChannelPipeline 中
        .addLast(new IdleStateHandler(0, 10, 0))
        .addLast(new HeartbeatEncode())
        .addLast(new EchoClientHandle())
    ;
  }
}

所以當(dāng)應(yīng)用啟動每隔 10 秒會檢測是否發(fā)送過消息，不然就會發(fā)送心跳信息。

服務(wù)端心跳

服務(wù)器端的心跳其實也是類似，也需要在 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("已經(jīng)5秒沒有收到信息！");
        //向客戶端發(fā)送消息
        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 之間的關(guān)系
    NettySocketHolder.put(customProtocol.getId(),(NioSocketChannel)ctx.channel()) ;
  }
}

這里有點(diǎn)需要注意：

當(dāng)有多個客戶端連上來時，服務(wù)端需要區(qū)分開，不然響應(yīng)消息就會發(fā)生混亂。

所以每當(dāng)有個連接上來的時候，我們都將當(dāng)前的 Channel 與連上的客戶端 ID 進(jìn)行關(guān)聯(lián)（因此每個連上的客戶端 ID 都必須唯一）。

這里采用了一個 Map 來保存這個關(guān)系，并且在斷開連接時自動取消這個關(guān)聯(lián)。

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

啟動引導(dǎo)程序：

Component

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("關(guān)閉 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 中，也會有一個定時任務(wù)，每5秒校驗一次是否有收到消息，否則就主動發(fā)送一次請求。

因為測試是有兩個客戶端連上所以有兩個日志。

自定義協(xié)議

上文其實都看到了：服務(wù)端與客戶端采用的是自定義的 POJO 進(jìn)行通訊的。

所以需要在客戶端進(jìn)行編碼，服務(wù)端進(jìn)行解碼，也都只需要各自實現(xiàn)一個編解碼器即可。

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

  }
}

也就是說消息的前八個字節(jié)為 header，剩余的全是 content。

服務(wù)端的解碼器：

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

  }
}

只需要按照剛才的規(guī)則進(jìn)行解碼即可。

實現(xiàn)原理

其實聯(lián)想到 IdleStateHandler 的功能，自然也能想到它實現(xiàn)的原理：

應(yīng)該會存在一個定時任務(wù)的線程去處理這些消息。

來看看它的源碼：

首先是構(gòu)造函數(shù):

  public IdleStateHandler(
      int readerIdleTimeSeconds,
      int writerIdleTimeSeconds,
      int allIdleTimeSeconds) {

    this(readerIdleTimeSeconds, writerIdleTimeSeconds, allIdleTimeSeconds,
       TimeUnit.SECONDS);
  }

其實就是初始化了幾個數(shù)據(jù)：

  • readerIdleTimeSeconds：一段時間內(nèi)沒有數(shù)據(jù)讀取

  • writerIdleTimeSeconds：一段時間內(nèi)沒有數(shù)據(jù)發(fā)送

  • allIdleTimeSeconds：以上兩種滿足其中一個即可

因為 IdleStateHandler 也是一種 ChannelHandler，所以會在 channelActive 中初始化任務(wù)：

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

也就是會按照我們給定的時間初始化出定時任務(wù)。

接著在任務(wù)真正執(zhí)行時進(jìn)行判斷：

  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 監(jiān)控

由于整合了 SpringBoot 之后不但可以利用 Spring 幫我們管理對象，也可以利用它來做應(yīng)用監(jiān)控。

actuator 監(jiān)控

當(dāng)我們?yōu)橐肓?

<dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-actuator</artifactId>
    </dependency>

就開啟了 SpringBoot 的 actuator 監(jiān)控功能，他可以暴露出很多監(jiān)控端點(diǎn)供我們使用。

如一些應(yīng)用中的一些統(tǒng)計數(shù)據(jù)：

存在的 Beans：

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

但是如果我想監(jiān)控現(xiàn)在我的服務(wù)端有多少客戶端連上來了，分別的 ID 是多少？

其實就是實時查看我內(nèi)部定義的那個關(guān)聯(lián)關(guān)系的 Map。

這就需要暴露自定義端點(diǎn)了。

自定義端點(diǎn)

暴露的方式也很簡單：

繼承 AbstractEndpoint 并復(fù)寫其中的 invoke 函數(shù)：

public class CustomEndpoint extends AbstractEndpoint<Map<Long,NioSocketChannel>> {
  /**
   * 監(jiān)控端點(diǎn)的 訪問地址
   * @param id
   */
  public CustomEndpoint(String id) {
    //false 表示不是敏感端點(diǎn)
    super(id, false);
  }
  @Override
  public Map<Long, NioSocketChannel> invoke() {
    return NettySocketHolder.getMAP();
  }
}

其實就是返回了 Map 中的數(shù)據(jù)。

再配置一個該類型的 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

這樣其實監(jiān)控功能已經(jīng)可以滿足了，但能不能展示的更美觀、并且多個應(yīng)用也可以方便查看呢？

有這樣的開源工具幫我們做到了：

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

簡單來說我們可以利用該工具將 actuator 暴露出來的接口可視化并聚合的展示在頁面中：

接入也很簡單，首先需要引入依賴：

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

并在配置文件中加入：

# 關(guān)閉健康檢查權(quán)限
management.security.enabled=false
# SpringAdmin 地址
spring.boot.admin.url=http://127.0.0.1:8888

在啟動應(yīng)用之前先講 SpringBootAdmin 部署好：

這個應(yīng)用就是一個純粹的 SpringBoot ，只需要在主函數(shù)上加入 @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 的頁面中就可以查看很多應(yīng)用信息了。

更多內(nèi)容請參考官方指南：

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

自定義監(jiān)控數(shù)據(jù)

其實我們完全可以借助 actuator 以及這個可視化頁面幫我們監(jiān)控一些簡單的度量信息。

比如我在客戶端和服務(wù)端中寫了兩個 Rest 接口用于向?qū)Ψ桨l(fā)送消息。

只是想要記錄分別發(fā)送了多少次：

客戶端

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

  /**
   * 統(tǒng)計 service
   */
  @Autowired
  private CounterService counterService;

  @Autowired
  private HeartbeatClient heartbeatClient ;

  /**
   * 向服務(wù)端發(fā)消息
   * @param sendMsgReqVO
   * @return
   */
  @ApiOperation("客戶端發(fā)送消息")
  @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 ，利用它來幫我們記錄數(shù)據(jù)。

關(guān)于“SpringBoot如何整合Netty心跳機(jī)制”這篇文章就分享到這里了，希望以上內(nèi)容可以對大家有一定的幫助，使各位可以學(xué)到更多知識，如果覺得文章不錯，請把它分享出去讓更多的人看到。

分享題目：SpringBoot如何整合Netty心跳機(jī)制
文章URL：http://muchs.cn/article48/jpdgep.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供企業(yè)網(wǎng)站制作、網(wǎng)站設(shè)計、營銷型網(wǎng)站建設(shè)、用戶體驗、企業(yè)建站、電子商務(wù)

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請盡快告知，我們將會在第一時間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時需注明來源： 創(chuàng)新互聯(lián)

成都seo排名網(wǎng)站優(yōu)化