package com.maibu.netty; import com.fastbee.common.MiddleConstant; import com.fastbee.netty.handler.*; import io.netty.bootstrap.Bootstrap; 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 org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.concurrent.TimeUnit; /** * Netty客户端实现类 * 提供与TCP服务器的连接管理、消息发送、重连机制和心跳检测功能 */ public class NettyClient { private static final Logger logger = LoggerFactory.getLogger(NettyClient.class); // private static volatile NettyClient instance; public String token; public String userName; public String requestDeviceId; //前端需要连接的设备 public String currentDeviceId; //当前连接的设备 public Long heartBeatTimer; public String key; //userName + web + token public String host; public int port; public int reconnectDelay = 5; public int heartbeatInterval = 5; public EventLoopGroup group; public Bootstrap bootstrap; public Channel channel; public Long lastSwitchTime = -1L; public Long Interval = 5000L; public static final byte[] HEARTBEAT_PACKET = new byte[]{(byte) 0xAB, (byte) 0xAA, (byte) 0xFF, (byte) 0xAA, (byte) 0xAB}; // /** // * 获取单例实例(无回调) // */ // public static NettyClient getInstance() { // if (instance == null) { // synchronized (NettyClient.class) { // if (instance == null) { // instance = new NettyClient(); // } // } // } // return instance; // } /** * 初始化并连接到服务器 */ public void initConnect(String host, int port) { if (channel != null && channel.isActive()) { logger.info("Netty客户端已连接,不重复初始化"); return; } this.host = host; this.port = port; connect(); } /** * 初始化Netty客户端 */ public void initClient() { group = new NioEventLoopGroup(); bootstrap = new Bootstrap(); bootstrap.group(group) .channel(NioSocketChannel.class) .option(ChannelOption.SO_KEEPALIVE, true) .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000) .handler(new ChannelInitializer() { @Override protected void initChannel(SocketChannel ch) { ChannelPipeline pipeline = ch.pipeline(); // 设置共享属性 ch.attr(MiddleConstant.ATT_MASTER_KEY).set(key); // 空闲检测处理器(心跳) pipeline.addLast(new IdleStateHandler(0, heartbeatInterval, 0, TimeUnit.SECONDS)); pipeline.addLast(new MiddleHeaderFooterDecoder()); // 编解码器 pipeline.addLast(new MiddleHexDecoder()); pipeline.addLast(new MiddleHexEncoder()); // 自定义处理器 pipeline.addLast(new ClientHandler()); } }); } /** * 连接到服务器 */ public void connect() { if (bootstrap == null || host == null || port <= 0) { logger.error("Netty客户端未正确初始化"); return; } logger.info("正在连接到Netty服务器: {}:{}", host, port); bootstrap.connect(host, port).addListener((ChannelFutureListener) future -> { if (future.isSuccess()) { channel = future.channel(); logger.info("Netty客户端连接成功: {}:{}", host, port); // notifyConnectionStatus(true); } else { logger.error("Netty客户端连接失败,{}秒后重试: {}", reconnectDelay, future.cause().getMessage()); // notifyConnectionStatus(false); // 重连 group.schedule(this::connect, reconnectDelay, TimeUnit.SECONDS); } }); } /** * 断开连接 */ public void disconnect() { if (channel != null) { channel.close(); channel = null; } // notifyConnectionStatus(false); logger.info("Netty客户端已断开连接"); } /** * 发送数据 */ public void sendData(byte[] data) { if (channel != null && channel.isActive()) { channel.writeAndFlush(data).addListener((ChannelFutureListener) future -> { if (future.isSuccess()) { logger.debug("Netty客户端发送数据成功: {}", bytesToHex(data)); } else { logger.error("Netty客户端发送数据失败: {}", future.cause().getMessage()); } }); } else { logger.error("Netty客户端未连接,发送数据失败"); } } /** * 发送十六进制字符串数据 */ public void sendHexData(String hexData) { try { byte[] data = hexStringToBytes(hexData); sendData(data); } catch (IllegalArgumentException e) { logger.error("无效的十六进制字符串: {}", hexData); } } /** * 十六进制字符串转字节数组 */ private static byte[] hexStringToBytes(String hexString) { hexString = hexString.replaceAll("\\s+", ""); if (hexString.length() % 2 != 0) { throw new IllegalArgumentException("十六进制字符串长度必须为偶数"); } byte[] result = new byte[hexString.length() / 2]; for (int i = 0; i < hexString.length(); i += 2) { result[i / 2] = (byte) Integer.parseInt(hexString.substring(i, i + 2), 16); } return result; } /** * 字节数组转十六进制字符串 */ private static String bytesToHex(byte[] bytes) { StringBuilder sb = new StringBuilder(); for (byte b : bytes) { sb.append(String.format("%02X ", b)); } return sb.toString().trim(); } /** * 检查客户端是否已连接 */ public boolean isConnected() { return channel != null && channel.isActive(); } /** * 设置重连延迟时间(秒) */ public void setReconnectDelay(int reconnectDelay) { if (reconnectDelay > 0) { this.reconnectDelay = reconnectDelay; } } /** * 设置心跳间隔时间(秒) */ public void setHeartbeatInterval(int heartbeatInterval) { if (heartbeatInterval > 0) { this.heartbeatInterval = heartbeatInterval; } } }