package com.ard.work.component.netty;
|
|
import com.ard.work.component.netty.command.BaseCmd;
|
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.domain.cache.TyTcpServerCache;
|
import com.ard.work.domain.entity.ArdServerInfo;
|
import com.ard.work.service.IArdServerService;
|
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.stereotype.Component;
|
import javax.annotation.PostConstruct;
|
import javax.annotation.Resource;
|
import java.net.InetSocketAddress;
|
import java.util.Map;
|
import java.util.concurrent.ConcurrentHashMap;
|
import java.util.concurrent.TimeUnit;
|
|
@Component
|
@Slf4j
|
public class TyTcpClientManager {
|
/**
|
* 以 serverId 作为唯一 key
|
*/
|
public static final Map<String, TcpClientSession> channels = new ConcurrentHashMap<>();
|
|
private final EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
|
|
@Resource
|
private IArdServerService ardServerService; // ← 从数据库读取边缘服务器配置
|
|
@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, 3, 0, TimeUnit.SECONDS));
|
pipeline.addLast("encoder", new MessageProtocolEncoder());
|
pipeline.addLast("decoder", new MessageProtocolDecoder());
|
pipeline.addLast("handler", clientHandler);
|
}
|
});
|
return bootstrap;
|
}
|
|
/**
|
* 初始化与六合服务器的连接
|
*/
|
public void initConnection(String serverId, String host, Integer port) {
|
String key = host + ":" + port;
|
log.info("初始化连接边缘服务器 [{}] ({})", serverId, key);
|
|
// 如果存在旧连接则关闭
|
if (channels.containsKey(serverId)) {
|
TcpClientSession oldSession = channels.get(serverId);
|
if (oldSession != null && oldSession.getChannelFuture() != null) {
|
oldSession.getChannelFuture().channel().close();
|
}
|
}
|
|
// 异步连接 TCP 服务端
|
Bootstrap bootstrap = getBootstrap();
|
ChannelFuture future = bootstrap.connect(host, port).addListener((ChannelFuture futureListener) -> {
|
if (futureListener.isSuccess()) {
|
log.info("✅ 边缘服务器 [{}] 连接成功 ({})", serverId, key);
|
Channel channel = futureListener.channel();
|
// 绑定 serverId 到 channel
|
channel.attr(AttributeKey.valueOf("serverId")).set(serverId);
|
TyTcpServerCache.putTcpServerCache(key, true);
|
} else {
|
log.warn("❌ 边缘服务器 [{}] 连接失败 ({})，10 秒后重试", serverId, key);
|
TyTcpServerCache.putTcpServerCache(key, false);
|
final EventLoop eventLoop = futureListener.channel().eventLoop();
|
eventLoop.schedule(() -> initConnection(serverId, host, port), 10, TimeUnit.SECONDS);
|
}
|
});
|
|
// 存入连接缓存
|
TcpClientSession session = TcpClientSession.builder()
|
.ClientId("TY_" + serverId)
|
.channelFuture(future)
|
.isReConnect(true)
|
.serverId(serverId)
|
.host(host)
|
.port(port)
|
.build();
|
|
channels.put(serverId, session);
|
}
|
|
/**
|
* 向指定边缘服务器发送消息
|
*/
|
public void sendMessage(Long serverId, BaseCmd msg) {
|
try {
|
TcpClientSession session = channels.get(serverId);
|
if (session == null) {
|
log.error("未找到六合服务器 [{}] 的连接", serverId);
|
return;
|
}
|
|
ChannelFuture future = session.getChannelFuture();
|
if (future == null || !future.channel().isActive()) {
|
log.warn("六合服务器 [{}] 连接未激活，无法发送消息", serverId);
|
return;
|
}
|
|
InetSocketAddress remote = (InetSocketAddress) future.channel().remoteAddress();
|
future.channel().writeAndFlush(msg);
|
log.debug("→ 向六合服务器 [{}] ({}:{}) 发送消息，命令码：0x{}",
|
serverId, remote.getHostString(), remote.getPort(),
|
String.format("%02X", msg.getCmdCode() & 0xFF));
|
} catch (Exception e) {
|
log.error("发送消息到六合服务器 [{}] 异常：{}", serverId, e.getMessage());
|
}
|
}
|
|
/**
|
* 启动时从数据库加载所有边缘服务器并建立 TCP 连接
|
*/
|
@PostConstruct
|
public void createAllTyConnections() {
|
try {
|
var serverList = ardServerService.selectArdServerManagementList(new ArdServerInfo()); // ← 从数据库查询所有服务器配置
|
if (serverList == null || serverList.isEmpty()) {
|
log.warn("未在数据库中找到边缘服务器配置，跳过连接初始化");
|
return;
|
}
|
|
for (ArdServerInfo serverInfo : serverList) {
|
initConnection(serverInfo.getId(), serverInfo.getHost(), serverInfo.getTcpPort());
|
}
|
|
log.info("共初始化 {} 个边缘服务器连接", serverList.size());
|
} catch (Exception e) {
|
log.error("创建边缘服务器连接异常：{}", e.getMessage(), e);
|
}
|
}
|
}
|
