BJ_ZTEITS-Developers / parking-cloud-yudao

Commit 09041a24d765ce8649d9607e56fbccdfb5491429

Authored by YunaiV
1 parent 5ead81be

feat（iot）：【协议改造】tcp 初步改造（100%）：基于 code review 进一步完善，对应 iot-tcp-fix-plan.md

Inline Side-by-side
Showing 5 changed files with 87 additions and 24 deletions
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/IotTcpProtocol.java
  View file @09041a2
... ... @@ -20,6 +20,7 @@ import io.vertx.core.net.NetServerOptions;
20 20 import io.vertx.core.net.PemKeyCertOptions;
21 21 import lombok.Getter;
22 22 import lombok.extern.slf4j.Slf4j;
  23 +import org.springframework.util.Assert;
23 24  
24 25 /**
25 26 * IoT TCP 协议实现
... ... @@ -89,11 +90,12 @@ public class IotTcpProtocol implements IotProtocol {
89 90 this.serializer = serializerManager.get(serializeType);
90 91 // 初始化帧编解码器
91 92 IotTcpConfig tcpConfig = properties.getTcp();
92   - IotTcpConfig.CodecConfig codecConfig = tcpConfig != null ? tcpConfig.getCodec() : null;
93   - this.frameCodec = IotTcpFrameCodecFactory.create(codecConfig);
  93 + Assert.notNull(tcpConfig, "TCP 协议配置（tcp）不能为空");
  94 + Assert.notNull(tcpConfig.getCodec(), "TCP 拆包配置（tcp.codec）不能为空");
  95 + this.frameCodec = IotTcpFrameCodecFactory.create(tcpConfig.getCodec());
94 96  
95 97 // 初始化连接管理器
96   - this.connectionManager = new IotTcpConnectionManager();
  98 + this.connectionManager = new IotTcpConnectionManager(tcpConfig.getMaxConnections());
97 99  
98 100 // 初始化下行消息订阅者
99 101 IotTcpDownstreamHandler downstreamHandler = new IotTcpDownstreamHandler(connectionManager, frameCodec, serializer);
... ... @@ -117,7 +119,7 @@ public class IotTcpProtocol implements IotProtocol {
117 119 return;
118 120 }
119 121  
120   - // 1.1 创建 Vertx 实例（每个 Protocol 独立管理）
  122 + // 1.1 创建 Vertx 实例
121 123 this.vertx = Vertx.vertx();
122 124  
123 125 // 1.2 创建服务器选项
... ... @@ -126,8 +128,9 @@ public class IotTcpProtocol implements IotProtocol {
126 128 .setPort(properties.getPort())
127 129 .setTcpKeepAlive(true)
128 130 .setTcpNoDelay(true)
129   - .setReuseAddress(true);
130   - if (tcpConfig != null && Boolean.TRUE.equals(tcpConfig.getSslEnabled())) {
  131 + .setReuseAddress(true)
  132 + .setIdleTimeout((int) (tcpConfig.getKeepAliveTimeoutMs() / 1000)); // 设置空闲超时
  133 + if (Boolean.TRUE.equals(tcpConfig.getSslEnabled())) {
131 134 PemKeyCertOptions pemKeyCertOptions = new PemKeyCertOptions()
132 135 .setKeyPath(tcpConfig.getSslKeyPath())
133 136 .setCertPath(tcpConfig.getSslCertPath());
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/codec/delimiter/IotTcpDelimiterFrameCodec.java
  View file @09041a2
1 1 package cn.iocoder.yudao.module.iot.gateway.protocol.tcp.codec.delimiter;
2 2  
  3 +import cn.hutool.core.util.StrUtil;
3 4 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.IotTcpConfig;
4 5 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.codec.IotTcpCodecTypeEnum;
5 6 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.codec.IotTcpFrameCodec;
... ... @@ -28,6 +29,11 @@ import org.springframework.util.Assert;
28 29 public class IotTcpDelimiterFrameCodec implements IotTcpFrameCodec {
29 30  
30 31 /**
  32 + * 最大记录大小（64KB），防止 DoS 攻击
  33 + */
  34 + private static final int MAX_RECORD_SIZE = 65536;
  35 +
  36 + /**
31 37 * 解析后的分隔符字节数组
32 38 */
33 39 private final byte[] delimiterBytes;
... ... @@ -45,6 +51,7 @@ public class IotTcpDelimiterFrameCodec implements IotTcpFrameCodec {
45 51 @Override
46 52 public RecordParser createDecodeParser(Handler<Buffer> handler) {
47 53 RecordParser parser = RecordParser.newDelimited(Buffer.buffer(delimiterBytes));
  54 + parser.maxRecordSize(MAX_RECORD_SIZE); // 设置最大记录大小，防止 DoS 攻击
48 55 // 处理完整消息（不包含分隔符）
49 56 parser.handler(handler);
50 57 parser.exceptionHandler(ex -> {
... ... @@ -76,7 +83,7 @@ public class IotTcpDelimiterFrameCodec implements IotTcpFrameCodec {
76 83 .replace("\\r", "\r")
77 84 .replace("\\n", "\n")
78 85 .replace("\\t", "\t");
79   - return parsed.getBytes();
  86 + return StrUtil.utf8Bytes(parsed);
80 87 }
81 88  
82 89 }
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/codec/length/IotTcpFixedLengthFrameCodec.java
  View file @09041a2
... ... @@ -46,6 +46,11 @@ public class IotTcpFixedLengthFrameCodec implements IotTcpFrameCodec {
46 46  
47 47 @Override
48 48 public Buffer encode(byte[] data) {
  49 + // 校验数据长度不能超过固定长度
  50 + if (data.length > fixedLength) {
  51 + throw new IllegalArgumentException(String.format(
  52 + "数据长度 %d 超过固定长度 %d", data.length, fixedLength));
  53 + }
49 54 Buffer buffer = Buffer.buffer(fixedLength);
50 55 buffer.appendBytes(data);
51 56 // 如果数据不足固定长度，填充 0（RecordParser.newFixed 解码时按固定长度读取，所以发送端需要填充）
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/handler/upstream/IotTcpUpstreamHandler.java
  View file @09041a2
... ... @@ -25,7 +25,7 @@ import io.vertx.core.parsetools.RecordParser;
25 25 import lombok.extern.slf4j.Slf4j;
26 26 import org.springframework.util.Assert;
27 27  
28   -import static cn.iocoder.yudao.framework.common.exception.enums.GlobalErrorCodeConstants.SUCCESS;
  28 +import static cn.iocoder.yudao.framework.common.exception.enums.GlobalErrorCodeConstants.*;
29 29 import static cn.iocoder.yudao.framework.common.exception.enums.GlobalErrorCodeConstants.UNAUTHORIZED;
30 30  
31 31 /**
... ... @@ -95,8 +95,8 @@ public class IotTcpUpstreamHandler implements Handler&lt;NetSocket&gt; {
95 95 try {
96 96 processMessage(clientId, buffer, socket);
97 97 } catch (Exception e) {
98   - log.error("[handle][消息处理失败，客户端 ID: {}，地址: {}，错误: {}]",
99   - clientId, socket.remoteAddress(), e.getMessage());
  98 + log.error("[handle][消息处理失败，客户端 ID: {}，地址: {}]",
  99 + clientId, socket.remoteAddress(), e);
100 100 socket.close();
101 101 }
102 102 };
... ... @@ -114,20 +114,40 @@ public class IotTcpUpstreamHandler implements Handler&lt;NetSocket&gt; {
114 114 * @param socket 网络连接
115 115 */
116 116 private void processMessage(String clientId, Buffer buffer, NetSocket socket) {
117   - // 1. 反序列化消息
118   - IotDeviceMessage message = serializer.deserialize(buffer.getBytes());
119   - Assert.notNull(message, "反序列化后消息为空");
  117 + IotDeviceMessage message = null;
  118 + try {
  119 + // 1. 反序列化消息
  120 + message = serializer.deserialize(buffer.getBytes());
  121 + if (message == null) {
  122 + sendErrorResponse(socket, null, null, BAD_REQUEST.getCode(), "消息反序列化失败");
  123 + return;
  124 + }
120 125  
121   - // 2. 根据消息类型路由处理
122   - if (AUTH_METHOD.equals(message.getMethod())) {
123   - // 认证请求
124   - handleAuthenticationRequest(clientId, message, socket);
125   - } else if (IotDeviceMessageMethodEnum.DEVICE_REGISTER.getMethod().equals(message.getMethod())) {
126   - // 设备动态注册请求
127   - handleRegisterRequest(clientId, message, socket);
128   - } else {
129   - // 业务消息
130   - handleBusinessRequest(clientId, message, socket);
  126 + // 2. 根据消息类型路由处理
  127 + if (AUTH_METHOD.equals(message.getMethod())) {
  128 + // 认证请求
  129 + handleAuthenticationRequest(clientId, message, socket);
  130 + } else if (IotDeviceMessageMethodEnum.DEVICE_REGISTER.getMethod().equals(message.getMethod())) {
  131 + // 设备动态注册请求
  132 + handleRegisterRequest(clientId, message, socket);
  133 + } else {
  134 + // 业务消息
  135 + handleBusinessRequest(clientId, message, socket);
  136 + }
  137 + } catch (IllegalArgumentException e) {
  138 + // 参数校验失败，返回 400
  139 + log.warn("[processMessage][参数校验失败，客户端 ID: {}，错误: {}]", clientId, e.getMessage());
  140 + String requestId = message != null ? message.getRequestId() : null;
  141 + String method = message != null ? message.getMethod() : null;
  142 + sendErrorResponse(socket, requestId, method, BAD_REQUEST.getCode(), e.getMessage());
  143 + } catch (Exception e) {
  144 + // 其他异常，返回 500 并重新抛出让上层关闭连接
  145 + log.error("[processMessage][处理消息失败，客户端 ID: {}]", clientId, e);
  146 + String requestId = message != null ? message.getRequestId() : null;
  147 + String method = message != null ? message.getMethod() : null;
  148 + sendErrorResponse(socket, requestId, method,
  149 + INTERNAL_SERVER_ERROR.getCode(), INTERNAL_SERVER_ERROR.getMsg());
  150 + throw e;
131 151 }
132 152 }
133 153  
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/manager/IotTcpConnectionManager.java
  View file @09041a2
... ... @@ -22,6 +22,11 @@ import java.util.concurrent.ConcurrentHashMap;
22 22 public class IotTcpConnectionManager {
23 23  
24 24 /**
  25 + * 最大连接数
  26 + */
  27 + private final int maxConnections;
  28 +
  29 + /**
25 30 * 连接信息映射：NetSocket -> 连接信息
26 31 */
27 32 private final Map<NetSocket, ConnectionInfo> connectionMap = new ConcurrentHashMap<>();
... ... @@ -31,6 +36,24 @@ public class IotTcpConnectionManager {
31 36 */
32 37 private final Map<Long, NetSocket> deviceSocketMap = new ConcurrentHashMap<>();
33 38  
  39 + public IotTcpConnectionManager(int maxConnections) {
  40 + this.maxConnections = maxConnections;
  41 + }
  42 +
  43 + /**
  44 + * 获取当前连接数
  45 + */
  46 + public int getConnectionCount() {
  47 + return connectionMap.size();
  48 + }
  49 +
  50 + /**
  51 + * 检查是否可以接受新连接
  52 + */
  53 + public boolean canAcceptConnection() {
  54 + return connectionMap.size() < maxConnections;
  55 + }
  56 +
34 57 /**
35 58 * 注册设备连接（包含认证信息）
36 59 *
... ... @@ -39,6 +62,10 @@ public class IotTcpConnectionManager {
39 62 * @param connectionInfo 连接信息
40 63 */
41 64 public void registerConnection(NetSocket socket, Long deviceId, ConnectionInfo connectionInfo) {
  65 + // 检查连接数是否已达上限
  66 + if (connectionMap.size() >= maxConnections) {
  67 + throw new IllegalStateException("连接数已达上限: " + maxConnections);
  68 + }
42 69 // 如果设备已有其他连接，先清理旧连接
43 70 NetSocket oldSocket = deviceSocketMap.get(deviceId);
44 71 if (oldSocket != null && oldSocket != socket) {
... ... @@ -67,7 +94,8 @@ public class IotTcpConnectionManager {
67 94 return;
68 95 }
69 96 Long deviceId = connectionInfo.getDeviceId();
70   - deviceSocketMap.remove(deviceId);
  97 + // 仅当 deviceSocketMap 中的 socket 是当前 socket 时才移除，避免误删新连接
  98 + deviceSocketMap.remove(deviceId, socket);
71 99 log.info("[unregisterConnection][注销设备连接，设备 ID: {}，连接: {}]", deviceId, socket.remoteAddress());
72 100 }
73 101  
... ...