springboot整合netty實現(xiàn)心跳檢測和自動重連

更新時間：2024年11月21日 09:20:50 作者：如夜了我衣衫太薄便歸家靠路燈°

本文主要介紹了Spring Boot中整合Netty實現(xiàn)心跳檢測和自動重連,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧

1. 引入依賴

在 pom.xml 中添加 Netty 和 Spring Boot 相關依賴。

<dependencies>
    <!-- Spring Boot Starter -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</artifactId>
    </dependency>
    
    <!-- Netty Dependency -->
    <dependency>
        <groupId>io.netty</groupId>
        <artifactId>netty-all</artifactId>
        <version>4.1.63.Final</version>
    </dependency>
    
    <!-- 其他相關依賴 -->
</dependencies>

2. 配置 Netty 服務端

創(chuàng)建一個 Netty 服務器啟動類，配置心跳檢測機制。

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

@Component
public class NettyServer implements CommandLineRunner {

    private final int port = 8080;

    @Override
    public void run(String... args) throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .childHandler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 public void initChannel(SocketChannel ch) throws Exception {
                     ch.pipeline().addLast(new IdleStateHandler(5, 7, 10, TimeUnit.SECONDS));
                     ch.pipeline().addLast(new HeartbeatHandler());
                 }
             })
             .option(ChannelOption.SO_BACKLOG, 128)
             .childOption(ChannelOption.SO_KEEPALIVE, true);

            ChannelFuture f = b.bind(port).sync();
            f.channel().closeFuture().sync();
        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }
}

3. 實現(xiàn)心跳檢測處理器

創(chuàng)建一個 HeartbeatHandler 類處理心跳檢測。

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

public class HeartbeatHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            if (event.state() == IdleState.READER_IDLE) {
                System.out.println("讀空閑");
                // 關閉連接
                ctx.close();
            } else if (event.state() == IdleState.WRITER_IDLE) {
                System.out.println("寫空閑");
            } else if (event.state() == IdleState.ALL_IDLE) {
                System.out.println("讀寫空閑");
                // 發(fā)送心跳包
                ctx.writeAndFlush("ping\n");
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}

4. 配置 Netty 客戶端

創(chuàng)建一個 Netty 客戶端啟動類，實現(xiàn)自動重連和心跳檢測。

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

@Component
public class NettyClient {

    private final String host = "localhost";
    private final int port = 8080;
    private final int MAX_RETRY = 5;
    private int retry = 0;

    public void start() {
        EventLoopGroup group = new NioEventLoopGroup();

        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
             .channel(NioSocketChannel.class)
             .option(ChannelOption.SO_KEEPALIVE, true)
             .handler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 public void initChannel(SocketChannel ch) throws Exception {
                     ch.pipeline().addLast(new IdleStateHandler(0, 4, 0, TimeUnit.SECONDS));
                     ch.pipeline().addLast(new ClientHeartbeatHandler());
                 }
             });

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

    private void connect(Bootstrap b) {
        b.connect(host, port).addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (future.isSuccess()) {
                    System.out.println("連接服務器成功");
                } else {
                    System.out.println("連接服務器失敗，嘗試重連");
                    retry++;
                    if (retry < MAX_RETRY) {
                        future.channel().eventLoop().schedule(() -> connect(b), 2 << retry, TimeUnit.SECONDS);
                    } else {
                        System.out.println("重連失敗次數(shù)達到最大，放棄連接");
                    }
                }
            }
        });
    }
}

5. 實現(xiàn)客戶端心跳處理器

創(chuàng)建一個 ClientHeartbeatHandler 類處理心跳包。

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

public class ClientHeartbeatHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            if (event.state() == IdleState.WRITER_IDLE) {
                System.out.println("發(fā)送心跳包");
                ctx.writeAndFlush("ping\n");
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("連接斷開，嘗試重連");
        // 在這里實現(xiàn)重連邏輯
        // 比如： ctx.channel().eventLoop().schedule(() -> connect(), 5, TimeUnit.SECONDS);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}

6. 啟動 Spring Boot 應用

在 Spring Boot 的主類中啟動 Netty 服務器和客戶端。

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

import javax.annotation.PostConstruct;

@SpringBootApplication
public class NettySpringBootApplication {

    @Autowired
    private NettyServer nettyServer;

    @Autowired
    private NettyClient nettyClient;

    public static void main(String[] args) {
        SpringApplication.run(NettySpringBootApplication.class, args);
    }

    @PostConstruct
    public void startNetty() {
        new Thread(() -> {
            try {
                nettyServer.run();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }).start();

        new Thread(() -> nettyClient.start()).start();
    }
}