SpringBoot整合Netty心跳机制过程详解
前言
Netty是一个高性能的NIO网络框架,本文基于SpringBoot以常见的心跳机制来认识Netty。
最终能达到的效果:
- 客户端每隔N秒检测是否需要发送心跳。
- 服务端也每隔N秒检测是否需要发送心跳。
- 服务端可以主动push消息到客户端。
- 基于SpringBoot监控,可以查看实时连接以及各种应用信息。
IdleStateHandler
Netty可以使用IdleStateHandler来实现连接管理,当连接空闲时间太长(没有发送、接收消息)时则会触发一个事件,我们便可在该事件中实现心跳机制。
客户端心跳
当客户端空闲了N秒没有给服务端发送消息时会自动发送一个心跳来维持连接。
核心代码代码如下:
publicclassEchoClientHandleextendsSimpleChannelInboundHandler{ privatefinalstaticLoggerLOGGER=LoggerFactory.getLogger(EchoClientHandle.class); @Override publicvoiduserEventTriggered(ChannelHandlerContextctx,Objectevt)throwsException{ if(evtinstanceofIdleStateEvent){ IdleStateEventidleStateEvent=(IdleStateEvent)evt; if(idleStateEvent.state()==IdleState.WRITER_IDLE){ LOGGER.info("已经10秒没有发送信息!"); //向服务端发送消息 CustomProtocolheartBeat=SpringBeanFactory.getBean("heartBeat",CustomProtocol.class); ctx.writeAndFlush(heartBeat).addListener(ChannelFutureListener.CLOSE_ON_FAILURE); } } super.userEventTriggered(ctx,evt); } @Override protectedvoidchannelRead0(ChannelHandlerContextchannelHandlerContext,ByteBufin)throwsException{ //从服务端收到消息时被调用 LOGGER.info("客户端收到消息={}",in.toString(CharsetUtil.UTF_8)); } }
实现非常简单,只需要在事件回调中发送一个消息即可。
由于整合了SpringBoot,所以发送的心跳信息是一个单例的Bean。
@Configuration publicclassHeartBeatConfig{ @Value("${channel.id}") privatelongid; @Bean(value="heartBeat") publicCustomProtocolheartBeat(){ returnnewCustomProtocol(id,"ping"); } }
这里涉及到了自定义协议的内容,请继续查看下文。
当然少不了启动引导:
@Component publicclassHeartbeatClient{ privatefinalstaticLoggerLOGGER=LoggerFactory.getLogger(HeartbeatClient.class); privateEventLoopGroupgroup=newNioEventLoopGroup(); @Value("${netty.server.port}") privateintnettyPort; @Value("${netty.server.host}") privateStringhost; privateSocketChannelchannel; @PostConstruct publicvoidstart()throwsInterruptedException{ Bootstrapbootstrap=newBootstrap(); bootstrap.group(group) .channel(NioSocketChannel.class) .handler(newCustomerHandleInitializer()) ; ChannelFuturefuture=bootstrap.connect(host,nettyPort).sync(); if(future.isSuccess()){ LOGGER.info("启动Netty成功"); } channel=(SocketChannel)future.channel(); } } publicclassCustomerHandleInitializerextendsChannelInitializer{ @Override protectedvoidinitChannel(Channelch)throwsException{ ch.pipeline() //10秒没发送消息将IdleStateHandler添加到ChannelPipeline中 .addLast(newIdleStateHandler(0,10,0)) .addLast(newHeartbeatEncode()) .addLast(newEchoClientHandle()) ; } }
所以当应用启动每隔10秒会检测是否发送过消息,不然就会发送心跳信息。
服务端心跳
服务器端的心跳其实也是类似,也需要在ChannelPipeline中添加一个IdleStateHandler。
publicclassHeartBeatSimpleHandleextendsSimpleChannelInboundHandler{ privatefinalstaticLoggerLOGGER=LoggerFactory.getLogger(HeartBeatSimpleHandle.class); privatestaticfinalByteBufHEART_BEAT=Unpooled.unreleasableBuffer(Unpooled.copiedBuffer(newCustomProtocol(123456L,"pong").toString(),CharsetUtil.UTF_8)); /** *取消绑定 *@paramctx *@throwsException */ @Override publicvoidchannelInactive(ChannelHandlerContextctx)throwsException{ NettySocketHolder.remove((NioSocketChannel)ctx.channel()); } @Override publicvoiduserEventTriggered(ChannelHandlerContextctx,Objectevt)throwsException{ if(evtinstanceofIdleStateEvent){ IdleStateEventidleStateEvent=(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 protectedvoidchannelRead0(ChannelHandlerContextctx,CustomProtocolcustomProtocol)throwsException{ LOGGER.info("收到customProtocol={}",customProtocol); //保存客户端与Channel之间的关系 NettySocketHolder.put(customProtocol.getId(),(NioSocketChannel)ctx.channel()); } }
这里有点需要注意:
当有多个客户端连上来时,服务端需要区分开,不然响应消息就会发生混乱。
所以每当有个连接上来的时候,我们都将当前的Channel与连上的客户端ID进行关联(因此每个连上的客户端ID都必须唯一)。
这里采用了一个Map来保存这个关系,并且在断开连接时自动取消这个关联。
publicclassNettySocketHolder{ privatestaticfinalMapMAP=newConcurrentHashMap<>(16); publicstaticvoidput(Longid,NioSocketChannelsocketChannel){ MAP.put(id,socketChannel); } publicstaticNioSocketChannelget(Longid){ returnMAP.get(id); } publicstaticMap getMAP(){ returnMAP; } publicstaticvoidremove(NioSocketChannelnioSocketChannel){ MAP.entrySet().stream().filter(entry->entry.getValue()==nioSocketChannel).forEach(entry->MAP.remove(entry.getKey())); } }
启动引导程序:
Component
Component publicclassHeartBeatServer{ privatefinalstaticLoggerLOGGER=LoggerFactory.getLogger(HeartBeatServer.class); privateEventLoopGroupboss=newNioEventLoopGroup(); privateEventLoopGroupwork=newNioEventLoopGroup(); @Value("${netty.server.port}") privateintnettyPort; /** *启动Netty * *@return *@throwsInterruptedException */ @PostConstruct publicvoidstart()throwsInterruptedException{ ServerBootstrapbootstrap=newServerBootstrap() .group(boss,work) .channel(NioServerSocketChannel.class) .localAddress(newInetSocketAddress(nettyPort)) //保持长连接 .childOption(ChannelOption.SO_KEEPALIVE,true) .childHandler(newHeartbeatInitializer()); ChannelFuturefuture=bootstrap.bind().sync(); if(future.isSuccess()){ LOGGER.info("启动Netty成功"); } } /** *销毁 */ @PreDestroy publicvoiddestroy(){ boss.shutdownGracefully().syncUninterruptibly(); work.shutdownGracefully().syncUninterruptibly(); LOGGER.info("关闭Netty成功"); } } publicclassHeartbeatInitializerextendsChannelInitializer{ @Override protectedvoidinitChannel(Channelch)throwsException{ ch.pipeline() //五秒没有收到消息将IdleStateHandler添加到ChannelPipeline中 .addLast(newIdleStateHandler(5,0,0)) .addLast(newHeartbeatDecoder()) .addLast(newHeartBeatSimpleHandle()); } }
也是同样将IdleStateHandler添加到ChannelPipeline中,也会有一个定时任务,每5秒校验一次是否有收到消息,否则就主动发送一次请求。
因为测试是有两个客户端连上所以有两个日志。
自定义协议
上文其实都看到了:服务端与客户端采用的是自定义的POJO进行通讯的。
所以需要在客户端进行编码,服务端进行解码,也都只需要各自实现一个编解码器即可。
CustomProtocol:
publicclassCustomProtocolimplementsSerializable{ privatestaticfinallongserialVersionUID=4671171056588401542L; privatelongid; privateStringcontent; //省略getter/setter }
客户端的编码器:
publicclassHeartbeatEncodeextendsMessageToByteEncoder{ @Override protectedvoidencode(ChannelHandlerContextctx,CustomProtocolmsg,ByteBufout)throwsException{ out.writeLong(msg.getId()); out.writeBytes(msg.getContent().getBytes()); } }
也就是说消息的前八个字节为header,剩余的全是content。
服务端的解码器:
publicclassHeartbeatDecoderextendsByteToMessageDecoder{ @Override protectedvoiddecode(ChannelHandlerContextctx,ByteBufin,List
只需要按照刚才的规则进行解码即可。
实现原理
其实联想到IdleStateHandler的功能,自然也能想到它实现的原理:
应该会存在一个定时任务的线程去处理这些消息。
来看看它的源码:
首先是构造函数:
publicIdleStateHandler( intreaderIdleTimeSeconds, intwriterIdleTimeSeconds, intallIdleTimeSeconds){ this(readerIdleTimeSeconds,writerIdleTimeSeconds,allIdleTimeSeconds, TimeUnit.SECONDS); }
其实就是初始化了几个数据:
- readerIdleTimeSeconds:一段时间内没有数据读取
- writerIdleTimeSeconds:一段时间内没有数据发送
- allIdleTimeSeconds:以上两种满足其中一个即可
因为IdleStateHandler也是一种ChannelHandler,所以会在channelActive中初始化任务:
@Override publicvoidchannelActive(ChannelHandlerContextctx)throwsException{ //Thismethodwillbeinvokedonlyifthishandlerwasadded //beforechannelActive()eventisfired.Ifauseraddsthishandler //afterthechannelActive()event,initialize()willbecalledbybeforeAdd(). initialize(ctx); super.channelActive(ctx); } privatevoidinitialize(ChannelHandlerContextctx){ //Avoidthecasewheredestroy()iscalledbeforeschedulingtimeouts. //See:https://github.com/netty/netty/issues/143 switch(state){ case1: case2: return; } state=1; initOutputChanged(ctx); lastReadTime=lastWriteTime=ticksInNanos(); if(readerIdleTimeNanos>0){ readerIdleTimeout=schedule(ctx,newReaderIdleTimeoutTask(ctx), readerIdleTimeNanos,TimeUnit.NANOSECONDS); } if(writerIdleTimeNanos>0){ writerIdleTimeout=schedule(ctx,newWriterIdleTimeoutTask(ctx), writerIdleTimeNanos,TimeUnit.NANOSECONDS); } if(allIdleTimeNanos>0){ allIdleTimeout=schedule(ctx,newAllIdleTimeoutTask(ctx), allIdleTimeNanos,TimeUnit.NANOSECONDS); } }
也就是会按照我们给定的时间初始化出定时任务。
接着在任务真正执行时进行判断:
privatefinalclassReaderIdleTimeoutTaskextendsAbstractIdleTask{ ReaderIdleTimeoutTask(ChannelHandlerContextctx){ super(ctx); } @Override protectedvoidrun(ChannelHandlerContextctx){ longnextDelay=readerIdleTimeNanos; if(!reading){ nextDelay-=ticksInNanos()-lastReadTime; } if(nextDelay<=0){ //Readerisidle-setanewtimeoutandnotifythecallback. readerIdleTimeout=schedule(ctx,this,readerIdleTimeNanos,TimeUnit.NANOSECONDS); booleanfirst=firstReaderIdleEvent; firstReaderIdleEvent=false; try{ IdleStateEventevent=newIdleStateEvent(IdleState.READER_IDLE,first); channelIdle(ctx,event); }catch(Throwablet){ ctx.fireExceptionCaught(t); } }else{ //Readoccurredbeforethetimeout-setanewtimeoutwithshorterdelay. readerIdleTimeout=schedule(ctx,this,nextDelay,TimeUnit.NANOSECONDS); } } }
如果满足条件则会生成一个IdleStateEvent事件。
SpringBoot监控
由于整合了SpringBoot之后不但可以利用Spring帮我们管理对象,也可以利用它来做应用监控。
actuator监控
当我们为引入了:
org.springframework.boot spring-boot-starter-actuator
就开启了SpringBoot的actuator监控功能,他可以暴露出很多监控端点供我们使用。
如一些应用中的一些统计数据:
存在的Beans:
更多信息请查看:https://docs.spring.io/spring-boot/docs/current/reference/html/production-ready-endpoints.html
但是如果我想监控现在我的服务端有多少客户端连上来了,分别的ID是多少?
其实就是实时查看我内部定义的那个关联关系的Map。
这就需要暴露自定义端点了。
自定义端点
暴露的方式也很简单:
继承AbstractEndpoint并复写其中的invoke函数:
publicclassCustomEndpointextendsAbstractEndpoint
其实就是返回了Map中的数据。
再配置一个该类型的Bean即可:
@Configuration publicclassEndPointConfig{ @Value("${monitor.channel.map.key}") privateStringchannelMap; @Bean publicCustomEndpointbuildEndPoint(){ CustomEndpointcustomEndpoint=newCustomEndpoint(channelMap); returncustomEndpoint; } }
这样我们就可以通过配置文件中的monitor.channel.map.key来访问了:
整合SBA
这样其实监控功能已经可以满足了,但能不能展示的更美观、并且多个应用也可以方便查看呢?
有这样的开源工具帮我们做到了:
https://github.com/codecentric/spring-boot-admin
简单来说我们可以利用该工具将actuator暴露出来的接口可视化并聚合的展示在页面中:
接入也很简单,首先需要引入依赖:
de.codecentric spring-boot-admin-starter-client
并在配置文件中加入:
#关闭健康检查权限 management.security.enabled=false #SpringAdmin地址 spring.boot.admin.url=http://127.0.0.1:8888
在启动应用之前先讲SpringBootAdmin部署好:
这个应用就是一个纯粹的SpringBoot,只需要在主函数上加入@EnableAdminServer注解。
@SpringBootApplication @Configuration @EnableAutoConfiguration @EnableAdminServer publicclassAdminApplication{ publicstaticvoidmain(String[]args){ SpringApplication.run(AdminApplication.class,args); } }
引入:
de.codecentric spring-boot-admin-starter-server 1.5.7 de.codecentric spring-boot-admin-server-ui 1.5.6
之后直接启动就行了。
这样我们在SpringBootAdmin的页面中就可以查看很多应用信息了。
更多内容请参考官方指南:
http://codecentric.github.io/spring-boot-admin/1.5.6/
自定义监控数据
其实我们完全可以借助actuator以及这个可视化页面帮我们监控一些简单的度量信息。
比如我在客户端和服务端中写了两个Rest接口用于向对方发送消息。
只是想要记录分别发送了多少次:
客户端
@Controller @RequestMapping("/") publicclassIndexController{ /** *统计service */ @Autowired privateCounterServicecounterService; @Autowired privateHeartbeatClientheartbeatClient; /** *向服务端发消息 *@paramsendMsgReqVO *@return */ @ApiOperation("客户端发送消息") @RequestMapping("sendMsg") @ResponseBody publicBaseResponsesendMsg(@RequestBodySendMsgReqVOsendMsgReqVO){ BaseResponse res=newBaseResponse(); heartbeatClient.sendMsg(newCustomProtocol(sendMsgReqVO.getId(),sendMsgReqVO.getMsg())); //利用actuator来自增 counterService.increment(Constants.COUNTER_CLIENT_PUSH_COUNT); SendMsgResVOsendMsgResVO=newSendMsgResVO(); sendMsgResVO.setMsg("OK"); res.setCode(StatusEnum.SUCCESS.getCode()); res.setMessage(StatusEnum.SUCCESS.getMessage()); res.setDataBody(sendMsgResVO); returnres; } }
只要我们引入了actuator的包,那就可以直接注入counterService,利用它来帮我们记录数据。
总结
以上就是一个简单Netty心跳示例,并演示了SpringBoot的监控,之后会继续更新Netty相关内容,欢迎关注及指正。
本文所有代码:
https://github.com/crossoverJie/netty-action
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持毛票票。
声明:本文内容来源于网络,版权归原作者所有,内容由互联网用户自发贡献自行上传,本网站不拥有所有权,未作人工编辑处理,也不承担相关法律责任。如果您发现有涉嫌版权的内容,欢迎发送邮件至:czq8825#qq.com(发邮件时,请将#更换为@)进行举报,并提供相关证据,一经查实,本站将立刻删除涉嫌侵权内容。