Netty 是一個高性能的 NIO 網絡框架,本文基於 SpringBoot 以常見的心跳機制來認識 Netty。html
最終能達到的效果:java
效果以下:git
Netty 可使用 IdleStateHandler 來實現鏈接管理,當鏈接空閒時間太長(沒有發送、接收消息)時則會觸發一個事件,咱們即可在該事件中實現心跳機制。github
當客戶端空閒了 N 秒沒有給服務端發送消息時會自動發送一個心跳來維持鏈接。spring
核心代碼代碼以下:bootstrap
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") ; } }
這裏涉及到了自定義協議的內容,請繼續查看下文。app
固然少不了啓動引導:框架
@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 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, 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); }
其實就是初始化了幾個數據:
由於 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, 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); } } }
若是知足條件則會生成一個 IdleStateEvent 事件。
因爲整合了 SpringBoot 以後不但能夠利用 Spring 幫咱們管理對象,也能夠利用它來作應用監控。
當咱們爲引入了:
<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
來訪問了:
一個客戶端鏈接時:
兩個客戶端鏈接時:
這樣其實監控功能已經能夠知足了,但能不能展現的更美觀、而且多個應用也能夠方便查看呢?
有這樣的開源工具幫咱們作到了:
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 ,利用它來幫咱們記錄數據。
當咱們調用該接口時:
在監控頁面中能夠查詢剛纔的調用狀況:
服務端主動 push 消息也是相似,只是須要在發送時候根據客戶端的 ID 查詢到具體的 Channel 發送:
以上就是一個簡單 Netty 心跳示例,並演示了 SpringBoot 的監控,以後會繼續更新 Netty 相關內容,歡迎關注及指正。
本文全部代碼:
https://github.com/crossoverJie/netty-action
最近在總結一些 Java 相關的知識點,感興趣的朋友能夠一塊兒維護。
地址: https://github.com/crossoverJie/Java-Interview