SpringBoot整合Netty心跳機(jī)制過程詳解
前言
Netty 是一個(gè)高性能的 NIO 網(wǎng)絡(luò)框架，本文基于 SpringBoot 以常見的心跳機(jī)制來認(rèn)識(shí) Netty。
最終能達(dá)到的效果：
- 客戶端每隔 N 秒檢測是否需要發(fā)送心跳。
- 服務(wù)端也每隔 N 秒檢測是否需要發(fā)送心跳。
- 服務(wù)端可以主動(dòng) push 消息到客戶端。
- 基于 SpringBoot 監(jiān)控，可以查看實(shí)時(shí)連接以及各種應(yīng)用信息。
IdleStateHandler
Netty 可以使用 IdleStateHandler 來實(shí)現(xiàn)連接管理，當(dāng)連接空閑時(shí)間太長（沒有發(fā)送、接收消息）時(shí)則會(huì)觸發(fā)一個(gè)事件，我們便可在該事件中實(shí)現(xiàn)心跳機(jī)制。
客戶端心跳
當(dāng)客戶端空閑了 N 秒沒有給服務(wù)端發(fā)送消息時(shí)會(huì)自動(dòng)發(fā)送一個(gè)心跳來維持連接。
核心代碼代碼如下：
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ù)端收到消息時(shí)被調(diào)用
LOGGER.info("客戶端收到消息={}",in.toString(CharsetUtil.UTF_8)) ;
}
}
實(shí)現(xiàn)非常簡單，只需要在事件回調(diào)中發(fā)送一個(gè)消息即可。
由于整合了 SpringBoot ，所以發(fā)送的心跳信息是一個(gè)單例的 Bean。
@Configuration
public class HeartBeatConfig {
@Value("${channel.id}")
private long id ;
@Bean(value = "heartBeat")
public CustomProtocol heartBeat(){
return new CustomProtocol(id,"ping") ;
}
}
這里涉及到了自定義協(xié)議的內(nèi)容，請(qǐng)繼續(xù)查看下文。
當(dāng)然少不了啟動(dòng)引導(dǎo)：
@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("啟動(dòng) 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)用啟動(dòng)每隔 10 秒會(huì)檢測是否發(fā)送過消息，不然就會(huì)發(fā)送心跳信息。
服務(wù)端心跳
服務(wù)器端的心跳其實(shí)也是類似，也需要在 ChannelPipeline 中添加一個(gè) 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()) ;
}
}
這里有點(diǎn)需要注意：
當(dāng)有多個(gè)客戶端連上來時(shí)，服務(wù)端需要區(qū)分開，不然響應(yīng)消息就會(huì)發(fā)生混亂。
所以每當(dāng)有個(gè)連接上來的時(shí)候，我們都將當(dāng)前的 Channel 與連上的客戶端 ID 進(jìn)行關(guān)聯(lián)（因此每個(gè)連上的客戶端 ID 都必須唯一）。
這里采用了一個(gè) Map 來保存這個(gè)關(guān)系，并且在斷開連接時(shí)自動(dòng)取消這個(gè)關(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òng)引導(dǎo)程序：
Component
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;
/**
* 啟動(dòng) 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("啟動(dòng) 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 中，也會(huì)有一個(gè)定時(shí)任務(wù)，每5秒校驗(yàn)一次是否有收到消息，否則就主動(dòng)發(fā)送一次請(qǐng)求。
因?yàn)闇y試是有兩個(gè)客戶端連上所以有兩個(gè)日志。
自定義協(xié)議
上文其實(shí)都看到了：服務(wù)端與客戶端采用的是自定義的 POJO 進(jìn)行通訊的。
所以需要在客戶端進(jìn)行編碼，服務(wù)端進(jìn)行解碼，也都只需要各自實(shí)現(xiàn)一個(gè)編解碼器即可。
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()) ;
}
}
也就是說消息的前八個(gè)字節(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ī)則進(jìn)行解碼即可。
實(shí)現(xiàn)原理
其實(shí)聯(lián)想到 IdleStateHandler 的功能，自然也能想到它實(shí)現(xiàn)的原理：
應(yīng)該會(huì)存在一個(gè)定時(shí)任務(wù)的線程去處理這些消息。
來看看它的源碼：
首先是構(gòu)造函數(shù):
public IdleStateHandler(
int readerIdleTimeSeconds,
int writerIdleTimeSeconds,
int allIdleTimeSeconds) {
this(readerIdleTimeSeconds, writerIdleTimeSeconds, allIdleTimeSeconds,
TimeUnit.SECONDS);
}
其實(shí)就是初始化了幾個(gè)數(shù)據(jù)：
- readerIdleTimeSeconds：一段時(shí)間內(nèi)沒有數(shù)據(jù)讀取
- writerIdleTimeSeconds：一段時(shí)間內(nèi)沒有數(shù)據(jù)發(fā)送
- allIdleTimeSeconds：以上兩種滿足其中一個(gè)即可
因?yàn)?IdleStateHandler 也是一種 ChannelHandler，所以會(huì)在 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);
}
}
也就是會(huì)按照我們給定的時(shí)間初始化出定時(shí)任務(wù)。
接著在任務(wù)真正執(zhí)行時(shí)進(jìn)行判斷：
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);
}
}
}
如果滿足條件則會(huì)生成一個(gè) IdleStateEvent 事件。
SpringBoot 監(jiān)控
由于整合了 SpringBoot 之后不但可以利用 Spring 幫我們管理對(duì)象，也可以利用它來做應(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)控端點(diǎn)供我們使用。
如一些應(yīng)用中的一些統(tǒng)計(jì)數(shù)據(jù)：
存在的 Beans：
更多信息請(qǐng)查看：https://docs.spring.io/spring-boot/docs/current/reference/html/production-ready-endpoints.html
但是如果我想監(jiān)控現(xiàn)在我的服務(wù)端有多少客戶端連上來了，分別的 ID 是多少？
其實(shí)就是實(shí)時(shí)查看我內(nèi)部定義的那個(gè)關(guān)聯(lián)關(guān)系的 Map。
這就需要暴露自定義端點(diǎn)了。
自定義端點(diǎn)
暴露的方式也很簡單：
繼承 AbstractEndpoint 并復(fù)寫其中的 invoke 函數(shù)：
public class CustomEndpoint extends AbstractEndpoint<Map<Long,NioSocketChannel>> {
/**
* 監(jiān)控端點(diǎn)的 訪問地址
* @param id
*/
public CustomEndpoint(String id) {
//false 表示不是敏感端點(diǎn)
super(id, false);
}
@Override
public Map<Long, NioSocketChannel> invoke() {
return NettySocketHolder.getMAP();
}
}
其實(shí)就是返回了 Map 中的數(shù)據(jù)。
再配置一個(gè)該類型的 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
這樣其實(shí)監(jiān)控功能已經(jīng)可以滿足了，但能不能展示的更美觀、并且多個(gè)應(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
在啟動(dòng)應(yīng)用之前先講 SpringBootAdmin 部署好：
這個(gè)應(yīng)用就是一個(gè)純粹的 SpringBoot ，只需要在主函數(shù)上加入 @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>
之后直接啟動(dòng)就行了。
這樣我們?cè)?SpringBootAdmin 的頁面中就可以查看很多應(yīng)用信息了。
更多內(nèi)容請(qǐng)參考官方指南：
http://codecentric.github.io/spring-boot-admin/1.5.6/
自定義監(jiān)控?cái)?shù)據(jù)
其實(shí)我們完全可以借助 actuator 以及這個(gè)可視化頁面幫我們監(jiān)控一些簡單的度量信息。
比如我在客戶端和服務(wù)端中寫了兩個(gè) Rest 接口用于向?qū)Ψ桨l(fā)送消息。
只是想要記錄分別發(fā)送了多少次：
客戶端
@Controller
@RequestMapping("/")
public class IndexController {
/**
* 統(tǒng)計(jì) 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é)
以上就是一個(gè)簡單 Netty 心跳示例，并演示了 SpringBoot 的監(jiān)控，之后會(huì)繼續(xù)更新 Netty 相關(guān)內(nèi)容，歡迎關(guān)注及指正。
本文所有代碼：
https://github.com/crossoverJie/netty-action
以上就是本文的全部內(nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。
相關(guān)文章
Java中byte輸出write到文件的實(shí)現(xiàn)方法講解
今天小編就為大家分享一篇關(guān)于Java中byte輸出write到文件的實(shí)現(xiàn)方法講解，小編覺得內(nèi)容挺不錯(cuò)的，現(xiàn)在分享給大家，具有很好的參考價(jià)值，需要的朋友一起跟隨小編來看看吧2019-03-03
面試官:Java中new Object()到底占用幾個(gè)字節(jié)
這篇文章主要介紹了面試官:Java中new Object()到底占用幾個(gè)字節(jié),本文給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友可以參考下2021-02-02
圖形界面（簡稱GUI）是指采用圖形方式顯示的計(jì)算機(jī)操作用戶界面。與早期計(jì)算機(jī)使用的命令行界面相比，圖形界面對(duì)于用戶來說在視覺上更易于接受，本篇精講Java語言中關(guān)于圖形用戶界面的單選按鈕2022-02-02
Java?Lombok實(shí)現(xiàn)手機(jī)號(hào)碼校驗(yàn)的示例代碼
手機(jī)號(hào)碼校驗(yàn)通常是系統(tǒng)開發(fā)中最基礎(chǔ)的功能之一，本文主要介紹了Java?Lombok實(shí)現(xiàn)手機(jī)號(hào)碼校驗(yàn)的示例代碼，文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2023-07-07
這篇文章主要介紹了springBoot整合redis使用案例詳解,本文通過圖文實(shí)例相結(jié)合給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友可以參考下2021-09-09
使用React和springboot做前后端分離項(xiàng)目的步驟方式
這篇文章主要介紹了使用React和springboot做前后端分離項(xiàng)目的步驟方式,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2023-08-08
這篇文章主要為大家詳細(xì)介紹了Struts2+uploadify多文件上傳實(shí)例，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2018-08-08
Java中shiro框架和security框架的區(qū)別
這篇文章主要介紹了Java中shiro框架和security框架的區(qū)別,shiro和security作為兩款流行的功能強(qiáng)大的且易于使用的java安全認(rèn)證框架，在近些年中的項(xiàng)目開發(fā)過程中使用廣泛，今天我們就來一起了解一下兩者的區(qū)別2023-08-08
java如何實(shí)現(xiàn)判斷文件的真實(shí)類型
本篇文章主要介紹了java如何實(shí)現(xiàn)判斷文件的真實(shí)類型，小編覺得挺不錯(cuò)的，現(xiàn)在分享給大家，也給大家做個(gè)參考。一起跟隨小編過來看看吧2017-08-08
在Java中實(shí)現(xiàn)讓線程按照自己指定的順序執(zhí)行
這篇文章主要介紹了在Java中實(shí)現(xiàn)讓線程按照自己指定的順序執(zhí)行，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教2022-06-06