package com.ard.work.component.netty;
|
|
import com.ard.work.component.netty.command.register.HeartbeatCmd;
|
import com.ard.work.component.netty.handler.ClientHandler;
|
import com.ard.work.component.netty.handler.MessageProtocolDecoder;
|
import com.ard.work.component.netty.handler.MessageProtocolEncoder;
|
import com.ard.work.component.netty.cache.TyTcpServerCache;
|
import com.ard.work.domain.entity.ArdServerInfo;
|
import com.ard.work.service.IArdDroneInfoService;
|
import com.ard.work.service.IArdServerInfoService;
|
import io.netty.bootstrap.Bootstrap;
|
import io.netty.channel.*;
|
import io.netty.channel.nio.NioEventLoopGroup;
|
import io.netty.channel.socket.SocketChannel;
|
import io.netty.channel.socket.nio.NioSocketChannel;
|
import io.netty.handler.timeout.IdleStateHandler;
|
import io.netty.util.AttributeKey;
|
import lombok.extern.slf4j.Slf4j;
|
import org.springframework.scheduling.annotation.Scheduled;
|
import org.springframework.stereotype.Component;
|
|
import javax.annotation.PostConstruct;
|
import javax.annotation.Resource;
|
import java.util.concurrent.TimeUnit;
|
|
|
@Component
|
@Slf4j
|
public class TyTcpClientManager {
|
|
|
private final EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
|
|
@Resource
|
private IArdServerInfoService ardServerService; // ← 从数据库读取边缘服务器配置
|
|
@Resource
|
private IArdDroneInfoService ardDroneService;
|
|
@Resource
|
private ClientHandler clientHandler;
|
|
/**
|
* 初始化 Bootstrap
|
*/
|
private Bootstrap getBootstrap() {
|
Bootstrap bootstrap = new Bootstrap();
|
bootstrap.group(eventLoopGroup)
|
.channel(NioSocketChannel.class)
|
.option(ChannelOption.TCP_NODELAY, true)
|
.option(ChannelOption.SO_KEEPALIVE, true)
|
.handler(new ChannelInitializer<SocketChannel>() {
|
@Override
|
protected void initChannel(SocketChannel socketChannel) {
|
ChannelPipeline pipeline = socketChannel.pipeline();
|
pipeline.addLast("heart", new IdleStateHandler(0, 5, 0, TimeUnit.SECONDS));
|
pipeline.addLast("encoder", new MessageProtocolEncoder());
|
pipeline.addLast("decoder", new MessageProtocolDecoder());
|
pipeline.addLast("handler", clientHandler);
|
}
|
});
|
return bootstrap;
|
}
|
|
/**
|
* 新增或重连服务器
|
*/
|
public void connectServer(ArdServerInfo serverInfo) {
|
String serverId = serverInfo.getId();
|
String host = serverInfo.getHost();
|
Integer port = serverInfo.getTcpPort();
|
|
// 如果已有旧连接，先断开
|
disconnectServer(serverId);
|
|
Bootstrap bootstrap = getBootstrap();
|
bootstrap.connect(host, port).addListener((ChannelFuture f) -> {
|
if (f.isSuccess()) {
|
log.info("✅ 服务器 [{}] 连接成功", serverId);
|
Channel channel = f.channel();
|
channel.attr(AttributeKey.valueOf("serverId")).set(serverId);
|
TyTcpServerCache.putTcpServerCache(serverId, true);
|
// 修改状态为： 1 - 在线
|
updateServerAndDroneStatus(serverId, "1");
|
|
// 成功后再加入 channels
|
TyTcpServerCache.putChannel(serverId, channel);
|
|
} else {
|
log.warn("❌ 服务器 [{}] 连接失败", serverId);
|
TyTcpServerCache.putTcpServerCache(serverId, false);
|
// 断开服务器，修改状态为： 0 - 离线
|
updateServerAndDroneStatus(serverId, "0");
|
}
|
});
|
}
|
|
/**
|
* 断开服务器连接
|
*/
|
public void disconnectServer(String serverId) {
|
Channel channel = TyTcpServerCache.getChannel(serverId);
|
if (channel != null) {
|
channel.close();
|
log.info("断开服务器 [{}] 连接", serverId);
|
// 断开服务器，修改状态为： 0 - 离线
|
updateServerAndDroneStatus(serverId, "0");
|
TyTcpServerCache.removeChannel(serverId);
|
}
|
TyTcpServerCache.removeTcpServerCache(serverId);
|
}
|
|
/**
|
* 同步修改服务和无人机状态
|
*/
|
private void updateServerAndDroneStatus(String serverId, String status) {
|
ardServerService.updateServerStatus(serverId, status);
|
ardDroneService.updateDroneStatus(serverId, status);
|
}
|
|
/**
|
* 启动时从数据库加载所有边缘服务器并建立 TCP 连接
|
*/
|
@PostConstruct
|
public void initConnectServers() {
|
try {
|
var serverList = ardServerService.selectArdServerInfoList(new ArdServerInfo() {{ setUseDataScope(false); }}); // ← 从数据库查询所有服务器配置
|
if (serverList == null || serverList.isEmpty()) {
|
log.warn("未在数据库中找到边缘服务器配置，跳过连接初始化");
|
return;
|
}
|
|
for (ArdServerInfo serverInfo : serverList) {
|
connectServer(serverInfo);
|
}
|
|
log.info("共初始化 {} 个边缘服务器连接", serverList.size());
|
} catch (Exception e) {
|
log.error("创建边缘服务器连接异常：{}", e.getMessage(), e);
|
}
|
}
|
|
@Scheduled(fixedDelay = 5000) // 每 5 秒检查一次
|
public void reconnectFailedServers() {
|
TyTcpServerCache.getTcpServerCacheMap().forEach((key, state) -> {
|
if (Boolean.FALSE.equals(state)) { // 连接失败的服务器
|
ArdServerInfo serverInfo = ardServerService.selectArdServerInfoById(key);
|
if (serverInfo != null) {
|
log.warn("服务器 [{}] 连接失败，尝试重连", key);
|
connectServer(serverInfo);
|
}
|
}
|
});
|
}
|
|
/**
|
* 定时向所有已连接的边缘服务器服务器发送心跳（写空闲已经发心跳了，这里暂时取消）
|
*/
|
// @Scheduled(fixedRate = 5000) // 每5秒执行一次
|
public void sendHeartBeat() {
|
long timestamp = System.currentTimeMillis();
|
TyTcpServerCache.getChannels().stream()
|
// 过滤出仍然活跃的连接
|
.filter(channel -> channel != null && channel.isActive())
|
.forEach(channel -> {
|
String serverId = (String) channel.attr(AttributeKey.valueOf("serverId")).get();
|
HeartbeatCmd heartbeat = new HeartbeatCmd(timestamp);
|
channel.writeAndFlush(heartbeat);
|
log.info("❤\uFE0F 向边缘服务器 [{}]发送心跳，时间戳={}", serverId, timestamp);
|
});
|
}
|
}
|
