2. 程式人生 > >聊聊心跳機制及netty心跳實現

聊聊心跳機制及netty心跳實現

阿新 • • 發佈：2019-05-19

　　我們在使用netty的時候會使用一個引數，ChannelOption.SO_KEEPALIVE為true, 設定好了之後再Linux系統才會對keepalive生效，但是linux裡邊需要配置幾個引數，tcp_keepalive_time, tcp_keepalive_invl, tcp_keepalive_probes，如果不配置的時候都會是預設值。

　　tcp_keepalive_time 即給一個TCP連線傳送心跳包最後的時間間隔某一段時間後繼續傳送心跳包，允許空閒的時間，然後再次傳送心跳包，預設時間為7200秒，即2個小時發一次心跳包。

      tcp_keepalive_invl，傳送存活探測時候未收到對方回執的時候，需要間隔一段時間繼續傳送。預設為75秒。

　　tcp_keepalive_probes，如果發了存活探測的時候沒有收到對方的回執，那麼需要繼續傳送探測的次數，此時預設值為9次，也就是未收到回執的時候需要傳送9次。

　　再理一次，間隔tcp_keepalive_time之後傳送心跳探測，如果未收到對方回執的時候，需要間隔tcp_keepalive_invl設定的時間繼續傳送，一共需要傳送tcp_keepalive_probes的次數。　　

      這個是Linux系統的配置，如果要使用Linux的此功能需要設定SO_KEEPALIVE為true，同時設定其他幾個引數。系統預設的SO_KEEPALIVE為false。因為這些情況的差異，所以netty提供了自己實現心跳的機制。

　　netty有心跳的實現方法 IdleStateHandler，其中有讀空閒時間，寫空閒時間，讀寫空閒時間，只要有一個滿足條件會觸發userEventTriggered方法。

public IdleStateHandler(
            int readerIdleTimeSeconds,
            int writerIdleTimeSeconds,
            int allIdleTimeSeconds)

　　定義個訊息內容吧，長度為Type的長度1 + 實際內容的長度5 = 6。Length為2個位元組，Type為1個型別。

  +----------+----------+----------------+ 
  |  Length  |Type(byte)| Actual Content |
  |   0x06   |    1     |    "HELLO"     |    
  +----------+----------+----------------+

　　定義公共的inbound方法，用於進行channelRead, sendPing, sendPong, userEventTriggered 方法。

package com.hqs.heartbeat.common;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleStateEvent;

import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author huangqingshi
 * @Date 2019-05-11
 */
public abstract class CustomeHeartbeatHandler extends SimpleChannelInboundHandler<ByteBuf> {

    public static final byte PING = 1;
    public static final byte PONG = 2;
    public static final byte CUSTOM_MSG = 3;

    protected String name;
    private AtomicInteger heartbeatCount = new AtomicInteger(0);

    public CustomeHeartbeatHandler(String name) {
        this.name = name;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf) throws Exception {
        if(byteBuf.getByte(2) == PING) {
            sendPong(channelHandlerContext);
        } else if(byteBuf.getByte(2) == PONG) {
            System.out.println("get pong msg from " + channelHandlerContext
                    .channel().remoteAddress());
        } else {
            handleData(channelHandlerContext, byteBuf);
        }
    }

    protected abstract void handleData(ChannelHandlerContext channelHandlerContext,
                                       ByteBuf byteBuf);


    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("channel read : " + msg);
        ByteBuf byteBuf = (ByteBuf) msg;
        System.out.println(byteBuf.getByte(2));
        super.channelRead(ctx, msg);
    }

    protected void sendPong(ChannelHandlerContext channelHandlerContext) {
        ByteBuf buf = channelHandlerContext.alloc().buffer(3);
        buf.writeShort(3);
        buf.writeByte(PONG);
        channelHandlerContext.writeAndFlush(buf);
        heartbeatCount.incrementAndGet();
        System.out.println("send pong message to " + channelHandlerContext.channel().remoteAddress());
    }

    protected void sendPing(ChannelHandlerContext channelHandlerContext) {
        ByteBuf buf = channelHandlerContext.alloc().buffer(3);
        buf.writeShort(3);
        buf.writeByte(PING);
        channelHandlerContext.writeAndFlush(buf);
        heartbeatCount.incrementAndGet();
        System.out.println("send ping message to " + channelHandlerContext.channel().remoteAddress());
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if(evt instanceof IdleStateEvent){
            IdleStateEvent e = (IdleStateEvent) evt;
            switch (e.state()) {
                case ALL_IDLE:
                    handlALLIdle(ctx);
                    break;
                case READER_IDLE:
                    handlReadIdle(ctx);
                    break;
                case WRITER_IDLE:
                    handlWriteIdle(ctx);
                    break;
                 default:
                     break;
            }
        }
    }

    protected void handlReadIdle(ChannelHandlerContext channelHandlerContext) {
        System.out.println("READ_IDLE---");
    }

    protected void handlWriteIdle(ChannelHandlerContext channelHandlerContext) {
        System.out.println("WRITE_IDLE---");
    }

    protected void handlALLIdle(ChannelHandlerContext channelHandlerContext) {
        System.out.println("ALL_IDLE---");
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("channel:" + ctx.channel().remoteAddress() + " is active");
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("channel:" + ctx.channel().remoteAddress() + " is inactive");
    }
}

　　定義Server的方法，設定讀超時為10秒，採用固定長度方法進行內容分割：LengthFieldBasedFrameDecoder(1024, 0, 2, -2, 0)，長度為1K 。一個主執行緒接收請求，四個執行緒處理請求。埠號設定為9999。

package com.hqs.heartbeat.server;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.timeout.IdleStateHandler;

/**
 * @author huangqingshi
 * @Date 2019-05-11
 */
public class Server {

    public static void main(String[] args) {
        NioEventLoopGroup boss = new NioEventLoopGroup(1);
        NioEventLoopGroup worker = new NioEventLoopGroup(4);

        try {
            ServerBootstrap bootstrapServer = new ServerBootstrap();
            bootstrapServer.group(boss, worker).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    ChannelPipeline channelPipeline = ch.pipeline();
                    channelPipeline.addLast(new IdleStateHandler(10, 0, 0));
                    channelPipeline.addLast(new LengthFieldBasedFrameDecoder(1024, 0,2, -2, 0));
                    channelPipeline.addLast(new ServerHandler());
                }
            });
            Channel channel = bootstrapServer.bind(9999).sync().channel();
            channel.closeFuture().sync();
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            boss.shutdownGracefully();
            worker.shutdownGracefully();
        }
    }
}

　　Server的handler的處理方法：

package com.hqs.heartbeat.server;

import com.hqs.heartbeat.common.CustomeHeartbeatHandler;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;

/**
 * @author huangqingshi
 * @Date 2019-05-11
 */
public class ServerHandler extends CustomeHeartbeatHandler {

    public ServerHandler() {
        super("server");
    }

    @Override
    protected void handleData(ChannelHandlerContext channelHandlerContext,
                              ByteBuf byteBuf) {
        byte[] data = new byte[byteBuf.readableBytes() - 3];
        ByteBuf responseBuf = Unpooled.copiedBuffer(byteBuf);
        byteBuf.skipBytes(3);
        byteBuf.readBytes(data);
        String content = new String(data);
        System.out.println(name + " get content : " + content);
        channelHandlerContext.writeAndFlush(responseBuf);
    }

    @Override
    protected void handlReadIdle(ChannelHandlerContext channelHandlerContext) {
        super.handlReadIdle(channelHandlerContext);
        System.out.println(" client " + channelHandlerContext.channel().remoteAddress() + " reader timeout close it --");
        channelHandlerContext.close();
    }
}

　　定義Client類，所有超時時間為5秒，如果5秒沒有讀寫的話則傳送ping，如果失去連線之後inactive了就會重新連線，採用10秒出發一次。

package com.hqs.heartbeat.client;

import com.hqs.heartbeat.common.CustomeHeartbeatHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.timeout.IdleStateHandler;

import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * @author huangqingshi
 * @Date 2019-05-11
 */
public class Client {


    private NioEventLoopGroup workGroup = new NioEventLoopGroup(4);
    private Channel channel;
    private Bootstrap bootstrap;

    public static void main(String[] args) throws InterruptedException {
        Client client = new Client();
        client.start();
        client.sendData();
    }

    public void start() {

        try {
            bootstrap = new Bootstrap();

            bootstrap.group(workGroup).channel(NioSocketChannel.class).option(ChannelOption.SO_KEEPALIVE, true)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline channelPipeline = ch.pipeline()
                                    .addLast(new IdleStateHandler(0,0,5))
                                    .addLast(new LengthFieldBasedFrameDecoder(1024, 0, 2, -2, 0))
                                    .addLast(new ClientHandler(Client.this));
                        }
                    });
            doConnect();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    public void sendData() throws InterruptedException {
        Random random = new Random(System.currentTimeMillis());
        for(int i = 0; i < 10000; i++) {
            if(channel != null && channel.isActive()) {
                String content = "client msg " + i;
                ByteBuf byteBuf = channel.alloc().buffer(3 + content.getBytes().length);
                byteBuf.writeShort(3 + content.getBytes().length);
                byteBuf.writeByte(CustomeHeartbeatHandler.CUSTOM_MSG);
                byteBuf.writeBytes(content.getBytes());
                channel.writeAndFlush(byteBuf);
            }

            Thread.sleep(random.nextInt(20000));

        }

    }

    public void doConnect() {
        if(channel != null && channel.isActive()) {
            return;
        }

        ChannelFuture future = bootstrap
                .connect("127.0.0.1", 9999);
        future.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if(future.isSuccess()) {
                    channel = future.channel();
                    System.out.println("connect to server successfully");
                } else {
                    System.out.println("Failed to connect to server, try after 10s");

                    future.channel().eventLoop().schedule(new Runnable() {
                        @Override
                        public void run() {
                            doConnect();
                        }
                    }, 10, TimeUnit.SECONDS);
                }
            }
        });
    }

}

　　定義clientHandler方法，讀取時跳過長度+型別 2+1 三個位元組，然後獲取訊息。連線斷開之後則進行重連。

package com.hqs.heartbeat.client;

import com.hqs.heartbeat.common.CustomeHeartbeatHandler;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;

/**
 * @author huangqingshi
 * @Date 2019-05-11
 */
public class ClientHandler extends CustomeHeartbeatHandler {

    private Client client;

    public ClientHandler(Client client) {
        super("client");
        this.client = client;
    }

    @Override
    protected void handleData(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf) {
        byte[] data = new byte[byteBuf.readableBytes() - 3];
        byteBuf.skipBytes(3);
        byteBuf.readBytes(data);
        String content = new String(data);
        System.out.println(name + " get content:" + content);
    }

    @Override
    protected void handlALLIdle(ChannelHandlerContext channelHandlerContext) {
        super.handlALLIdle(channelHandlerContext);
        sendPing(channelHandlerContext);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        super.channelInactive(ctx);
        client.doConnect();
    }
}

 

　　好了，總體的netty心跳實現機制就這麼多，希望能幫助到大家。

　　github地址：https://github.com/stonehqs/heartbeat

相關推薦

聊聊心跳機制及netty心跳實現

　　我們在使用netty的時候會使用一個引數，ChannelOption.SO_KEEPALIVE為true, 設定好了之後再Linux系統才會對keepalive生效，但是linux裡邊需要配置幾個引數，tcp_keepalive_time, tcp_keepalive_invl, tcp_keepaliv

Android TCP長連線 心跳機制及實現

維護任何一個長連線都需要心跳機制，客戶端傳送一個心跳給伺服器，伺服器給客戶端一個心跳應答， 這樣雙方都知道他們之間的連線是沒有斷開。【客戶端先發送給服務端】 如果超過一個時間的閾值，客戶端沒有收到伺服器的應答，或者伺服器沒有收到客戶端的心跳， 那麼對

Http Cookie機制及Cookie的實現原理

轉自：https://itbilu.com/other/relate/4J4n8fIPe.html Cookie是進行網站使用者身份，實現服務端Session會話持久化的一種非常好方式。Cookie最早由Netscape公司開發，現在由 IETF 的RFC 6265標準備對其規範，已被所有主流瀏

Java併發機制及鎖的實現原理

Java併發程式設計概述 併發程式設計的目的是為了讓程式執行得更快，但是，並不是啟動更多的執行緒就能讓程式最大限度地併發執行。在進行併發程式設計時，如果希望通過多執行緒執行任務讓程式執行得更快，會面臨非常多的挑戰，比如上下文切換的問題、死鎖的問題，以及受限於硬體和軟體的資源

Netty 超時機制及心跳程序實現

timeunit pip demo ade reads unpooled 一個 apt group 本文介紹了 Netty 超時機制的原理，以及如何在連接閑置時發送一個心跳來維持連接。 Netty 超時機制的介紹 Netty 的超時類型 IdleState 主要分為： AL

心跳機制tcp keepalive的討論、應用及“斷網”、"斷電"檢測的C代碼實現(Windows環境下)

astar har 心跳 存在 假設 clu ali clean struct 版權聲明：本文為博主原創文章，轉載時請務必註明本文地址， 禁止用於任何商業用途， 否則會用法律維權。 https://blog.csdn.net/stpeace/article/details/

Netty 實現心跳機制.md

ann 構造 red 通過 case 程序 evel exc 基本使用 netty 心跳機制示例，使用netty4，IdleStateHandler 實現。 本文假設你已經了解了Netty的使用，或者至少寫過netty的helloworld，知道了netty的基本使用。我們

netty 實現長連線，心跳機制，以及重連

實現的功能 心跳機制 and 長連線 and 重連機制 and 應用訊息傳輸： 這裡最關鍵的就是通過netty中的 IdleStateHandler 超時機制來實現心跳和重連 然後通過org.msgpack編碼器來實現跨平臺資料傳輸， 在這實現的功能就是通過Scanner來輸

基於Netty實現的WebSocket聊天室--支援多人同時線上及定時心跳檢測

專案簡介 基於Netty實現的WebSocket聊天室，實現的功能如下： 支援暱稱登入； 支援多人同時線上； 同步顯示線上人數； 支援文字和表情的內容； 瀏覽器與伺服器保持長連線，定時心跳檢測；

C#之實現Scoket心跳機制

TCP網路長連線 手機能夠使用聯網功能是因為手機底層實現了TCP/IP協議，可以使手機終端通過無線網路建立TCP連線。TCP協議可以對上層網路提供介面，使上層網路資料的傳輸建立在“無差別”的網路之上。 建立起一個TCP連線需要經過“三次握手”： 第一次握手：客戶端傳送syn包(syn=j)到伺服

螞蟻金服通訊框架 SOFABolt 解析 | 超時控制機制及心跳機制

SOFA Scalable Open Financial Architecture 是螞蟻金服自主研發的金融級分散式中介軟體，包含了構建金融級雲原生架構所需的各個元件，是在金融場景裡錘鍊出來的最佳實踐。 本文為《螞蟻金服通訊框架 SOFABolt 解析》系列第五篇，作者胡蘿蔔、丞一。

Netty4實現心跳檢測及IdleStateHandler原始碼分析

1.什麼是心跳檢測？     判斷對方是否正常執行，一般採用定時傳送簡單的通訊包，如果在指定時間內未接收到對方響應，則判定對方已經宕掉。用於檢測TCP的異常斷開。     心跳包一般就是客戶端傳送給服務端的簡單訊息，如果服務端幾分鐘內沒有收到客戶端

螞蟻金服通訊框架SOFABolt解析|超時控制機制及心跳機制

SOFAScalable Open Financial Architecture 是螞蟻金服自主研發的金融級分散式中介軟體，包含了構建金融級雲原生架構所需的各個元件，是在金融場景裡錘鍊出來的最佳實踐。 本文為《螞蟻金服通訊框架 SOFABolt 解析》系列第五篇，作

NETTY 心跳機制

end value trap parseint acceptor blog channel 出發 每次 最近工作比較忙，但閑暇之余還是看了阿裏的馮家春（fengjiachun）的github上的開源代碼Jupiter，寫的RPC框架讓我感嘆人外有人，廢話不多說，下面的代碼全

Python3 Socket與Socket心跳機制簡單實現

什麼是 Socket? Socket又稱"套接字"，應用程式通常通過"套接字"向網路發出請求或者應答網路請求，使主機間或者一臺計算機上的程序間可以通訊。 socket()函式 Python 中，我們用 socket（）函式來建立套接字，語法格式如下： socket.s

Netty(一) SpringBoot 整合長連線心跳機制

前言 Netty 是一個高效能的 NIO 網路框架，本文基於 SpringBoot 以常見的心跳機制來認識 Netty。 最終能達到的效果： 客戶端每隔 N 秒檢測是否需要傳送心跳。 服務端也每隔 N 秒檢測是否需要傳送心跳。 服務端可以主動 push 訊息到客戶端。 基於 SpringBo

回合制遊戲網路通訊協議及心跳機制調研

回合制策略遊戲 回合制策略遊戲是策略遊戲的一種子型別，所有的玩家輪流自己的回合，只有自己的回合，才能夠進行操縱。早期的戰略由於硬體運算能力有限，在考量遊戲樂趣的情況下，多半採取這種型式。 主要分類 戰棋類遊戲 SLG:角色扮演因素較少，戰鬥以整體策

Netty學習(八)-Netty的心跳機制

1. 如何實現心跳機制一般實現心跳機制由兩種方式： TCP協議自帶的心跳機制來實現；在應用層來實現。但是TCP協議自帶的心跳機制系統預設是設定的是2小時的心跳頻率。它檢查不到機器斷電、網線拔出、防火牆這些斷線。而且邏輯層處理斷線可能也不是那麼好處理。另外該心跳機制是與TCP協議繫結的，那如果我們要是使用

java零碎要點009---java實現伺服器心跳機制,TCP握手

什麼是心跳包？ 心跳包就是在客戶端和伺服器間定時通知對方自己狀態的一個自己定義的命令字，按照一定的時間間隔傳送，類似於心跳，所以叫做心跳包。 　　  用來判斷對方（裝置，程序或其它網元）是否正常執行，採用定時傳送簡單的通訊包，如果在指定時間段內未收到對方響應，則判斷對方已經離線。

Socket長連線Android端心跳機制實現

1. 把socket連結和心跳功能都放在一個Service中，為什麼要放在Service中？ 一般我們這種socket幾乎是跟app的生命週期一樣長，甚至更長。不管在不在Service中去完成操作，我們都得開非同步執行緒，雖然Service並不是