电脑知识|欧美黑人一区二区三区|软件|欧美黑人一级爽快片淫片高清|系统|欧美黑人狂野猛交老妇|数据库|服务器|编程开发|网络运营|知识问答|技术教程文章 - 好吧啦网

您的位置:首頁技術文章
文章詳情頁

SpringBoot整合Netty心跳機制過程詳解

瀏覽:7日期:2023-05-30 10:11:17

前言

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。

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

這里涉及到了自定義協議的內容,請繼續查看下文。

當然少不了啟動引導:

@Componentpublic 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

Componentpublic 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 即可:

@Configurationpublic 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@EnableAdminServerpublic 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 ,利用它來幫我們記錄數據。

總結

以上就是一個簡單 Netty 心跳示例,并演示了 SpringBoot 的監控,之后會繼續更新 Netty 相關內容,歡迎關注及指正。

本文所有代碼:

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

以上就是本文的全部內容,希望對大家的學習有所幫助,也希望大家多多支持好吧啦網。

標簽: Spring
相關文章:
主站蜘蛛池模板: 博博会2021_中国博物馆及相关产品与技术博览会【博博会】 | 微动开关厂家-东莞市德沃电子科技有限公司| 三防漆–水性三防漆–水性浸渍漆–贝塔三防漆厂家 | 石家庄装修设计_室内家装设计_别墅装饰装修公司-石家庄金舍装饰官网 | 【法利莱住人集装箱厂家】—活动集装箱房,集装箱租赁_大品牌,更放心 | 重庆私家花园设计-别墅花园-庭院-景观设计-重庆彩木园林建设有限公司 | 选宝石船-陆地水上开采「精选」色选机械设备-青州冠诚重工机械有限公司 | 湖州织里童装_女童男童中大童装_款式多尺码全_织里儿童网【官网】-嘉兴嘉乐网络科技有限公司 | 干粉砂浆设备_干混砂浆生产线_腻子粉加工设备_石膏抹灰砂浆生产成套设备厂家_干粉混合设备_砂子烘干机--郑州铭将机械设备有限公司 | 专业生物有机肥造粒机,粉状有机肥生产线,槽式翻堆机厂家-郑州华之强重工科技有限公司 | 防火门-专业生产甲级不锈钢钢质防火门厂家资质齐全-广东恒磊安防设备有限公司 | 微妙网,专业的动画师、特效师、CG模型设计师网站! - wmiao.com 超声波电磁流量计-液位计-孔板流量计-料位计-江苏信仪自动化仪表有限公司 | 玄米影院| 同学聚会纪念册制作_毕业相册制作-成都顺时针宣传画册设计公司 | 济南货架定做_仓储货架生产厂_重型货架厂_仓库货架批发_济南启力仓储设备有限公司 | 哈尔滨京科脑康神经内科医院-哈尔滨治疗头痛医院-哈尔滨治疗癫痫康复医院 | 消泡剂_水处理消泡剂_切削液消泡剂_涂料消泡剂_有机硅消泡剂_广州中万新材料生产厂家 | 大鼠骨髓内皮祖细胞-小鼠神经元-无锡欣润生物科技有限公司 | 圆形振动筛_圆筛_旋振筛_三次元振动筛-河南新乡德诚生产厂家 | 深圳市索富通实业有限公司-可燃气体报警器 | 可燃气体探测器 | 气体检测仪 | 重庆小面培训_重庆小面技术培训学习班哪家好【终身免费复学】 | 商用绞肉机-熟肉切片机-冻肉切丁机-猪肉开条机 - 广州市正盈机械设备有限公司 | 锂电池砂磨机|石墨烯砂磨机|碳纳米管砂磨机-常州市奥能达机械设备有限公司 | 工控机-图像采集卡-PoE网卡-人工智能-工业主板-深圳朗锐智科 | 耐压仪-高压耐压仪|徐吉电气| 山东活动策划|济南活动公司|济南公关活动策划-济南锐嘉广告有限公司 | 电动球阀_不锈钢电动球阀_电动三通球阀_电动调节球阀_上海湖泉阀门有限公司 | 球盟会·(中国)官方网站| 脱硝喷枪-氨水喷枪-尿素喷枪-河北思凯淋环保科技有限公司 | 尚为传动-专业高精密蜗轮蜗杆,双导程蜗轮蜗杆,蜗轮蜗杆减速机,蜗杆减速机生产厂家 | 传递窗_超净|洁净工作台_高效过滤器-传递窗厂家广州梓净公司 | 滑板场地施工_极限运动场地设计_滑板公园建造_盐城天人极限运动场地建设有限公司 | 邢台人才网_邢台招聘网_邢台123招聘【智达人才网】 | 无锡网站建设-做网站-建网站-网页设计制作-阿凡达建站公司 | 智慧旅游_智慧景区_微景通-智慧旅游景区解决方案提供商 | 广州冷却塔维修厂家_冷却塔修理_凉水塔风机电机填料抢修-广东康明节能空调有限公司 | 百方网-百方电气网,电工电气行业专业的B2B电子商务平台 | 土壤检测仪器_行星式球磨仪_土壤团粒分析仪厂家_山东莱恩德智能科技有限公司 | 污泥烘干机-低温干化机-工业污泥烘干设备厂家-焦作市真节能环保设备科技有限公司 | 儿童乐园|游乐场|淘气堡招商加盟|室内儿童游乐园配套设备|生产厂家|开心哈乐儿童乐园 | YJLV22铝芯铠装电缆-MYPTJ矿用高压橡套电缆-天津市电缆总厂 |