javascript
Netty(一) SpringBoot 整合长连接心跳机制
?https://github.com/crossoverJie/JCSprout
前言
Netty 是一個高性能的 NIO 網絡框架，本文基于 SpringBoot 以常見的心跳機制來認識 Netty。
最終能達到的效果：
-
客戶端每隔 N 秒檢測是否需要發送心跳。
-
服務端也每隔 N 秒檢測是否需要發送心跳。
-
服務端可以主動 push 消息到客戶端。
-
基于 SpringBoot 監控，可以查看實時連接以及各種應用信息。
效果如下：
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 中初始化任務：
? ?@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：
更多信息請查看：https://docs.spring.io/spring-boot/docs/current/reference/html/production-ready-endpoints.html
但是如果我想監控現在我的服務端有多少客戶端連上來了，分別的 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
這樣其實監控功能已經可以滿足了，但能不能展示的更美觀、并且多個應用也可以方便查看呢？
有這樣的開源工具幫我們做到了：
https://github.com/codecentric/spring-boot-admin
簡單來說我們可以利用該工具將 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 的頁面中就可以查看很多應用信息了。
更多內容請參考官方指南：
http://codecentric.github.io/spring-boot-admin/1.5.6/
自定義監控數據
其實我們完全可以借助 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 相關內容，歡迎關注及指正。
本文所有代碼：
https://github.com/crossoverJie/netty-action
號外
最近在總結一些 Java 相關的知識點，感興趣的朋友可以一起維護。
地址: https://github.com/crossoverJie/Java-Interview
?
閱讀原文轉載于:https://www.cnblogs.com/bigben0123/p/9754370.html
總結
以上是生活随笔為你收集整理的Netty(一) SpringBoot 整合长连接心跳机制的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： Python离线安装PIL 模块（pil
- 下一篇： JS 向未声明的变量分配值（可删除）