Java Netty實現(xiàn)心跳機制過程解析
netty心跳機制示例，使用Netty實現(xiàn)心跳機制，使用netty4，IdleStateHandler 實現(xiàn)。Netty心跳機制，netty心跳檢測，netty,心跳
本文假設(shè)你已經(jīng)了解了Netty的使用，或者至少寫過netty的helloworld，知道了netty的基本使用。我們知道使用netty的時候，大多數(shù)的東西都與Handler有關(guān)，我們的業(yè)務(wù)邏輯基本都是在Handler中實現(xiàn)的。Netty中自帶了一個IdleStateHandler 可以用來實現(xiàn)心跳檢測。
心跳檢測的邏輯
本文中我們將要實現(xiàn)的心跳檢測邏輯是這樣的：服務(wù)端啟動后，等待客戶端連接，客戶端連接之后，向服務(wù)端發(fā)送消息。如果客戶端在“干活”那么服務(wù)端必定會收到數(shù)據(jù)，如果客戶端“閑下來了”那么服務(wù)端就接收不到這個客戶端的消息，既然客戶端閑下來了，不干事，那么何必浪費連接資源呢？所以服務(wù)端檢測到一定時間內(nèi)客戶端不活躍的時候，將客戶端連接關(guān)閉。本文要實現(xiàn)的邏輯步驟為：
- 啟動服務(wù)端，啟動客戶端
- 客戶端向服務(wù)端發(fā)送"I am alive"，并sleep隨機時間，用來模擬空閑。
- 服務(wù)端接收客戶端消息，并返回"copy that"，客戶端空閑時 計數(shù)+1.
- 服務(wù)端客戶端繼續(xù)通信
- 服務(wù)端檢測客戶端空閑太多，關(guān)閉連接?？蛻舳税l(fā)現(xiàn)連接關(guān)閉了，就退出了。
有了這個思路，我們先來編寫服務(wù)端。
心跳檢測服務(wù)端代碼
public class HeartBeatServer {
int port ;
public HeartBeatServer(int port){
this.port = port;
}
public void start(){
ServerBootstrap bootstrap = new ServerBootstrap();
EventLoopGroup boss = new NioEventLoopGroup();
EventLoopGroup worker = new NioEventLoopGroup();
try{
bootstrap.group(boss,worker)
.handler(new LoggingHandler(LogLevel.INFO))
.channel(NioServerSocketChannel.class)
.childHandler(new HeartBeatInitializer());
ChannelFuture future = bootstrap.bind(port).sync();
future.channel().closeFuture().sync();
}catch(Exception e){
e.printStackTrace();
}finally {
worker.shutdownGracefully();
boss.shutdownGracefully();
}
}
public static void main(String[] args) throws Exception {
HeartBeatServer server = new HeartBeatServer(8090);
server.start();
}
}
熟悉netty的同志，對于上面的模板一樣的代碼一定是在熟悉不過了。啥都不用看，只需要看childHandler(new HeartBeatInitializer()) 這一句。HeartBeatInitializer就是一個ChannelInitializer顧名思義，他就是在初始化channel的時做一些事情。我們所需要開發(fā)的業(yè)務(wù)邏輯Handler就是在這里添加的。其代碼如下：
public class HeartBeatInitializer extends ChannelInitializer<Channel> {
@Override
protected void initChannel(Channel channel) throws Exception {
ChannelPipeline pipeline = channel.pipeline();
pipeline.addLast("decoder", new StringDecoder());
pipeline.addLast("encoder", new StringEncoder());
pipeline.addLast(new IdleStateHandler(2,2,2, TimeUnit.SECONDS));
pipeline.addLast(new HeartBeatHandler());
}
}
代碼很簡單，我們先添加了StringDecoder,和StringEncoder。這兩個其實就是編解碼用的，下面的IdleStateHandler才是本次心跳的核心組件。我們可以看到IdleStateHandler的構(gòu)造函數(shù)中接收了4個參數(shù)，其定義如下：
public IdleStateHandler(long readerIdleTime, long writerIdleTime, long allIdleTime, TimeUnit unit);
三個空閑時間參數(shù)，以及時間參數(shù)的格式。我們的例子中設(shè)置的是2,2,2，意思就是客戶端2秒沒有讀/寫，這個超時時間就會被觸發(fā)。超時事件觸發(fā)就需要我們來處理了，這就是上的HeartBeatInitializer中最后一行的HeartBeatHandler所做的事情。代碼如下：
public class HeartBeatHandler extends SimpleChannelInboundHandler<String> {
int readIdleTimes = 0;
@Override
protected void channelRead0(ChannelHandlerContext ctx, String s) throws Exception {
System.out.println(" ====== > [server] message received : " + s);
if("I am alive".equals(s)){
ctx.channel().writeAndFlush("copy that");
}else {
System.out.println(" 其他信息處理 ... ");
}
}
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
IdleStateEvent event = (IdleStateEvent)evt;
String eventType = null;
switch (event.state()){
case READER_IDLE:
eventType = "讀空閑";
readIdleTimes ++; // 讀空閑的計數(shù)加1
break;
case WRITER_IDLE:
eventType = "寫空閑";
// 不處理
break;
case ALL_IDLE:
eventType ="讀寫空閑";
// 不處理
break;
}
System.out.println(ctx.channel().remoteAddress() + "超時事件：" +eventType);
if(readIdleTimes > 3){
System.out.println(" [server]讀空閑超過3次，關(guān)閉連接");
ctx.channel().writeAndFlush("you are out");
ctx.channel().close();
}
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
System.err.println("=== " + ctx.channel().remoteAddress() + " is active ===");
}
}
至此，我們的服務(wù)端寫好了。
心跳檢測客戶端代碼
netty的api設(shè)計使得編碼的模式非常具有通用性，所以客戶端代碼和服務(wù)端的代碼幾乎一樣：啟動client端的代碼幾乎一樣，也需要一個ChannelInitializer，也需要Handler。改動的地方很少，因此本文不對客戶端代碼進行詳細解釋。下面給出client端的完整代碼：
public class HeartBeatClient {
int port;
Channel channel;
Random random ;
public HeartBeatClient(int port){
this.port = port;
random = new Random();
}
public static void main(String[] args) throws Exception{
HeartBeatClient client = new HeartBeatClient(8090);
client.start();
}
public void start() {
EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
try{
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class)
.handler(new HeartBeatClientInitializer());
connect(bootstrap,port);
String text = "I am alive";
while (channel.isActive()){
sendMsg(text);
}
}catch(Exception e){
// do something
}finally {
eventLoopGroup.shutdownGracefully();
}
}
public void connect(Bootstrap bootstrap,int port) throws Exception{
channel = bootstrap.connect("localhost",8090).sync().channel();
}
public void sendMsg(String text) throws Exception{
int num = random.nextInt(10);
Thread.sleep(num * 1000);
channel.writeAndFlush(text);
}
static class HeartBeatClientInitializer extends ChannelInitializer<Channel> {
@Override
protected void initChannel(Channel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast("decoder", new StringDecoder());
pipeline.addLast("encoder", new StringEncoder());
pipeline.addLast(new HeartBeatClientHandler());
}
}
static class HeartBeatClientHandler extends SimpleChannelInboundHandler<String> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
System.out.println(" client received :" +msg);
if(msg!= null && msg.equals("you are out")) {
System.out.println(" server closed connection , so client will close too");
ctx.channel().closeFuture();
}
}
}
}
運行代碼
在上面的代碼寫好之后，我們先啟動服務(wù)端，然后在啟動客戶端。運行日志如下：
server端：
=== /127.0.0.1:57700 is active === ====== > [server] message received : I am alive ====== > [server] message received : I am alive /127.0.0.1:57700超時事件：寫空閑 /127.0.0.1:57700超時事件：讀空閑 /127.0.0.1:57700超時事件：讀寫空閑 /127.0.0.1:57700超時事件：寫空閑 /127.0.0.1:57700超時事件：讀空閑 /127.0.0.1:57700超時事件：讀寫空閑 /127.0.0.1:57700超時事件：寫空閑 ====== > [server] message received : I am alive /127.0.0.1:57700超時事件：寫空閑 /127.0.0.1:57700超時事件：讀寫空閑 /127.0.0.1:57700超時事件：讀空閑 /127.0.0.1:57700超時事件：寫空閑 /127.0.0.1:57700超時事件：讀寫空閑 /127.0.0.1:57700超時事件：讀空閑 [server]讀空閑超過3次，關(guān)閉連接
client端：
client sent msg and sleep 2 client received :copy that client received :copy that client sent msg and sleep 6 client sent msg and sleep 6 client received :copy that client received :you are out server closed connection , so client will close too Process finished with exit code 0
通過上面的運行日志，我們可以看到：
1.客戶端在與服務(wù)器成功建立之后，發(fā)送了3次'I am alive'，服務(wù)端也回應(yīng)了3次：'copy that'
2.由于客戶端消極怠工，超時了多次，服務(wù)端關(guān)閉了鏈接。
3.客戶端知道服務(wù)端拋棄自己之后，也關(guān)閉了連接，程序退出。
以上簡單了演示了一下，netty的心跳機制，其實主要就是使用了IdleStateHandler。源碼下載
以上就是本文的全部內(nèi)容，希望對大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。
相關(guān)文章
java多線程編程之使用thread類創(chuàng)建線程
在Java中創(chuàng)建線程有兩種方法：使用Thread類和使用Runnable接口。在使用Runnable接口時需要建立一個Thread實例2014-01-01
這篇文章主要介紹了配置javaw.exe雙擊運行jar包方式,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2024-01-01
一文看懂springboot實現(xiàn)短信服務(wù)功能
項目中的短信服務(wù)基本上上都會用到，簡單的注冊驗證碼，消息通知等等都會用到。這篇文章主要介紹了springboot 實現(xiàn)短信服務(wù)功能,需要的朋友可以參考下2019-10-10
Spring?Boot實現(xiàn)MyBatis動態(tài)創(chuàng)建表的操作語句
這篇文章主要介紹了Spring?Boot實現(xiàn)MyBatis動態(tài)創(chuàng)建表,MyBatis提供了動態(tài)SQL,我們可以通過動態(tài)SQL,傳入表名等信息然組裝成建表和操作語句,本文通過案例講解展示我們的設(shè)計思路,需要的朋友可以參考下2024-01-01
這篇文章主要介紹了java中VO的使用解析，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-10-10
這篇文章主要介紹了Java將微信和支付寶支付的個二維碼合二為一的方法，文中通過示例代碼介紹的非常詳細，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2019-08-08
這篇文章主要為大家詳細介紹了java自定義封裝StringUtils常用工具類，具有一定的參考價值，感興趣的小伙伴們可以參考一下2018-03-03
idea快捷鍵生成getter和setter,有構(gòu)造參數(shù),無構(gòu)造參數(shù),重寫toString方式
這篇文章主要介紹了java之idea快捷鍵生成getter和setter,有構(gòu)造參數(shù),無構(gòu)造參數(shù),重寫toString方式,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2023-11-11
本文通過實例代碼給大家分享了模擬Mybatis的實現(xiàn)方法，需要的朋友參考下吧2017-09-09
這篇文章主要介紹了Java中的snowflake算法詳解,Snowflake算法產(chǎn)生是為了滿足Twitter每秒上萬條消息的請求,每條消息都必須分配一條唯一的id,這些id還需要一些大致的順序,并且在分布式系統(tǒng)中不同機器產(chǎn)生的id必須不同,需要的朋友可以參考下2023-08-08