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

阿新 • • 發佈：2018-12-04

實現的功能

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

在這實現的功能就是通過Scanner來輸入訊息得到服務端的迴應，超過設定的超時時間就觸發超時事件來進行心跳傳輸，如果服務端宕機客戶端就會一直髮起重連

所需要的依賴：

        <!-- 解碼and編碼器 -->
        <!-- https://mvnrepository.com/artifact/org.msgpack/msgpack --> 

        <dependency>
            <groupId>org.msgpack</groupId>
            <artifactId>msgpack</artifactId>
            <version>0.6.12</version>
        </dependency>
        <!-- netty 核心依賴 -->
        <!-- https://mvnrepository.com/artifact/io.netty/netty-all --> 

        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
            <version>4.1.6.Final</version>
        </dependency>

目錄結構：

匯入以上依賴
↓
建立配置模型model(模型類) ， TypeData(引數配置類)
↓
建立解碼and編碼器MsgPckDecode(解碼器) ，MsgPckEncode(編碼器)

↓
建立公用的控制器 Middleware
↓
建立客戶端及客戶端控制器client(客戶端啟動類) ， client3Handler(客戶端控制器)
↓
建立服務端以及控制器server(客戶端啟動類) ， server3Handler(客戶端控制器)

該類使用了msgpack ， It’s like JSON. but fast and small.

import java.io.Serializable;
import org.msgpack.annotation.Message;
/**
 * 訊息型別分離器
 * @author Administrator
 *
 */
@Message
public class Model implements Serializable{

    private static final long serialVersionUID = 1L;

    //型別
    private int type;

    //內容
    private String body;

    public String getBody() {
        return body;
    }

    public void setBody(String body) {
        this.body = body;
    }

    public int getType() {
        return type;
    }

    public void setType(int type) {
        this.type = type;
    }

    @Override
    public String toString() {
        return "Model [type=" + type + ", body=" + body + "]";
    }
}

編寫一個配置類介面,用於控制心跳包和應用訊息的處理

/**
 * 配置項
 * @author Administrator
 *
 */
public interface TypeData {

    byte PING = 1;

    byte PONG = 2;  
    //顧客
    byte CUSTOMER = 3;
}

MsgPckDecode(解碼器)

import java.util.List;
import org.msgpack.MessagePack;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageDecoder;

/**
 * 解碼器
 * @author Administrator
 *
 */
public class MsgPckDecode extends MessageToMessageDecoder<ByteBuf>{

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf msg,
            List<Object> out) throws Exception {

        final  byte[] array;

        final int length = msg.readableBytes();

        array = new byte[length];

        msg.getBytes(msg.readerIndex(), array, 0, length);

        MessagePack pack = new MessagePack();

        out.add(pack.read(array, Model.class));

    }
}

MsgPckEncode(編碼器)

import org.msgpack.MessagePack;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;

/**
 * 編碼器
 * @author Administrator
 *
 */
public class MsgPckEncode extends MessageToByteEncoder<Object>{

    @Override
    protected void encode(ChannelHandlerContext ctx, Object msg, ByteBuf buf)
            throws Exception {
        // TODO Auto-generated method stub
        MessagePack pack = new MessagePack();

        byte[] write = pack.write(msg);

        buf.writeBytes(write);

    }
}

建立公用的控制器 Middleware 這個就有意思了這就相當於一個樞紐站

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleStateEvent;

public abstract class Middleware extends ChannelInboundHandlerAdapter{

    protected String name;
    //記錄次數
    private int heartbeatCount = 0;

    //獲取server and client 傳入的值
    public Middleware(String name) {
        this.name = name;
    }
    /**
     *繼承ChannelInboundHandlerAdapter實現了channelRead就會監聽到通道里面的訊息
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg)
            throws Exception {
        Model m = (Model) msg; 
        int type = m.getType();
        switch (type) {
        case 1:
            sendPongMsg(ctx);
            break;
        case 2:
            System.out.println(name + " get  pong  msg  from" + ctx.channel().remoteAddress());
            break;
        case 3:
            handlerData(ctx,msg);
            break;  
        default:
            break;
        }
    }

    protected abstract void handlerData(ChannelHandlerContext ctx,Object msg);

    protected void sendPingMsg(ChannelHandlerContext ctx){
        Model model = new Model();

        model.setType(TypeData.PING);

        ctx.channel().writeAndFlush(model);

        heartbeatCount++;

        System.out.println(name + " send ping msg to " + ctx.channel().remoteAddress() + "count :" + heartbeatCount);
    }

    private void sendPongMsg(ChannelHandlerContext ctx) {

        Model model = new Model();

        model.setType(TypeData.PONG);

        ctx.channel().writeAndFlush(model);

        heartbeatCount++;

        System.out.println(name +" send pong msg to "+ctx.channel().remoteAddress() +" , count :" + heartbeatCount);
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt)
            throws Exception {
        IdleStateEvent stateEvent = (IdleStateEvent) evt;

        switch (stateEvent.state()) {
        case READER_IDLE:
            handlerReaderIdle(ctx);
            break;
        case WRITER_IDLE:
            handlerWriterIdle(ctx);
            break;
        case ALL_IDLE:
            handlerAllIdle(ctx);
            break;  
        default:
            break;
        }
    }

    protected void handlerAllIdle(ChannelHandlerContext ctx) {
        System.err.println("---ALL_IDLE---");       
    }

    protected void handlerWriterIdle(ChannelHandlerContext ctx) {
        System.err.println("---WRITER_IDLE---");        
    }


    protected void handlerReaderIdle(ChannelHandlerContext ctx) {
        System.err.println("---READER_IDLE---");    
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        // TODO Auto-generated method stub
        System.err.println(" ---"+ctx.channel().remoteAddress() +"----- is  action" );
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        // TODO Auto-generated method stub
        System.err.println(" ---"+ctx.channel().remoteAddress() +"----- is  inAction");
    }
}

建立client 客戶端：

import java.util.Scanner;
import java.util.concurrent.TimeUnit;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;

public class Client {

    private NioEventLoopGroup worker = new NioEventLoopGroup();

    private Channel channel;

    private Bootstrap bootstrap;

    public static void main(String[] args) {
        Client  client = new Client();

        client.start();

        client.sendData();      
    }

    private void start() {
        bootstrap = new Bootstrap();        
        bootstrap.group(worker)
        .channel(NioSocketChannel.class)
        .option(ChannelOption.TCP_NODELAY, true)
        .handler(new ChannelInitializer<Channel>() {
            @Override
            protected void initChannel(Channel ch) throws Exception {
                // TODO Auto-generated method stub
                ChannelPipeline pipeline = ch.pipeline();

                pipeline.addLast(new IdleStateHandler(0,0,5));

                pipeline.addLast(new MsgPckDecode());

                pipeline.addLast(new MsgPckEncode());

                pipeline.addLast(new Client3Handler(Client.this));              
            }           
        }); 
        doConnect();
    }

    /**
     * 連線服務端 and 重連
     */
    protected void doConnect() {

        if (channel != null && channel.isActive()){
            return;
        }       
        ChannelFuture connect = bootstrap.connect("127.0.0.1", 8081);
        //實現監聽通道連線的方法
        connect.addListener(new ChannelFutureListener() {

            @Override
            public void operationComplete(ChannelFuture channelFuture) throws Exception {

                if(channelFuture.isSuccess()){
                    channel = channelFuture.channel();
                    System.out.println("連線成功");
                }else{
                    System.out.println("每隔2s重連....");
                    channelFuture.channel().eventLoop().schedule(new Runnable() {

                        @Override
                        public void run() {
                            // TODO Auto-generated method stub
                            doConnect();
                        }
                    },2,TimeUnit.SECONDS);
                }   
            }
        });     
    }   
    /**
     * 向服務端傳送訊息
     */
    private void sendData() {
        Scanner sc= new Scanner(System.in); 
        for (int i = 0; i < 1000; i++) {

            if(channel != null && channel.isActive()){              
                //獲取一個鍵盤掃描器
                String nextLine = sc.nextLine();
                Model model = new Model();

                model.setType(TypeData.CUSTOMER);

                model.setBody(nextLine);

                channel.writeAndFlush(model);
            }
        }
    }
}

建立客戶端的handler控制器

import io.netty.channel.ChannelHandlerContext;
/**
*繼承我們自己編寫的樞紐站
*/
public class Client3Handler extends Middleware {

    private Client client;

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

    @Override
    protected void handlerData(ChannelHandlerContext ctx, Object msg) {
        // TODO Auto-generated method stub
        Model model = (Model) msg;
        System.out.println("client  收到資料： " + model.toString());
    }
    @Override
    protected void handlerAllIdle(ChannelHandlerContext ctx) {
        // TODO Auto-generated method stub
        super.handlerAllIdle(ctx);
        sendPingMsg(ctx);
    }   
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        // TODO Auto-generated method stub
        super.channelInactive(ctx);
        client.doConnect();
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
            throws Exception {      
        System.out.println(name + "exception :"+ cause.toString());
    }
}

建立服務端server、

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;

public class server {
    public static void main(String[] args) {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);

        EventLoopGroup workerGroup = new NioEventLoopGroup(4);
        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();

            serverBootstrap.group(bossGroup, workerGroup)
            .channel(NioServerSocketChannel.class)
            .localAddress(8081)
            .childHandler(new ChannelInitializer<Channel>() {

                @Override
                protected void initChannel(Channel ch) throws Exception {
                    // TODO Auto-generated method stub
                    ChannelPipeline pipeline = ch.pipeline();
                    pipeline.addLast(new IdleStateHandler(10,0,0));
                    pipeline.addLast(new MsgPckDecode());
                    pipeline.addLast(new MsgPckEncode());
                    pipeline.addLast(new server3Handler()); 
                }
            });         
            System.out.println("start server 8081 --");
            ChannelFuture sync = serverBootstrap.bind().sync();
            sync.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }finally{
            //優雅的關閉資源
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

建立serverHandler控制器

import io.netty.channel.ChannelHandlerContext;

public class server3Handler extends Middleware {

    public server3Handler() {
        super("server");
        // TODO Auto-generated constructor stub
    }
    @Override
    protected void handlerData(ChannelHandlerContext ctx, Object msg) {
        // TODO Auto-generated method stub
        Model model  = (Model) msg;
        System.out.println("server 接收資料 ： " +  model.toString());   
            model.setType(TypeData.CUSTOMER);
            model.setBody("---------------");
            ctx.channel().writeAndFlush(model);         
            System.out.println("server 傳送資料： " + model.toString()); 
    }
    @Override
    protected void handlerReaderIdle(ChannelHandlerContext ctx) {
        // TODO Auto-generated method stub
        super.handlerReaderIdle(ctx);
        System.err.println(" ---- client "+ ctx.channel().remoteAddress().toString() + " reader timeOut, --- close it");
        ctx.close();
    }   
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
            throws Exception {
        System.err.println( name +"  exception" + cause.toString());
    }   
}

先執行服務端，然後再啟動客戶端會根據設定的埠連線服務端，在客戶端輸入訊息就會得到服務端的迴應，如果超過5秒沒有進行讀寫就會觸發IdleStateHandler類超時事件來進行心跳包的傳輸，服務端未檢測到客戶端的讀寫或者心跳就會主動關閉channel通道

執行結果

server端
這裡寫圖片描述

client端
這裡寫圖片描述、

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

實現的功能

執行結果

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

Hadoop-hdfs的設計理念，block，namenode，datanode啟動過程，心跳機制，安全模式

Netty實現長連線簡單例子

高效實現長連線保活：手把手教你實現自適應的心跳保活機制

vue 彈性布局實現長圖垂直居上，短圖垂直居中

ActiveMQ的設置消息時長，事務，確認機制，持久化

TCP長連線與短連線、心跳機制

利用web work實現多執行緒非同步機制，打造頁面單步除錯IDE

長連線、短連線、心跳機制與斷線重連（轉載） Socket的長連線和短連線

15.TCP長連線與短連線、心跳機制

pytorch綜合多個弱分類器，投票機制，進行手寫數字分類（boosting）

Python基礎（四）--- Python多執行緒介紹，開啟執行緒的三種方式，time模組，join，Daemon，Lock、Rlock，事件機制，Timer

長連線及心跳保活原理簡介

JavaScript WebSocket實現長連線通訊連線重連

iOS使用GCDSocketManager實現長連線

Android之通過socket.io實現長連線

java執行過程,初始化的順序,建構函式,變數作用域，反射機制，面向物件的特徵

為什麼保持長連線需要心跳包？

Android socket通訊的長連線與心跳檢測

tcp重發機制，確認機制，視窗機制

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

實現的功能

執行結果

相關推薦