springboot整合netty实现心跳检测和自动重连_Java

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 服务端

创建一个 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. 实现心跳检测处理器

创建一个 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("读写空闲");
                // 发送心跳包
                ctx.writeandflush("ping\n");
            }
        } else {
            super.usereventtriggered(ctx, evt);
        }
    }

    @override
    public void exceptioncaught(channelhandlercontext ctx, throwable cause) {
        cause.printstacktrace();
        ctx.close();
    }
}

4. 配置 netty 客户端

创建一个 netty 客户端启动类，实现自动重连和心跳检测。

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("重连失败次数达到最大，放弃连接");
                    }
                }
            }
        });
    }
}

5. 实现客户端心跳处理器

创建一个 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("发送心跳包");
                ctx.writeandflush("ping\n");
            }
        } else {
            super.usereventtriggered(ctx, evt);
        }
    }

    @override
    public void channelinactive(channelhandlercontext ctx) throws exception {
        system.out.println("连接断开，尝试重连");
        // 在这里实现重连逻辑
        // 比如： 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();
    }
}