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

您的位置:首頁技術(shù)文章
文章詳情頁

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

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

前言

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

最終能達到的效果:

客戶端每隔 N 秒檢測是否需要發(fā)送心跳。 服務(wù)端也每隔 N 秒檢測是否需要發(fā)送心跳。 服務(wù)端可以主動 push 消息到客戶端。 基于 SpringBoot 監(jiān)控,可以查看實時連接以及各種應(yīng)用信息。

IdleStateHandler

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

客戶端心跳

當(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。

@Configurationpublic 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):

@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 秒沒發(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()) ; }}

這里有點需要注意:

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

所以每當(dāng)有個連接上來的時候,我們都將當(dāng)前的 Channel 與連上的客戶端 ID 進行關(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

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('關(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 進行通訊的。

所以需要在客戶端進行編碼,服務(wù)端進行解碼,也都只需要各自實現(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ī)則進行解碼即可。

實現(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í)行時進行判斷:

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)控端點供我們使用。

如一些應(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。

這就需要暴露自定義端點了。

自定義端點

暴露的方式也很簡單:

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

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

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

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

這樣其實監(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@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 的頁面中就可以查看很多應(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ù)。

總結(jié)

以上就是一個簡單 Netty 心跳示例,并演示了 SpringBoot 的監(jiān)控,之后會繼續(xù)更新 Netty 相關(guān)內(nèi)容,歡迎關(guān)注及指正。

本文所有代碼:

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

以上就是本文的全部內(nèi)容,希望對大家的學(xué)習(xí)有所幫助,也希望大家多多支持好吧啦網(wǎng)。

標(biāo)簽: Spring
相關(guān)文章:
主站蜘蛛池模板: 网站优化公司_北京网站优化_抖音短视频代运营_抖音关键词seo优化排名-通则达网络 | 顶空进样器-吹扫捕集仪-热脱附仪-二次热解吸仪-北京华盛谱信仪器 | 贝朗斯动力商城(BRCPOWER.COM) - 买叉车蓄电池上贝朗斯商城,价格更超值,品质有保障! | 昆明网络公司|云南网络公司|昆明网站建设公司|昆明网页设计|云南网站制作|新媒体运营公司|APP开发|小程序研发|尽在昆明奥远科技有限公司 | 原色会计-合肥注册公司_合肥代理记账公司_营业执照代办 | 招商帮-一站式网络营销服务|搜索营销推广|信息流推广|短视视频营销推广|互联网整合营销|网络推广代运营|招商帮企业招商好帮手 | 雄松华章(广州华章MBA)官网-专注MBA/MPA/MPAcc/MEM辅导培训 | 洛阳网站建设_洛阳网站优化_网站建设平台_洛阳香河网络科技有限公司 | 天津暖气片厂家_钢制散热器_天津铜铝复合暖气片_维尼罗散热器 | 陕西安玻璃自动感应门-自动重叠门-磁悬浮平开门厂家【捷申达门业】 | 食品机械专用传感器-落料放大器-低价接近开关-菲德自控技术(天津)有限公司 | 新能源汽车电池软连接,铜铝复合膜柔性连接,电力母排-容发智能科技(无锡)有限公司 | 【连江县榕彩涂料有限公司】官方网站 | 捷码低代码平台 - 3D数字孪生_大数据可视化开发平台「免费体验」 | 深圳VI设计-画册设计-LOGO设计-包装设计-品牌策划公司-[智睿画册设计公司] | 连续密炼机_双转子连续密炼机_连续式密炼机-南京永睿机械制造有限公司 | 欧版反击式破碎机-欧版反击破-矿山石料破碎生产线-青州奥凯诺机械 | 苏州西朗门业-欧盟CE|莱茵UL双认证的快速卷帘门品牌厂家 | 10吨无线拉力计-2吨拉力计价格-上海佳宜电子科技有限公司 | 商秀—企业短视频代运营_抖音企业号托管 | 5nd音乐网|最新流行歌曲|MP3歌曲免费下载|好听的歌|音乐下载 免费听mp3音乐 | 不锈钢法兰-碳钢法兰-法兰盘生产加工厂家-[鼎捷峰]-不锈钢法兰-碳钢法兰-法兰盘生产加工厂家-[鼎捷峰] | 东莞市海宝机械有限公司-不锈钢分选机-硅胶橡胶-生活垃圾-涡电流-静电-金属-矿石分选机 | 佛山市钱丰金属不锈钢蜂窝板定制厂家|不锈钢装饰线条|不锈钢屏风| 电梯装饰板|不锈钢蜂窝板不锈钢工艺板材厂家佛山市钱丰金属制品有限公司 | 工业冷却塔维修厂家_方形不锈钢工业凉水塔维修改造方案-广东康明节能空调有限公司 | 西点培训学校_法式西点培训班_西点师培训_西点蛋糕培训-广州烘趣西点烘焙培训学院 | 耐磨焊丝,堆焊焊丝,耐磨药芯焊丝,碳化钨焊丝-北京耐默公司 | 科研ELISA试剂盒,酶联免疫检测试剂盒,昆虫_植物ELISA酶免试剂盒-上海仁捷生物科技有限公司 | 3d打印服务,3d打印汽车,三维扫描,硅胶复模,手板,快速模具,深圳市精速三维打印科技有限公司 | 手持式浮游菌采样器-全排二级生物安全柜-浙江孚夏医疗科技有限公司 | 考勤系统_考勤管理系统_网络考勤软件_政企|集团|工厂复杂考勤工时统计排班管理系统_天时考勤 | 上海APP开发-APP制作-APP定制开发-上海APP开发制作公司-咏熠科技 | 行业分析:提及郑州火车站附近真有 特殊按摩 ?2025实地踩坑指南 新手如何避坑不踩雷 | 大流量卧式砂磨机_强力分散机_双行星双动力混合机_同心双轴搅拌机-莱州市龙跃化工机械有限公司 | 有机肥设备生产制造厂家,BB掺混肥搅拌机、复合肥设备生产线,有机肥料全部加工设备多少钱,对辊挤压造粒机,有机肥造粒设备 -- 郑州程翔重工机械有限公司 | 伶俐嫂培训学校_月嫂培训班在哪里报名学费是多少_月嫂免费政府培训中心推荐 | 微波萃取合成仪-电热消解器价格-北京安合美诚科学仪器有限公司 | 盛源真空泵|空压机-浙江盛源空压机制造有限公司-【盛源官网】 | 外观设计_设备外观设计_外观设计公司_产品外观设计_机械设备外观设计_东莞工业设计公司-意品深蓝 | 天津次氯酸钠酸钙溶液-天津氢氧化钠厂家-天津市辅仁化工有限公司 | 手持气象站_便携式气象站_农业气象站_负氧离子监测站-山东万象环境 |