SpringBoot整合Netty心跳機制過程詳解
阿新 • • 發佈：2020-02-27
前言
Netty 是一個高效能的 NIO 網路框架，本文基於 SpringBoot 以常見的心跳機制來認識 Netty。
最終能達到的效果：
- 客戶端每隔 N 秒檢測是否需要傳送心跳。
- 服務端也每隔 N 秒檢測是否需要傳送心跳。
- 服務端可以主動 push 訊息到客戶端。
- 基於 SpringBoot 監控，可以檢視實時連線以及各種應用資訊。
IdleStateHandler
Netty 可以使用 IdleStateHandler 來實現連線管理，當連線空閒時間太長（沒有傳送、接收訊息）時則會觸發一個事件，我們便可在該事件中實現心跳機制。
客戶端心跳
當客戶端空閒了 N 秒沒有給服務端傳送訊息時會自動傳送一個心跳來維持連線。
核心程式碼程式碼如下：
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("已經 10 秒沒有傳送資訊！");
//向服務端傳送訊息
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 {
//從服務端收到訊息時被呼叫
LOGGER.info("客戶端收到訊息={}",in.toString(CharsetUtil.UTF_8)) ;
}
} 實現非常簡單，只需要在事件回撥中傳送一個訊息即可。
由於整合了 SpringBoot ，所以傳送的心跳資訊是一個單例的 Bean。
@Configuration
public class HeartBeatConfig {
@Value("${channel.id}")
private long id ;
@Bean(value = "heartBeat")
public CustomProtocol heartBeat(){
return new CustomProtocol(id,"ping") ;
}
}
這裡涉及到了自定義協議的內容，請繼續檢視下文。
當然少不了啟動引導：
@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("啟動 Netty 成功");
}
channel = (SocketChannel) future.channel();
}
}
public class CustomerHandleInitializer extends ChannelInitializer<Channel> {
@Override
protected void initChannel(Channel ch) throws Exception {
ch.pipeline()
//10 秒沒傳送訊息 將IdleStateHandler 新增到 ChannelPipeline 中
.addLast(new IdleStateHandler(0,10,0))
.addLast(new HeartbeatEncode())
.addLast(new EchoClientHandle())
;
}
}
所以當應用啟動每隔 10 秒會檢測是否傳送過訊息，不然就會發送心跳資訊。
服務端心跳
伺服器端的心跳其實也是類似，也需要在 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("已經5秒沒有收到資訊！");
//向客戶端傳送訊息
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 之間的關係
NettySocketHolder.put(customProtocol.getId(),(NioSocketChannel)ctx.channel()) ;
}
}
這裡有點需要注意：
當有多個客戶端連上來時，服務端需要區分開，不然響應訊息就會發生混亂。
所以每當有個連線上來的時候，我們都將當前的 Channel 與連上的客戶端 ID 進行關聯（因此每個連上的客戶端 ID 都必須唯一）。
這裡採用了一個 Map 來儲存這個關係，並且在斷開連線時自動取消這個關聯。
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()));
}
}
啟動載入程式：
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;
/**
* 啟動 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("關閉 Netty 成功");
}
}
public class HeartbeatInitializer extends ChannelInitializer<Channel> {
@Override
protected void initChannel(Channel ch) throws Exception {
ch.pipeline()
//五秒沒有收到訊息 將IdleStateHandler 新增到 ChannelPipeline 中
.addLast(new IdleStateHandler(5,0))
.addLast(new HeartbeatDecoder())
.addLast(new HeartBeatSimpleHandle());
}
}
也是同樣將IdleStateHandler 新增到 ChannelPipeline 中，也會有一個定時任務，每5秒校驗一次是否有收到訊息，否則就主動傳送一次請求。
因為測試是有兩個客戶端連上所以有兩個日誌。
自定義協議
上文其實都看到了：服務端與客戶端採用的是自定義的 POJO 進行通訊的。
所以需要在客戶端進行編碼，服務端進行解碼，也都只需要各自實現一個編解碼器即可。
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()) ;
}
}
也就是說訊息的前八個位元組為 header，剩餘的全是 content。
服務端的解碼器：
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) ;
}
}
只需要按照剛才的規則進行解碼即可。
實現原理
其實聯想到 IdleStateHandler 的功能，自然也能想到它實現的原理：
應該會存在一個定時任務的執行緒去處理這些訊息。
來看看它的原始碼：
首先是建構函式:
public IdleStateHandler(
int readerIdleTimeSeconds,int writerIdleTimeSeconds,int allIdleTimeSeconds) {
this(readerIdleTimeSeconds,writerIdleTimeSeconds,allIdleTimeSeconds,TimeUnit.SECONDS);
}
其實就是初始化了幾個資料：
- readerIdleTimeSeconds：一段時間內沒有資料讀取
- writerIdleTimeSeconds：一段時間內沒有資料傳送
- allIdleTimeSeconds：以上兩種滿足其中一個即可
因為 IdleStateHandler 也是一種 ChannelHandler，所以會在 channelActive 中初始化任務：
@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);
}
}
也就是會按照我們給定的時間初始化出定時任務。
接著在任務真正執行時進行判斷：
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,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,nextDelay,TimeUnit.NANOSECONDS);
}
}
}
如果滿足條件則會生成一個 IdleStateEvent 事件。
SpringBoot 監控
由於整合了 SpringBoot 之後不但可以利用 Spring 幫我們管理物件，也可以利用它來做應用監控。
actuator 監控
當我們為引入了:
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
就開啟了 SpringBoot 的 actuator 監控功能，他可以暴露出很多監控端點供我們使用。
如一些應用中的一些統計資料：
存在的 Beans：
更多資訊請檢視：https://docs.spring.io/spring-boot/docs/current/reference/html/production-ready-endpoints.html
但是如果我想監控現在我的服務端有多少客戶端連上來了，分別的 ID 是多少？
其實就是實時檢視我內部定義的那個關聯關係的 Map。
這就需要暴露自定義端點了。
自定義端點
暴露的方式也很簡單：
繼承 AbstractEndpoint 並複寫其中的 invoke 函式：
public class CustomEndpoint extends AbstractEndpoint<Map<Long,NioSocketChannel>> {
/**
* 監控端點的 訪問地址
* @param id
*/
public CustomEndpoint(String id) {
//false 表示不是敏感端點
super(id,false);
}
@Override
public Map<Long,NioSocketChannel> invoke() {
return NettySocketHolder.getMAP();
}
}
其實就是返回了 Map 中的資料。
再配置一個該型別的 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
這樣其實監控功能已經可以滿足了，但能不能展示的更美觀、並且多個應用也可以方便檢視呢？
有這樣的開源工具幫我們做到了：
https://github.com/codecentric/spring-boot-admin
簡單來說我們可以利用該工具將 actuator 暴露出來的介面視覺化並聚合的展示在頁面中：
接入也很簡單，首先需要引入依賴：
<dependency>
<groupId>de.codecentric</groupId>
<artifactId>spring-boot-admin-starter-client</artifactId>
</dependency>
並在配置檔案中加入：
# 關閉健康檢查許可權 management.security.enabled=false # SpringAdmin 地址 spring.boot.admin.url=http://127.0.0.1:8888
在啟動應用之前先講 SpringBootAdmin 部署好：
這個應用就是一個純粹的 SpringBoot ，只需要在主函式上加入 @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>
之後直接啟動就行了。
這樣我們在 SpringBootAdmin 的頁面中就可以檢視很多應用資訊了。
更多內容請參考官方指南：
http://codecentric.github.io/spring-boot-admin/1.5.6/
自定義監控資料
其實我們完全可以藉助 actuator 以及這個視覺化頁面幫我們監控一些簡單的度量資訊。
比如我在客戶端和服務端中寫了兩個 Rest 介面用於向對方傳送訊息。
只是想要記錄分別傳送了多少次：
客戶端
@Controller
@RequestMapping("/")
public class IndexController {
/**
* 統計 service
*/
@Autowired
private CounterService counterService;
@Autowired
private HeartbeatClient heartbeatClient ;
/**
* 向服務端發訊息
* @param sendMsgReqVO
* @return
*/
@ApiOperation("客戶端傳送訊息")
@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 ，利用它來幫我們記錄資料。
總結
以上就是一個簡單 Netty 心跳示例，並演示了 SpringBoot 的監控，之後會繼續更新 Netty 相關內容，歡迎關注及指正。
本文所有程式碼：
https://github.com/crossoverJie/netty-action
以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支援我們。