netty的心跳檢測與斷線重連

服務端代碼

NettyServer

package com.example.netty.netty2;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class NettyServer {

    private NettyServerChannelInitializer serverChannelInitializer = null;

    private int port = 8888;


    public void bind() throws Exception {
        //配置服務端的NIO線程組
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            serverChannelInitializer = new NettyServerChannelInitializer();
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    //保持長連接配接
                    .childOption(ChannelOption.SO_KEEPALIVE,true)
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    .childHandler(serverChannelInitializer);

            //綁定端口，同步等待成功
            ChannelFuture f = b.bind(port).sync();


            //等待伺服器監聽端口關閉
            f.channel().closeFuture().sync();
        } finally {
            //釋放線程池資源
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws Exception {
        new  NettyServer().bind();
    }

}

自定義channel類裡面涉及有拆包，粘包，編解碼以及其他自定義操作(心跳檢測)

package com.example.netty.netty2;

import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.timeout.IdleStateHandler;
import java.util.concurrent.TimeUnit;


public class NettyServerChannelInitializer extends ChannelInitializer<SocketChannel> {

    private NettyServerHandler handler ;



    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        ChannelPipeline pipeline = socketChannel.pipeline();
        //解決TCP粘包拆包的問題，以特定的字元結尾（$_）
        pipeline.addLast(new DelimiterBasedFrameDecoder(Integer.MAX_VALUE, Unpooled.copiedBuffer("$_".getBytes())));
        //字元串解碼和編碼
        pipeline.addLast("decoder", new StringDecoder());
        pipeline.addLast("encoder", new StringEncoder());
        pipeline.addLast(new IdleStateHandler(3,0,0,TimeUnit.SECONDS));
        //伺服器的邏輯
        handler = new NettyServerHandler();
        pipeline.addLast("handler", handler);
    }
}

NettyServerHandler 心跳檢測

package com.example.netty.netty2;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import lombok.extern.slf4j.Slf4j;


@Slf4j
public class NettyServerHandler extends SimpleChannelInboundHandler {

    /**
     * 心跳丢失次數
     */
    private int counter = 0;


    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("Client say : " + msg.toString());
        //重置心跳丢失次數
        counter = 0;
    }


    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.info("服務端被 : " + ctx.channel().remoteAddress().toString()+ " 連接配接上了 !");
        super.channelActive(ctx);
    }


    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            if (event.state().equals(IdleState.READER_IDLE)){
                // 空閑40s之後觸發 (心跳包丢失)
                if (counter >= 3) {
                    // 連續丢失3個心跳包 (斷開連接配接)
                    ctx.channel().close().sync();
                    log.error("已與"+ctx.channel().remoteAddress()+"斷開連接配接");
                    System.out.println("已與"+ctx.channel().remoteAddress()+"斷開連接配接");
                } else {
                    counter++;
                    log.debug(ctx.channel().remoteAddress() + "丢失了第 "+ctx.channel().remoteAddress() + counter + " 個心跳包");
                    System.out.println("丢失了"+ctx.channel().remoteAddress()+"第 " + counter + " 個心跳包");
                }
            }

        }
    }
}

用戶端

NettyClient

package com.example.netty.netty2;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoop;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import java.util.concurrent.TimeUnit;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;

@Slf4j
public class NettyClient {

    private String host;

    private int port;

    private static Channel channel;

    public NettyClient(){

    }

    public NettyClient(String host, int port) {
        this.host = host;
        this.port = port;
    }



    public void start()  {
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
                    .option(ChannelOption.SO_KEEPALIVE,true)
                    .channel(NioSocketChannel.class)
                    .handler(new ClientChannelInitializer(host,port));

            ChannelFuture f = b.connect(host,port);

//            //斷線重連
            f.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture channelFuture) throws Exception {
                    if (!channelFuture.isSuccess()) {
                        final EventLoop loop = channelFuture.channel().eventLoop();
                        loop.schedule(new Runnable() {
                            @Override
                            public void run() {
                                log.error("服務端連結不上，開始重連操作...");
                                System.err.println("服務端連結不上，開始重連操作...");
                                start();
                            }
                        }, 1L, TimeUnit.SECONDS);
                    } else {
                        channel = channelFuture.channel();
                        log.info("服務端連結成功...");
                        System.err.println("服務端連結成功...");
                    }
                }
            });

        }catch (Exception e){
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        new NettyClient ("127.0.0.1",8888).start();
    }
}

ClientChannelInitializer

package com.example.netty.netty2;

import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.timeout.IdleStateHandler;
import java.util.concurrent.TimeUnit;

public class ClientChannelInitializer extends ChannelInitializer<SocketChannel> {

    private String host;
    private int port;

    public ClientChannelInitializer( String host, int port) {
        this.host = host;
        this.port = port;
    }

    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        ChannelPipeline pipeline = socketChannel.pipeline();
        //解決TCP粘包拆包的問題，以特定的字元結尾（$_）
        pipeline.addLast(new DelimiterBasedFrameDecoder(Integer.MAX_VALUE, Unpooled.copiedBuffer("$_".getBytes())));
        //字元串解碼和編碼
        pipeline.addLast("decoder", new StringDecoder());
        pipeline.addLast("encoder", new StringEncoder());
        //心跳檢測
        pipeline.addLast(new IdleStateHandler(0,10,0,TimeUnit.SECONDS));
        //用戶端的邏輯
        pipeline.addLast("handler", new NettyClientHandler(host,port));

    }
}

NettyClientHandler

package com.example.netty.netty2;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.EventLoop;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import java.util.concurrent.TimeUnit;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class NettyClientHandler extends SimpleChannelInboundHandler {

    private String host;
    private int port;
    private NettyClient nettyClinet;
    private String tenantId;


    public NettyClientHandler(String host, int port) {
        this.host = host;
        this.port = port;
        nettyClinet = new NettyClient(host,port);
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, Object o)
            throws Exception {
        System.out.println("Server say : " + o.toString());
    }


    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("通道已連接配接！！");
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("斷線了。。。。。。");
        //使用過程中斷線重連
        final EventLoop eventLoop = ctx.channel().eventLoop();
        eventLoop.schedule(new Runnable() {
            @Override
            public void run() {
                nettyClinet.start();
            }
        }, 1, TimeUnit.SECONDS);

        ctx.fireChannelInactive();
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt)
            throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            if (event.state().equals(IdleState.READER_IDLE)) {
                System.out.println("READER_IDLE");

            } else if (event.state().equals(IdleState.WRITER_IDLE)) {
                /**發送心跳,保持長連接配接*/
                String  s = "ping$_";
                ctx.channel().writeAndFlush(s);
                log.debug("心跳發送成功!");
                System.out.println("心跳發送成功!");
            } else if (event.state().equals(IdleState.ALL_IDLE)) {
                System.out.println("ALL_IDLE");
            }
        }
        super.userEventTriggered(ctx, evt);
    }

}

心跳檢測類裡面關鍵代碼IdleStateHandler

netty的心跳檢測與斷線重連

readerIdleTime：為讀逾時時間（即測試端一定時間内未接受到被測試端消息）
writerIdleTime：為寫逾時時間（即測試端一定時間内向被測試端發送消息）
allIdleTime：所有類型的逾時時間
比如在這裡，服務端每隔3S檢測一下，若超過3次，就自動斷開連接配接，
用戶端是每個10S中發送一個心跳包到服務端。
so,如非異常情況，會一直保持通信，因為服務端在将要端口連接配接的時候，這個時候用戶端發送了一個心跳包。

netty的心跳檢測與斷線重連

繼續閱讀

關于Gradle配置的小結

Java小案例——随機數猜測随機數猜測

nginx location中斜線的位置的重要性

27 Best Free Eclipse Plug-ins for Java Developer to be ProductiveCode Quality PluginsText Editor PluginsDependency ManagementVersion Control Integration PluginsFramework Development Continuous Integration Related PluginsOther Utility Plugins

Java String.format方法的簡單使用

neo4j之cypher使用文檔

GitHub連夜封殺！這份阿裡 10W 字内部 Java 字面試手冊到底有多強？

spark/scala關于【資源檔案】加載方法概述外部檔案加載方案測試資源檔案打包入jar包中小結

mybatis_入門程式Mybatis入門

AOP程式設計_Android優雅權限架構(1)概念基礎，2021金三銀四前言正文大綱正文

Effective Java 8:通用程式設計

OOM三種類型

工廠模式-三種類型

【遞歸】高效率求2的n次幂

win10本地scala和spark安裝安裝scala安裝spark

scala (3) Function 和 Method