前言

Netty 是一個(gè)高性能的 NIO 網(wǎng)絡(luò)框架，本文基于 SpringBoot 以常見的心跳機(jī)制來認(rèn)識(shí) Netty。

最終能達(dá)到的效果：

客戶端每隔 N 秒檢測(cè)是否需要發(fā)送心跳。 服務(wù)端也每隔 N 秒檢測(cè)是否需要發(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。

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

@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('啟動(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ì)檢測(cè)是否發(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

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; /** * 啟動(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(cè)試是有兩個(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 即可：

@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

這樣其實(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@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>

之后直接啟動(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í)有所幫助，也希望大家多多支持好吧啦網(wǎng)。