SpringBoot整合Netty心跳机制过程详解

2023-07-31 15:37:03 284

前言

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

publicstaticMapgetMAP(){
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

SpringBoot整合Netty心跳机制过程详解

热门推荐

随机推荐