概述

Netty是由JBOSS提供的一個java開源框架。Netty提供異步的、事件驅動的網絡應用程序框架和工具，用以快速開發高性能、高可靠性的網絡服務器和客戶端程序。也就是說，Netty 是一個基於NIO的客戶、服務器端編程框架，使用Netty 可以確保你快速和簡單的開發出一個網絡應用，例如實現了某種協議的客戶、服務端應用。Netty相當於簡化和流線化了網絡應用的編程開發過程，例如：基於TCP和UDP的socket服務開發。 “快速”和“簡單”並不用產生維護性或性能上的問題。Netty 是一個吸收了多種協議（包括FTP、SMTP、HTTP等各種二進制文本協議）的實現經驗，並經過相當精心設計的項目。最終，Netty 成功的找到了一種方式，在保證易於開發的同時還保證了其應用的性能，穩定性和伸縮性。

詳細

代碼下載：http://www.demodashi.com/demo/15012.html

詳細

本篇demo實現的功能是基於netty的心跳機制和長連接以及重連機制，最關鍵的就是通過netty中的 IdleStateHandler 的超時機制來實現心跳和重連，然后通過org.msgpack編碼器來實現跨平台數據傳輸，

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

一、運行效果

服務端：

客戶端：

二、實現過程

在maven pom文件添加依賴：

       <!-- 解碼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.33.Final</version>
	  </dependency>

導入以上依賴 
↓ 
創建配置模型model(模型類) ， TypeData(參數配置類) 
↓ 
創建解碼and編碼器MsgPckDecode(解碼器) ，MsgPckEncode(編碼器) 
↓ 
創建各自的控制器 AbstractClientChannelInboundHandleAdapter，AbstractServerChannelInboundHandleAdapter
↓ 
創建客戶端及客戶端控制器Client(客戶端啟動類) ， ClientHandler(客戶端控制器) 
↓ 
創建服務端以及控制器Server(客戶端啟動類) ， ServerHandler(客戶端控制器)

ps:本demo使用了msgpack ， It’s like JSON. but fast and small.

package com.zxh.demo.model;

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 + "]";
    }
}

編寫一個配置類接口,用於控制心跳包和應用消息的處理

package com.zxh.demo.model;

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

    byte PING = 1;

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

創建MsgPckDecode(解碼器)

package com.zxh.demo.model;

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(編碼器)
```

package com.zxh.demo.model;

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);

    }
}

```
創建client客戶端：
```

package com.zxh.demo.client;

import java.util.Scanner;
import java.util.concurrent.TimeUnit;

import com.zxh.demo.model.Model;
import com.zxh.demo.model.MsgPckDecode;
import com.zxh.demo.model.MsgPckEncode;
import com.zxh.demo.model.TypeData;

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 ClientHandler(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() {
                            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);
            }
        }
    }
}

```
創建Server服務端：
```

package com.zxh.demo.server;
import com.zxh.demo.model.MsgPckDecode;
import com.zxh.demo.model.MsgPckEncode;

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 ServerHandler()); 
                }
            });         
            System.out.println("start server by port 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();
        }
    }
}

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

三、項目結構圖

四、補充

所謂的心跳, 即在 TCP 長連接中, 客戶端和服務器之間定期發送的一種特殊的數據包, 通知對方自己還在線, 以確保 TCP 連接的有效性.因為網絡的不可靠性, 有可能在 TCP 保持長連接的過程中, 由於某些突發情況, 例如網線被拔出, 突然掉電等, 會造成服務器和客戶端的連接中斷. 在這些突發情況下, 如果恰好服務器和客戶端之間沒有交互的話, 那么它們是不能在短時間內發現對方已經掉線的. 為了解決這個問題, 我們就需要引入心跳機制. 心跳機制的工作原理是: 在服務器和客戶端之間一定時間內沒有數據交互時, 即處於 idle 狀態時, 客戶端或服務器會發送一個特殊的數據包給對方, 當接收方收到這個數據報文后, 也立即發送一個特殊的數據報文, 回應發送方, 此即一個 PING-PONG 交互. 自然地, 當某一端收到心跳消息后, 就知道了對方仍然在線, 這就確保 TCP 連接的有效性

代碼下載：http://www.demodashi.com/demo/15012.html

注：本文著作權歸作者，由demo大師發表，拒絕轉載，轉載需要作者授權

免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。

猜您在找 Netty心跳機制-長連接 Netty(一) SpringBoot 整合長連接心跳機制 Netty實現心跳機制 [轉]Android TCP長連接心跳機制及實現聊聊心跳機制及netty心跳實現 Netty（六）：Netty中的連接管理（心跳機制和定時斷線重連） Java實現心跳機制長連接、心跳和斷線重連 Netty生產級的心跳和重連機制 Eureka 心跳機制