sion/sionrui
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat：【IoT 物联网】新增 TCP 二进制和 JSON 编解码器，重构 TCP 协议处理逻辑

Browse Source
This commit is contained in:
haohao
2025-07-22 00:11:46 +08:00
parent 6a117c9d55
commit bd8052f56b
19 changed files with 1868 additions and 2575 deletions
Download Patch File Download Diff File Expand all files Collapse all files

378
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/codec/tcp/IotTcpBinaryDeviceMessageCodec.java Normal file
View File

@@ -0,0 +1,378 @@
package cn.iocoder.yudao.module.iot.gateway.codec.tcp;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
import cn.iocoder.yudao.module.iot.gateway.codec.IotDeviceMessageCodec;
import io.vertx.core.buffer.Buffer;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
/**
* TCP 二进制格式 {@link IotDeviceMessage} 编解码器
*
* 使用自定义二进制协议格式：
* 包头(4字节) | 地址长度(2字节) | 设备地址(变长) | 功能码(2字节) | 消息序号(2字节) | 包体数据(变长)
*
* @author 芋道源码
*/
@Component
@Slf4j
public class IotTcpBinaryDeviceMessageCodec implements IotDeviceMessageCodec {
/**
* 编解码器类型
*/
public static final String TYPE = "TCP_BINARY";
// ==================== 常量定义 ====================
@Override
public byte[] encode(IotDeviceMessage message) {
if (message == null || StrUtil.isEmpty(message.getMethod())) {
throw new IllegalArgumentException("消息或方法不能为空");
}
try {
// 1. 确定功能码（只支持数据上报和心跳）
short code = MessageMethod.STATE_ONLINE.equals(message.getMethod()) ?
TcpDataPackage.CODE_HEARTBEAT : TcpDataPackage.CODE_MESSAGE_UP;
// 2. 构建简化负载
String payload = buildSimplePayload(message);
// 3. 构建 TCP 数据包
String deviceAddr = message.getDeviceId() != null ? String.valueOf(message.getDeviceId()) : "default";
short mid = (short) (System.currentTimeMillis() % Short.MAX_VALUE);
TcpDataPackage dataPackage = new TcpDataPackage(deviceAddr, code, mid, payload);
// 4. 编码为字节流
return encodeTcpDataPackage(dataPackage).getBytes();
} catch (Exception e) {
log.error("[encode][编码失败] 方法: {}", message.getMethod(), e);
throw new TcpCodecException("TCP 消息编码失败", e);
}
}
@Override
public IotDeviceMessage decode(byte[] bytes) {
if (bytes == null || bytes.length == 0) {
throw new IllegalArgumentException("待解码数据不能为空");
}
try {
// 1. 解码 TCP 数据包
TcpDataPackage dataPackage = decodeTcpDataPackage(Buffer.buffer(bytes));
// 2. 根据功能码确定方法
String method = (dataPackage.getCode() == TcpDataPackage.CODE_HEARTBEAT) ?
MessageMethod.STATE_ONLINE : MessageMethod.PROPERTY_POST;
// 3. 解析负载数据和请求ID
PayloadInfo payloadInfo = parsePayloadInfo(dataPackage.getPayload());
// 4. 构建 IoT 设备消息（设置完整的必要参数）
IotDeviceMessage message = IotDeviceMessage.requestOf(
payloadInfo.getRequestId(), method, payloadInfo.getParams());
// 5. 设置设备相关信息
Long deviceId = parseDeviceId(dataPackage.getAddr());
message.setDeviceId(deviceId);
// 6. 设置TCP协议相关信息
message.setServerId(generateServerId(dataPackage));
// 7. 设置租户ID（TODO: 后续可以从设备信息中获取）
// message.setTenantId(getTenantIdByDeviceId(deviceId));
if (log.isDebugEnabled()) {
log.debug("[decode][解码成功] 设备ID: {}, 方法: {}, 请求ID: {}, 消息ID: {}",
deviceId, method, message.getRequestId(), message.getId());
}
return message;
} catch (Exception e) {
log.error("[decode][解码失败] 数据长度: {}", bytes.length, e);
throw new TcpCodecException("TCP 消息解码失败", e);
}
}
@Override
public String type() {
return TYPE;
}
/**
* 构建完整负载
*/
private String buildSimplePayload(IotDeviceMessage message) {
JSONObject payload = new JSONObject();
// 核心字段
payload.set(PayloadField.METHOD, message.getMethod());
if (message.getParams() != null) {
payload.set(PayloadField.PARAMS, message.getParams());
}
// 标识字段
if (StrUtil.isNotEmpty(message.getRequestId())) {
payload.set(PayloadField.REQUEST_ID, message.getRequestId());
}
if (StrUtil.isNotEmpty(message.getId())) {
payload.set(PayloadField.MESSAGE_ID, message.getId());
}
// 时间戳
payload.set(PayloadField.TIMESTAMP, System.currentTimeMillis());
return payload.toString();
}
// ==================== 编解码方法 ====================
/**
* 解析负载信息（包含requestId和params）
*/
private PayloadInfo parsePayloadInfo(String payload) {
if (StrUtil.isEmpty(payload)) {
return new PayloadInfo(null, null);
}
try {
JSONObject jsonObject = JSONUtil.parseObj(payload);
String requestId = jsonObject.getStr(PayloadField.REQUEST_ID);
if (StrUtil.isEmpty(requestId)) {
requestId = jsonObject.getStr(PayloadField.MESSAGE_ID);
}
Object params = jsonObject.get(PayloadField.PARAMS);
return new PayloadInfo(requestId, params);
} catch (Exception e) {
log.warn("[parsePayloadInfo][解析失败，返回原始字符串] 负载: {}", payload);
return new PayloadInfo(null, payload);
}
}
/**
* 从设备地址解析设备ID
*
* @param deviceAddr 设备地址字符串
* @return 设备ID
*/
private Long parseDeviceId(String deviceAddr) {
if (StrUtil.isEmpty(deviceAddr)) {
log.warn("[parseDeviceId][设备地址为空，返回默认ID]");
return 0L;
}
try {
// 尝试直接解析为Long
return Long.parseLong(deviceAddr);
} catch (NumberFormatException e) {
// 如果不是纯数字，可以使用哈希值或其他策略
log.warn("[parseDeviceId][设备地址不是数字格式: {}，使用哈希值]", deviceAddr);
return (long) deviceAddr.hashCode();
}
}
/**
* 生成服务ID
*
* @param dataPackage TCP数据包
* @return 服务ID
*/
private String generateServerId(TcpDataPackage dataPackage) {
// 使用协议类型 + 设备地址 + 消息序号生成唯一的服务ID
return String.format("tcp_%s_%d", dataPackage.getAddr(), dataPackage.getMid());
}
// ==================== 内部辅助方法 ====================
/**
* 编码 TCP 数据包
*
* @param dataPackage 数据包对象
* @return 编码后的字节流
* @throws IllegalArgumentException 如果数据包对象不正确
*/
private Buffer encodeTcpDataPackage(TcpDataPackage dataPackage) {
if (dataPackage == null) {
throw new IllegalArgumentException("数据包对象不能为空");
}
// 验证数据包
if (dataPackage.getAddr() == null || dataPackage.getAddr().isEmpty()) {
throw new IllegalArgumentException("设备地址不能为空");
}
if (dataPackage.getPayload() == null) {
throw new IllegalArgumentException("负载不能为空");
}
try {
Buffer buffer = Buffer.buffer();
// 1. 计算包体长度（除了包头 4 字节）
int payloadLength = dataPackage.getPayload().getBytes().length;
int totalLength = 2 + dataPackage.getAddr().length() + 2 + 2 + payloadLength;
// 2.1 写入包头：总长度（4 字节）
buffer.appendInt(totalLength);
// 2.2 写入设备地址长度（2 字节）
buffer.appendShort((short) dataPackage.getAddr().length());
// 2.3 写入设备地址（不定长）
buffer.appendBytes(dataPackage.getAddr().getBytes());
// 2.4 写入功能码（2 字节）
buffer.appendShort(dataPackage.getCode());
// 2.5 写入消息序号（2 字节）
buffer.appendShort(dataPackage.getMid());
// 2.6 写入包体数据（不定长）
buffer.appendBytes(dataPackage.getPayload().getBytes());
if (log.isDebugEnabled()) {
log.debug("[encodeTcpDataPackage][编码成功] 设备地址: {}, 功能码: {}, 消息序号: {}, 总长度: {}",
dataPackage.getAddr(), dataPackage.getCode(), dataPackage.getMid(), buffer.length());
}
return buffer;
} catch (Exception e) {
log.error("[encodeTcpDataPackage][编码失败] 数据包: {}", dataPackage, e);
throw new IllegalArgumentException("数据包编码失败: " + e.getMessage(), e);
}
}
/**
* 解码 TCP 数据包
*
* @param buffer 数据缓冲区
* @return 解码后的数据包
* @throws IllegalArgumentException 如果数据包格式不正确
*/
private TcpDataPackage decodeTcpDataPackage(Buffer buffer) {
if (buffer == null || buffer.length() < 8) {
throw new IllegalArgumentException("数据包长度不足");
}
try {
int index = 0;
// 1.1 跳过包头（4字节）
index += 4;
// 1.2 获取设备地址长度（2字节）
short addrLength = buffer.getShort(index);
index += 2;
// 1.3 获取设备地址
String addr = buffer.getBuffer(index, index + addrLength).toString();
index += addrLength;
// 1.4 获取功能码（2字节）
short code = buffer.getShort(index);
index += 2;
// 1.5 获取消息序号（2字节）
short mid = buffer.getShort(index);
index += 2;
// 1.6 获取包体数据
String payload = "";
if (index < buffer.length()) {
payload = buffer.getString(index, buffer.length());
}
// 2. 构建数据包对象
TcpDataPackage dataPackage = new TcpDataPackage(addr, code, mid, payload);
if (log.isDebugEnabled()) {
log.debug("[decodeTcpDataPackage][解码成功] 设备地址: {}, 功能码: {}, 消息序号: {}, 包体长度: {}",
addr, code, mid, payload.length());
}
return dataPackage;
} catch (Exception e) {
log.error("[decodeTcpDataPackage][解码失败] 数据长度: {}", buffer.length(), e);
throw new IllegalArgumentException("数据包解码失败: " + e.getMessage(), e);
}
}
/**
* 消息方法常量
*/
public static class MessageMethod {
public static final String PROPERTY_POST = "thing.property.post"; // 数据上报
public static final String STATE_ONLINE = "thing.state.online"; // 心跳
}
/**
* 负载字段名
*/
private static class PayloadField {
public static final String METHOD = "method";
public static final String PARAMS = "params";
public static final String TIMESTAMP = "timestamp";
public static final String REQUEST_ID = "requestId";
public static final String MESSAGE_ID = "msgId";
}
// ==================== TCP 数据包编解码方法 ====================
/**
* 负载信息类
*/
private static class PayloadInfo {
private String requestId;
private Object params;
public PayloadInfo(String requestId, Object params) {
this.requestId = requestId;
this.params = params;
}
public String getRequestId() { return requestId; }
public Object getParams() { return params; }
}
/**
* TCP 数据包内部类
*/
@Data
private static class TcpDataPackage {
// 功能码定义
public static final short CODE_REGISTER = 10;
public static final short CODE_REGISTER_REPLY = 11;
public static final short CODE_HEARTBEAT = 20;
public static final short CODE_HEARTBEAT_REPLY = 21;
public static final short CODE_MESSAGE_UP = 30;
public static final short CODE_MESSAGE_DOWN = 40;
private String addr;
private short code;
private short mid;
private String payload;
public TcpDataPackage(String addr, short code, short mid, String payload) {
this.addr = addr;
this.code = code;
this.mid = mid;
this.payload = payload;
}
}
// ==================== 自定义异常 ====================
/**
* TCP 编解码异常
*/
public static class TcpCodecException extends RuntimeException {
public TcpCodecException(String message) {
super(message);
}
public TcpCodecException(String message, Throwable cause) {
super(message, cause);
}
}
}

136
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/codec/tcp/IotTcpCodecManager.java Normal file
View File

@@ -0,0 +1,136 @@
package cn.iocoder.yudao.module.iot.gateway.codec.tcp;
import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
import cn.iocoder.yudao.module.iot.gateway.codec.IotDeviceMessageCodec;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
/**
* TCP编解码器管理器（简化版）
*
* 核心功能：
* - 自动协议检测（二进制 vs JSON）
* - 统一编解码接口
* - 默认使用JSON协议
*
* @author 芋道源码
*/
@Slf4j
@Component
public class IotTcpCodecManager implements IotDeviceMessageCodec {
public static final String TYPE = "TCP";
@Autowired
private IotTcpBinaryDeviceMessageCodec binaryCodec;
@Autowired
private IotTcpJsonDeviceMessageCodec jsonCodec;
/**
* 当前默认协议（JSON）
*/
private boolean useJsonByDefault = true;
@Override
public String type() {
return TYPE;
}
@Override
public byte[] encode(IotDeviceMessage message) {
// 默认使用JSON协议编码
return jsonCodec.encode(message);
}
@Override
public IotDeviceMessage decode(byte[] bytes) {
// 自动检测协议类型并解码
if (isJsonFormat(bytes)) {
if (log.isDebugEnabled()) {
log.debug("[decode][检测到JSON协议] 数据长度: {}字节", bytes.length);
}
return jsonCodec.decode(bytes);
} else {
if (log.isDebugEnabled()) {
log.debug("[decode][检测到二进制协议] 数据长度: {}字节", bytes.length);
}
return binaryCodec.decode(bytes);
}
}
// ==================== 便捷方法 ====================
/**
* 使用JSON协议编码
*/
public byte[] encodeJson(IotDeviceMessage message) {
return jsonCodec.encode(message);
}
/**
* 使用二进制协议编码
*/
public byte[] encodeBinary(IotDeviceMessage message) {
return binaryCodec.encode(message);
}
/**
* 获取当前默认协议
*/
public String getDefaultProtocol() {
return useJsonByDefault ? "JSON" : "BINARY";
}
/**
* 设置默认协议
*/
public void setDefaultProtocol(boolean useJson) {
this.useJsonByDefault = useJson;
log.info("[setDefaultProtocol][设置默认协议] 使用JSON: {}", useJson);
}
// ==================== 内部方法 ====================
/**
* 检测是否为JSON格式
*
* 检测规则：
* 1. 数据以 '{' 开头
* 2. 包含 "method" 或 "id" 字段
*/
private boolean isJsonFormat(byte[] bytes) {
if (bytes == null || bytes.length == 0) {
return useJsonByDefault;
}
try {
// 检测JSON格式：以 '{' 开头
if (bytes[0] == '{') {
// 进一步验证是否为有效JSON
String jsonStr = new String(bytes, 0, Math.min(bytes.length, 100));
return jsonStr.contains("\"method\"") || jsonStr.contains("\"id\"");
}
// 检测二进制格式：长度 >= 8 且符合二进制协议结构
if (bytes.length >= 8) {
// 读取包头（前4字节表示后续数据长度）
int expectedLength = ((bytes[0] & 0xFF) << 24) |
((bytes[1] & 0xFF) << 16) |
((bytes[2] & 0xFF) << 8) |
(bytes[3] & 0xFF);
// 验证长度是否合理
if (expectedLength == bytes.length - 4 && expectedLength > 0 && expectedLength < 1024 * 1024) {
return false; // 二进制格式
}
}
} catch (Exception e) {
log.warn("[isJsonFormat][协议检测异常] 使用默认协议: {}", getDefaultProtocol(), e);
}
// 默认使用当前设置的协议类型
return useJsonByDefault;
}
}

389
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/codec/tcp/IotTcpDeviceMessageCodec.java
View File

@@ -1,389 +0,0 @@
package cn.iocoder.yudao.module.iot.gateway.codec.tcp;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONException;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
import cn.iocoder.yudao.module.iot.gateway.codec.IotDeviceMessageCodec;
import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.protocol.TcpDataDecoder;
import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.protocol.TcpDataEncoder;
import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.protocol.TcpDataPackage;
import io.vertx.core.buffer.Buffer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
/**
* TCP {@link IotDeviceMessage} 编解码器
* <p>
* 参考 EMQX 设计理念：
* 1. 高性能编解码
* 2. 容错机制
* 3. 缓存优化
* 4. 监控统计
* 5. 资源管理
*
* @author 芋道源码
*/
@Component
@Slf4j
public class IotTcpDeviceMessageCodec implements IotDeviceMessageCodec {
/**
* 编解码器类型
*/
public static final String TYPE = "tcp";
// ==================== 方法映射 ====================
/**
* 消息方法到功能码的映射
*/
private static final Map<String, Short> METHOD_TO_CODE_MAP = new ConcurrentHashMap<>();
/**
* 功能码到消息方法的映射
*/
private static final Map<Short, String> CODE_TO_METHOD_MAP = new ConcurrentHashMap<>();
static {
// 初始化方法映射
// TODO @haohao：有没可能去掉这个 code 到 method 的映射哈？
initializeMethodMappings();
}
// ==================== 缓存管理 ====================
/**
* JSON 缓存，提升编解码性能
*/
private final Map<String, JSONObject> jsonCache = new ConcurrentHashMap<>();
/**
* 缓存最大大小
*/
private static final int MAX_CACHE_SIZE = 1000;
// ==================== 常量定义 ====================
/**
* 负载字段名
*/
public static class PayloadField {
public static final String TIMESTAMP = "timestamp";
public static final String MESSAGE_ID = "msgId";
public static final String DEVICE_ID = "deviceId";
public static final String PARAMS = "params";
public static final String DATA = "data";
public static final String CODE = "code";
public static final String MESSAGE = "message";
}
/**
* 消息方法映射
*/
public static class MessageMethod {
public static final String PROPERTY_POST = "thing.property.post";
public static final String PROPERTY_SET = "thing.property.set";
public static final String PROPERTY_GET = "thing.property.get";
public static final String EVENT_POST = "thing.event.post";
public static final String SERVICE_INVOKE = "thing.service.invoke";
public static final String CONFIG_PUSH = "thing.config.push";
public static final String OTA_UPGRADE = "thing.ota.upgrade";
public static final String STATE_ONLINE = "thing.state.online";
public static final String STATE_OFFLINE = "thing.state.offline";
}
// ==================== 初始化方法 ====================
/**
* 初始化方法映射
*/
private static void initializeMethodMappings() {
METHOD_TO_CODE_MAP.put(MessageMethod.PROPERTY_POST, TcpDataPackage.CODE_DATA_UP);
METHOD_TO_CODE_MAP.put(MessageMethod.PROPERTY_SET, TcpDataPackage.CODE_PROPERTY_SET);
METHOD_TO_CODE_MAP.put(MessageMethod.PROPERTY_GET, TcpDataPackage.CODE_PROPERTY_GET);
METHOD_TO_CODE_MAP.put(MessageMethod.EVENT_POST, TcpDataPackage.CODE_EVENT_UP);
METHOD_TO_CODE_MAP.put(MessageMethod.SERVICE_INVOKE, TcpDataPackage.CODE_SERVICE_INVOKE);
METHOD_TO_CODE_MAP.put(MessageMethod.CONFIG_PUSH, TcpDataPackage.CODE_DATA_DOWN);
METHOD_TO_CODE_MAP.put(MessageMethod.OTA_UPGRADE, TcpDataPackage.CODE_DATA_DOWN);
METHOD_TO_CODE_MAP.put(MessageMethod.STATE_ONLINE, TcpDataPackage.CODE_HEARTBEAT);
METHOD_TO_CODE_MAP.put(MessageMethod.STATE_OFFLINE, TcpDataPackage.CODE_HEARTBEAT);
// 反向映射
METHOD_TO_CODE_MAP.forEach((method, code) -> CODE_TO_METHOD_MAP.put(code, method));
}
// ==================== 编解码方法 ====================
@Override
public byte[] encode(IotDeviceMessage message) {
validateEncodeParams(message);
try {
if (log.isDebugEnabled()) {
log.debug("[encode][开始编码 TCP 消息] 方法: {}, 消息ID: {}",
message.getMethod(), message.getRequestId());
}
// 1. 获取功能码
short code = getCodeByMethodSafely(message.getMethod());
// 2. 构建负载
String payload = buildPayloadOptimized(message);
// 3. 构建 TCP 数据包
TcpDataPackage dataPackage = TcpDataPackage.builder()
.addr("")
.code(code)
.mid((short) 0)
.payload(payload)
.build();
// 4. 编码为字节流
Buffer buffer = TcpDataEncoder.encode(dataPackage);
byte[] result = buffer.getBytes();
// 5. 统计信息
if (log.isDebugEnabled()) {
log.debug("[encode][TCP 消息编码成功] 方法: {}, 数据长度: {}",
message.getMethod(), result.length);
}
return result;
} catch (Exception e) {
log.error("[encode][TCP 消息编码失败] 消息: {}", message, e);
throw new TcpCodecException("TCP 消息编码失败", e);
}
}
@Override
public IotDeviceMessage decode(byte[] bytes) {
validateDecodeParams(bytes);
try {
if (log.isDebugEnabled()) {
log.debug("[decode][开始解码 TCP 消息] 数据长度: {}", bytes.length);
}
// 1. 解码 TCP 数据包
Buffer buffer = Buffer.buffer(bytes);
TcpDataPackage dataPackage = TcpDataDecoder.decode(buffer);
// 2. 获取消息方法
String method = getMethodByCodeSafely(dataPackage.getCode());
// 3. 解析负载数据
Object params = parsePayloadOptimized(dataPackage.getPayload());
// 4. 构建 IoT 设备消息
IotDeviceMessage message = IotDeviceMessage.requestOf(method, params);
// 5. 统计信息
if (log.isDebugEnabled()) {
log.debug("[decode][TCP 消息解码成功] 方法: {}, 功能码: {}",
method, dataPackage.getCode());
}
return message;
} catch (Exception e) {
log.error("[decode][TCP 消息解码失败] 数据长度: {}, 数据内容: {}",
bytes.length, truncateData(bytes, 100), e);
throw new TcpCodecException("TCP 消息解码失败", e);
}
}
@Override
public String type() {
return TYPE;
}
// ==================== 内部辅助方法 ====================
/**
* 验证编码参数
*/
private void validateEncodeParams(IotDeviceMessage message) {
if (Objects.isNull(message)) {
throw new IllegalArgumentException("IoT 设备消息不能为空");
}
if (StrUtil.isEmpty(message.getMethod())) {
throw new IllegalArgumentException("消息方法不能为空");
}
}
/**
* 验证解码参数
*/
private void validateDecodeParams(byte[] bytes) {
if (Objects.isNull(bytes) || bytes.length == 0) {
throw new IllegalArgumentException("待解码数据不能为空");
}
if (bytes.length > 1024 * 1024) {
throw new IllegalArgumentException("数据包过大，超过 1MB 限制");
}
}
/**
* 安全获取功能码
*/
private short getCodeByMethodSafely(String method) {
Short code = METHOD_TO_CODE_MAP.get(method);
// 默认为数据上报
if (code == null) {
log.warn("[getCodeByMethodSafely][未知的消息方法: {}，使用默认功能码]", method);
return TcpDataPackage.CODE_DATA_UP;
}
return code;
}
/**
* 安全获取消息方法
*/
private String getMethodByCodeSafely(short code) {
String method = CODE_TO_METHOD_MAP.get(code);
if (method == null) {
log.warn("[getMethodByCodeSafely][未知的功能码: {}，使用默认方法]", code);
return MessageMethod.PROPERTY_POST; // 默认为属性上报
}
return method;
}
/**
* 优化的负载构建
*/
private String buildPayloadOptimized(IotDeviceMessage message) {
// 使用缓存键
// TODO @haohao：是不是不用缓存哈？
String cacheKey = message.getMethod() + "_" + message.getRequestId();
JSONObject cachedPayload = jsonCache.get(cacheKey);
if (cachedPayload != null) {
// 更新时间戳
cachedPayload.set(PayloadField.TIMESTAMP, System.currentTimeMillis());
return cachedPayload.toString();
}
// 创建新的负载
JSONObject payload = new JSONObject();
// 添加基础字段
addToPayloadIfNotNull(payload, PayloadField.MESSAGE_ID, message.getRequestId());
addToPayloadIfNotNull(payload, PayloadField.DEVICE_ID, message.getDeviceId());
addToPayloadIfNotNull(payload, PayloadField.PARAMS, message.getParams());
addToPayloadIfNotNull(payload, PayloadField.DATA, message.getData());
addToPayloadIfNotNull(payload, PayloadField.CODE, message.getCode());
addToPayloadIfNotEmpty(payload, PayloadField.MESSAGE, message.getMsg());
// 添加时间戳
payload.set(PayloadField.TIMESTAMP, System.currentTimeMillis());
// 缓存管理
if (jsonCache.size() < MAX_CACHE_SIZE) {
jsonCache.put(cacheKey, payload);
} else {
cleanJsonCacheIfNeeded();
}
return payload.toString();
}
/**
* 优化的负载解析
*/
private Object parsePayloadOptimized(String payload) {
if (StrUtil.isEmpty(payload)) {
return null;
}
try {
// 尝试从缓存获取
JSONObject cachedJson = jsonCache.get(payload);
if (cachedJson != null) {
return cachedJson.containsKey(PayloadField.PARAMS) ? cachedJson.get(PayloadField.PARAMS) : cachedJson;
}
// 解析 JSON 对象
JSONObject jsonObject = JSONUtil.parseObj(payload);
// 缓存解析结果
if (jsonCache.size() < MAX_CACHE_SIZE) {
jsonCache.put(payload, jsonObject);
}
return jsonObject.containsKey(PayloadField.PARAMS) ? jsonObject.get(PayloadField.PARAMS) : jsonObject;
} catch (JSONException e) {
log.warn("[parsePayloadOptimized][负载解析为JSON失败，返回原始字符串] 负载: {}", payload);
return payload;
} catch (Exception e) {
log.error("[parsePayloadOptimized][负载解析异常] 负载: {}", payload, e);
return payload;
}
}
/**
* 添加非空值到负载
*/
private void addToPayloadIfNotNull(JSONObject json, String key, Object value) {
if (ObjectUtil.isNotNull(value)) {
json.set(key, value);
}
}
/**
* 添加非空字符串到负载
*/
private void addToPayloadIfNotEmpty(JSONObject json, String key, String value) {
if (StrUtil.isNotEmpty(value)) {
json.set(key, value);
}
}
/**
* 清理JSON缓存
*/
private void cleanJsonCacheIfNeeded() {
if (jsonCache.size() > MAX_CACHE_SIZE) {
// 清理一半的缓存
int clearCount = jsonCache.size() / 2;
jsonCache.entrySet().removeIf(entry -> clearCount > 0 && Math.random() < 0.5);
if (log.isDebugEnabled()) {
log.debug("[cleanJsonCacheIfNeeded][JSON 缓存已清理] 当前缓存大小: {}", jsonCache.size());
}
}
}
/**
* 截断数据用于日志输出
*/
private String truncateData(byte[] data, int maxLength) {
if (data.length <= maxLength) {
return new String(data, StandardCharsets.UTF_8);
}
byte[] truncated = new byte[maxLength];
System.arraycopy(data, 0, truncated, 0, maxLength);
return new String(truncated, StandardCharsets.UTF_8) + "...(截断)";
}
// ==================== 自定义异常 ====================
/**
* TCP 编解码异常
*/
public static class TcpCodecException extends RuntimeException {
public TcpCodecException(String message) {
super(message);
}
public TcpCodecException(String message, Throwable cause) {
super(message, cause);
}
}
}

245
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/codec/tcp/IotTcpJsonDeviceMessageCodec.java Normal file
View File

@@ -0,0 +1,245 @@
package cn.iocoder.yudao.module.iot.gateway.codec.tcp;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
import cn.iocoder.yudao.module.iot.gateway.codec.IotDeviceMessageCodec;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import java.nio.charset.StandardCharsets;
/**
* TCP JSON格式 {@link IotDeviceMessage} 编解码器
*
* 采用纯JSON格式传输，参考EMQX和HTTP模块的数据格式
*
* JSON消息格式：
* {
* "id": "消息ID",
* "method": "消息方法",
* "deviceId": "设备ID",
* "productKey": "产品Key",
* "deviceName": "设备名称",
* "params": {...},
* "timestamp": 时间戳
* }
*
* @author 芋道源码
*/
@Slf4j
@Component
public class IotTcpJsonDeviceMessageCodec implements IotDeviceMessageCodec {
public static final String TYPE = "TCP_JSON";
// ==================== 常量定义 ====================
@Override
public String type() {
return TYPE;
}
@Override
public byte[] encode(IotDeviceMessage message) {
if (message == null || StrUtil.isEmpty(message.getMethod())) {
throw new IllegalArgumentException("消息或方法不能为空");
}
try {
// 构建JSON消息
JSONObject jsonMessage = buildJsonMessage(message);
// 转换为字节数组
String jsonString = jsonMessage.toString();
byte[] result = jsonString.getBytes(StandardCharsets.UTF_8);
if (log.isDebugEnabled()) {
log.debug("[encode][编码成功] 方法: {}, JSON长度: {}字节, 内容: {}",
message.getMethod(), result.length, jsonString);
}
return result;
} catch (Exception e) {
log.error("[encode][编码失败] 方法: {}", message.getMethod(), e);
throw new RuntimeException("JSON消息编码失败", e);
}
}
// ==================== 编解码方法 ====================
@Override
public IotDeviceMessage decode(byte[] bytes) {
if (bytes == null || bytes.length == 0) {
throw new IllegalArgumentException("待解码数据不能为空");
}
try {
// 转换为JSON字符串
String jsonString = new String(bytes, StandardCharsets.UTF_8);
if (log.isDebugEnabled()) {
log.debug("[decode][开始解码] JSON长度: {}字节, 内容: {}", bytes.length, jsonString);
}
// 解析JSON消息
JSONObject jsonMessage = JSONUtil.parseObj(jsonString);
// 构建IoT设备消息
IotDeviceMessage message = parseJsonMessage(jsonMessage);
if (log.isDebugEnabled()) {
log.debug("[decode][解码成功] 消息ID: {}, 方法: {}, 设备ID: {}",
message.getId(), message.getMethod(), message.getDeviceId());
}
return message;
} catch (Exception e) {
log.error("[decode][解码失败] 数据长度: {}", bytes.length, e);
throw new RuntimeException("JSON消息解码失败", e);
}
}
/**
* 编码数据上报消息
*/
public byte[] encodeDataReport(Object params, Long deviceId, String productKey, String deviceName) {
IotDeviceMessage message = createMessage(MessageMethod.PROPERTY_POST, params, deviceId, productKey, deviceName);
return encode(message);
}
/**
* 编码心跳消息
*/
public byte[] encodeHeartbeat(Long deviceId, String productKey, String deviceName) {
IotDeviceMessage message = createMessage(MessageMethod.STATE_ONLINE, null, deviceId, productKey, deviceName);
return encode(message);
}
// ==================== 便捷方法 ====================
/**
* 编码事件上报消息
*/
public byte[] encodeEventReport(Object params, Long deviceId, String productKey, String deviceName) {
IotDeviceMessage message = createMessage(MessageMethod.EVENT_POST, params, deviceId, productKey, deviceName);
return encode(message);
}
/**
* 构建JSON消息
*/
private JSONObject buildJsonMessage(IotDeviceMessage message) {
JSONObject jsonMessage = new JSONObject();
// 基础字段
jsonMessage.set(JsonField.ID, StrUtil.isNotEmpty(message.getId()) ? message.getId() : IdUtil.fastSimpleUUID());
jsonMessage.set(JsonField.METHOD, message.getMethod());
jsonMessage.set(JsonField.TIMESTAMP, System.currentTimeMillis());
// 设备信息
if (message.getDeviceId() != null) {
jsonMessage.set(JsonField.DEVICE_ID, message.getDeviceId());
}
// 参数
if (message.getParams() != null) {
jsonMessage.set(JsonField.PARAMS, message.getParams());
}
// 响应码和消息（用于下行消息）
if (message.getCode() != null) {
jsonMessage.set(JsonField.CODE, message.getCode());
}
if (StrUtil.isNotEmpty(message.getMsg())) {
jsonMessage.set(JsonField.MESSAGE, message.getMsg());
}
return jsonMessage;
}
/**
* 解析JSON消息
*/
private IotDeviceMessage parseJsonMessage(JSONObject jsonMessage) {
// 提取基础字段
String id = jsonMessage.getStr(JsonField.ID);
String method = jsonMessage.getStr(JsonField.METHOD);
Object params = jsonMessage.get(JsonField.PARAMS);
// 创建消息对象
IotDeviceMessage message = IotDeviceMessage.requestOf(id, method, params);
// 设置设备信息
Long deviceId = jsonMessage.getLong(JsonField.DEVICE_ID);
if (deviceId != null) {
message.setDeviceId(deviceId);
}
// 设置响应信息
Integer code = jsonMessage.getInt(JsonField.CODE);
if (code != null) {
message.setCode(code);
}
String msg = jsonMessage.getStr(JsonField.MESSAGE);
if (StrUtil.isNotEmpty(msg)) {
message.setMsg(msg);
}
// 设置服务ID（基于JSON格式）
message.setServerId(generateServerId(jsonMessage));
return message;
}
// ==================== 内部辅助方法 ====================
/**
* 创建消息对象
*/
private IotDeviceMessage createMessage(String method, Object params, Long deviceId, String productKey, String deviceName) {
IotDeviceMessage message = IotDeviceMessage.requestOf(method, params);
message.setDeviceId(deviceId);
return message;
}
/**
* 生成服务ID
*/
private String generateServerId(JSONObject jsonMessage) {
String id = jsonMessage.getStr(JsonField.ID);
Long deviceId = jsonMessage.getLong(JsonField.DEVICE_ID);
return String.format("tcp_json_%s_%s", deviceId != null ? deviceId : "unknown",
StrUtil.isNotEmpty(id) ? id.substring(0, Math.min(8, id.length())) : "noId");
}
/**
* 消息方法常量
*/
public static class MessageMethod {
public static final String PROPERTY_POST = "thing.property.post"; // 数据上报
public static final String STATE_ONLINE = "thing.state.online"; // 心跳
public static final String EVENT_POST = "thing.event.post"; // 事件上报
public static final String PROPERTY_SET = "thing.property.set"; // 属性设置
public static final String PROPERTY_GET = "thing.property.get"; // 属性获取
public static final String SERVICE_INVOKE = "thing.service.invoke"; // 服务调用
}
/**
* JSON字段名（参考EMQX和HTTP模块格式）
*/
private static class JsonField {
public static final String ID = "id";
public static final String METHOD = "method";
public static final String DEVICE_ID = "deviceId";
public static final String PRODUCT_KEY = "productKey";
public static final String DEVICE_NAME = "deviceName";
public static final String PARAMS = "params";
public static final String TIMESTAMP = "timestamp";
public static final String CODE = "code";
public static final String MESSAGE = "message";
}
}

16
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/config/IotGatewayConfiguration.java
View File

@@ -2,6 +2,7 @@ package cn.iocoder.yudao.module.iot.gateway.config;
import cn.iocoder.yudao.module.iot.core.biz.IotDeviceCommonApi;
import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;
import cn.iocoder.yudao.module.iot.gateway.codec.tcp.IotTcpCodecManager;
import cn.iocoder.yudao.module.iot.gateway.protocol.emqx.IotEmqxAuthEventProtocol;
import cn.iocoder.yudao.module.iot.gateway.protocol.emqx.IotEmqxDownstreamSubscriber;
import cn.iocoder.yudao.module.iot.gateway.protocol.emqx.IotEmqxUpstreamProtocol;
@@ -9,7 +10,6 @@ import cn.iocoder.yudao.module.iot.gateway.protocol.http.IotHttpDownstreamSubscr
import cn.iocoder.yudao.module.iot.gateway.protocol.http.IotHttpUpstreamProtocol;
import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.IotTcpDownstreamSubscriber;
import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.IotTcpUpstreamProtocol;
import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.manager.TcpDeviceConnectionManager;
import cn.iocoder.yudao.module.iot.gateway.service.device.IotDeviceService;
import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
import io.vertx.core.Vertx;
@@ -89,28 +89,22 @@ public class IotGatewayConfiguration {
return Vertx.vertx();
}
@Bean
public TcpDeviceConnectionManager tcpDeviceConnectionManager() {
return new TcpDeviceConnectionManager();
}
@Bean
public IotTcpUpstreamProtocol iotTcpUpstreamProtocol(IotGatewayProperties gatewayProperties,
TcpDeviceConnectionManager connectionManager,
IotDeviceService deviceService,
IotDeviceMessageService messageService,
IotDeviceCommonApi deviceApi,
IotTcpCodecManager codecManager,
Vertx tcpVertx) {
return new IotTcpUpstreamProtocol(gatewayProperties.getProtocol().getTcp(), connectionManager,
deviceService, messageService, deviceApi, tcpVertx);
return new IotTcpUpstreamProtocol(gatewayProperties.getProtocol().getTcp(),
deviceService, messageService, deviceApi, codecManager, tcpVertx);
}
@Bean
public IotTcpDownstreamSubscriber iotTcpDownstreamSubscriber(IotTcpUpstreamProtocol protocolHandler,
TcpDeviceConnectionManager connectionManager,
IotDeviceMessageService messageService,
IotMessageBus messageBus) {
return new IotTcpDownstreamSubscriber(protocolHandler, connectionManager, messageService, messageBus);
return new IotTcpDownstreamSubscriber(protocolHandler, messageService, messageBus);
}
}

140
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/IotTcpDownstreamSubscriber.java
View File

@@ -4,161 +4,67 @@ import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;
import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageSubscriber;
import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils;
import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.manager.TcpDeviceConnectionManager;
import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.router.IotTcpDownstreamHandler;
import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
/**
* IoT 网关 TCP 下游订阅者：接收下行给设备的消息
* <p>
* 参考 EMQX 设计理念：
* 1. 高性能消息路由
* 2. 容错机制
* 3. 状态监控
* 4. 资源管理
*
* @author 芋道源码
*/
@RequiredArgsConstructor
@Slf4j
public class IotTcpDownstreamSubscriber implements IotMessageSubscriber<IotDeviceMessage> {
private final IotTcpUpstreamProtocol protocolHandler;
private final TcpDeviceConnectionManager connectionManager;
private final IotDeviceMessageService messageService;
private final IotTcpDownstreamHandler downstreamHandler;
private final IotMessageBus messageBus;
private volatile IotTcpDownstreamHandler downstreamHandler;
private final IotTcpUpstreamProtocol protocol;
private final AtomicBoolean initialized = new AtomicBoolean(false);
private final AtomicLong processedMessages = new AtomicLong(0);
private final AtomicLong failedMessages = new AtomicLong(0);
public IotTcpDownstreamSubscriber(IotTcpUpstreamProtocol protocol,
IotDeviceMessageService messageService,
IotMessageBus messageBus) {
this.protocol = protocol;
this.messageBus = messageBus;
this.downstreamHandler = new IotTcpDownstreamHandler(messageService);
}
@PostConstruct
public void init() {
if (!initialized.compareAndSet(false, true)) {
log.warn("[init][TCP 下游消息订阅者已初始化，跳过重复初始化]");
return;
}
try {
// 初始化下游处理器
downstreamHandler = new IotTcpDownstreamHandler(connectionManager, messageService);
// 注册到消息总线
messageBus.register(this);
log.info("[init][TCP 下游消息订阅者初始化完成] Topic: {}, Group: {}",
getTopic(), getGroup());
} catch (Exception e) {
initialized.set(false);
log.error("[init][TCP 下游消息订阅者初始化失败]", e);
throw new RuntimeException("TCP 下游消息订阅者初始化失败", e);
}
}
@PreDestroy
public void destroy() {
if (!initialized.get()) {
return;
}
try {
log.info("[destroy][TCP 下游消息订阅者已关闭] 处理消息数: {}, 失败消息数: {}",
processedMessages.get(), failedMessages.get());
} catch (Exception e) {
log.error("[destroy][TCP 下游消息订阅者关闭失败]", e);
} finally {
initialized.set(false);
}
messageBus.register(this);
}
@Override
public String getTopic() {
return IotDeviceMessageUtils.buildMessageBusGatewayDeviceMessageTopic(protocolHandler.getServerId());
return IotDeviceMessageUtils.buildMessageBusGatewayDeviceMessageTopic(protocol.getServerId());
}
@Override
public String getGroup() {
return "tcp-downstream-" + protocolHandler.getServerId();
// 保证点对点消费，需要保证独立的 Group，所以使用 Topic 作为 Group
return getTopic();
}
@Override
public void onMessage(IotDeviceMessage message) {
if (!initialized.get()) {
log.warn("[onMessage][订阅者未初始化，跳过消息处理]");
return;
}
long startTime = System.currentTimeMillis();
log.debug("[onMessage][接收到下行消息, messageId: {}, method: {}, deviceId: {}]",
message.getId(), message.getMethod(), message.getDeviceId());
try {
processedMessages.incrementAndGet();
if (log.isDebugEnabled()) {
log.debug("[onMessage][收到下行消息] 设备 ID: {}, 方法: {}, 消息ID: {}",
message.getDeviceId(), message.getMethod(), message.getId());
}
// 参数校验
if (message.getDeviceId() == null) {
log.warn("[onMessage][下行消息设备 ID 为空，跳过处理] 消息: {}", message);
return;
}
// 检查连接状态
if (connectionManager.getClientByDeviceId(message.getDeviceId()) == null) {
log.warn("[onMessage][设备({})离线，跳过下行消息] 方法: {}",
message.getDeviceId(), message.getMethod());
// 1. 校验
String method = message.getMethod();
if (method == null) {
log.warn("[onMessage][消息方法为空, messageId: {}, deviceId: {}]",
message.getId(), message.getDeviceId());
return;
}
// 处理下行消息
// 2. 处理下行消息
downstreamHandler.handle(message);
// 性能监控
long processTime = System.currentTimeMillis() - startTime;
// TODO @haohao：1000 搞成静态变量；
if (processTime > 1000) { // 超过 1 秒的慢消息
log.warn("[onMessage][慢消息处理] 设备ID: {}, 方法: {}, 耗时: {}ms",
message.getDeviceId(), message.getMethod(), processTime);
}
} catch (Exception e) {
failedMessages.incrementAndGet();
log.error("[onMessage][处理下行消息失败] 设备ID: {}, 方法: {}, 消息: {}",
message.getDeviceId(), message.getMethod(), message, e);
log.error("[onMessage][处理下行消息失败, messageId: {}, method: {}, deviceId: {}]",
message.getId(), message.getMethod(), message.getDeviceId(), e);
}
}
// TODO @haohao：多余的要不先清理掉；
/**
* 获取订阅者统计信息
*/
public String getSubscriberStatistics() {
return String.format("TCP下游订阅者 - 已处理: %d, 失败: %d, 成功率: %.2f%%",
processedMessages.get(),
failedMessages.get(),
processedMessages.get() > 0
? (double) (processedMessages.get() - failedMessages.get()) / processedMessages.get() * 100
: 0.0);
}
/**
* 检查订阅者健康状态
*/
public boolean isHealthy() {
return initialized.get() && downstreamHandler != null;
}
}

129
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/IotTcpUpstreamProtocol.java
View File

@@ -2,8 +2,8 @@ package cn.iocoder.yudao.module.iot.gateway.protocol.tcp;
import cn.iocoder.yudao.module.iot.core.biz.IotDeviceCommonApi;
import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils;
import cn.iocoder.yudao.module.iot.gateway.codec.tcp.IotTcpCodecManager;
import cn.iocoder.yudao.module.iot.gateway.config.IotGatewayProperties;
import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.manager.TcpDeviceConnectionManager;
import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.router.IotTcpUpstreamHandler;
import cn.iocoder.yudao.module.iot.gateway.service.device.IotDeviceService;
import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
@@ -16,19 +16,8 @@ import jakarta.annotation.PreDestroy;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
/**
* IoT 网关 TCP 协议：接收设备上行消息
* <p>
* 负责接收设备上行消息，支持：
* 1. 设备注册
* 2. 心跳保活
* 3. 属性上报
* 4. 事件上报
* 5. 设备连接管理
*
* @author 芋道源码
*/
@@ -37,14 +26,14 @@ public class IotTcpUpstreamProtocol {
private final IotGatewayProperties.TcpProperties tcpProperties;
private final TcpDeviceConnectionManager connectionManager;
private final IotDeviceService deviceService;
private final IotDeviceMessageService messageService;
private final IotDeviceCommonApi deviceApi;
private final IotTcpCodecManager codecManager;
private final Vertx vertx;
@Getter
@@ -53,54 +42,30 @@ public class IotTcpUpstreamProtocol {
private NetServer netServer;
public IotTcpUpstreamProtocol(IotGatewayProperties.TcpProperties tcpProperties,
TcpDeviceConnectionManager connectionManager,
IotDeviceService deviceService,
IotDeviceMessageService messageService,
IotDeviceCommonApi deviceApi,
Vertx vertx) {
IotDeviceService deviceService,
IotDeviceMessageService messageService,
IotDeviceCommonApi deviceApi,
IotTcpCodecManager codecManager,
Vertx vertx) {
this.tcpProperties = tcpProperties;
this.connectionManager = connectionManager;
this.deviceService = deviceService;
this.messageService = messageService;
this.deviceApi = deviceApi;
this.codecManager = codecManager;
this.vertx = vertx;
this.serverId = IotDeviceMessageUtils.generateServerId(tcpProperties.getPort());
}
@PostConstruct
public void start() {
// 1. 启动 TCP 服务器
try {
startTcpServer();
log.info("[start][IoT 网关 TCP 协议处理器启动完成，服务器ID: {}]", serverId);
} catch (Exception e) {
log.error("[start][IoT 网关 TCP 协议处理器启动失败]", e);
// 抛出异常，中断 Spring 容器启动
throw new RuntimeException("IoT 网关 TCP 协议处理器启动失败", e);
}
}
@PreDestroy
public void stop() {
if (netServer != null) {
stopTcpServer();
log.info("[stop][IoT 网关 TCP 协议处理器已停止]");
}
}
/**
* 启动 TCP 服务器
*/
private void startTcpServer() {
// TODO @haohao：同类的，最好使用相同序号前缀，一个方法看起来有段落感。包括同类可以去掉之间的空格。例如说这里的，1. 2. 3. 4. 是初始化；5. 6. 是管理启动
// 1. 创建服务器选项
// 创建服务器选项
NetServerOptions options = new NetServerOptions()
.setPort(tcpProperties.getPort())
.setTcpKeepAlive(true)
.setTcpNoDelay(true)
.setReuseAddress(true);
// 2. 配置 SSL（如果启用）
// 配置 SSL（如果启用）
if (Boolean.TRUE.equals(tcpProperties.getSslEnabled())) {
PemKeyCertOptions pemKeyCertOptions = new PemKeyCertOptions()
.setKeyPath(tcpProperties.getSslKeyPath())
@@ -108,72 +73,32 @@ public class IotTcpUpstreamProtocol {
options.setSsl(true).setKeyCertOptions(pemKeyCertOptions);
}
// 3. 创建 TCP 服务器
// 创建服务器并设置连接处理器
netServer = vertx.createNetServer(options);
// 4. 设置连接处理器
netServer.connectHandler(socket -> {
log.info("[startTcpServer][新设备连接: {}]", socket.remoteAddress());
IotTcpUpstreamHandler handler = new IotTcpUpstreamHandler(
tcpProperties, connectionManager, deviceService, messageService, deviceApi, serverId);
IotTcpUpstreamHandler handler = new IotTcpUpstreamHandler(this, messageService, codecManager);
handler.handle(socket);
});
// 5. 同步启动服务器，等待结果
CountDownLatch latch = new CountDownLatch(1);
AtomicReference<Throwable> failure = new AtomicReference<>();
netServer.listen(result -> {
if (result.succeeded()) {
log.info("[startTcpServer][TCP 服务器启动成功] 端口: {}, 服务器ID: {}",
result.result().actualPort(), serverId);
} else {
log.error("[startTcpServer][TCP 服务器启动失败]", result.cause());
failure.set(result.cause());
}
latch.countDown();
});
// 6. 等待启动结果，设置超时
// 启动服务器
try {
if (!latch.await(10, TimeUnit.SECONDS)) {
throw new RuntimeException("TCP 服务器启动超时");
}
if (failure.get() != null) {
throw new RuntimeException("TCP 服务器启动失败", failure.get());
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException("TCP 服务器启动被中断", e);
netServer.listen().result();
log.info("[start][IoT 网关 TCP 协议启动成功，端口：{}]", tcpProperties.getPort());
} catch (Exception e) {
log.error("[start][IoT 网关 TCP 协议启动失败]", e);
throw e;
}
}
/**
* 停止 TCP 服务器
*/
private void stopTcpServer() {
if (netServer == null) {
return;
}
log.info("[stopTcpServer][准备关闭 TCP 服务器]");
CountDownLatch latch = new CountDownLatch(1);
// 异步关闭，并使用 Latch 等待结果
netServer.close(result -> {
if (result.succeeded()) {
log.info("[stopTcpServer][IoT 网关 TCP 协议处理器已停止]");
} else {
log.warn("[stopTcpServer][TCP 服务器关闭失败]", result.cause());
@PreDestroy
public void stop() {
if (netServer != null) {
try {
netServer.close().result();
log.info("[stop][IoT 网关 TCP 协议已停止]");
} catch (Exception e) {
log.error("[stop][IoT 网关 TCP 协议停止失败]", e);
}
latch.countDown();
});
try {
// 等待关闭完成，设置超时
if (!latch.await(10, TimeUnit.SECONDS)) {
log.warn("[stopTcpServer][关闭 TCP 服务器超时]");
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
log.warn("[stopTcpServer][等待 TCP 服务器关闭被中断]", e);
}
}
}

220
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/client/TcpDeviceClient.java
View File

@@ -1,220 +0,0 @@
package cn.iocoder.yudao.module.iot.gateway.protocol.tcp.client;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.net.NetSocket;
import io.vertx.core.parsetools.RecordParser;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.atomic.AtomicBoolean;
/**
* TCP 设备客户端：封装设备连接的基本信息和操作
* <p>
* 该类中的状态变更（如 authenticated, closed）使用 AtomicBoolean 保证原子性。
* 对 socket 的操作应在 Vert.x Event Loop 线程中执行，以避免并发问题。
*
* @author 芋道源码
*/
@Slf4j
public class TcpDeviceClient {
@Getter
private final String clientId;
@Getter
@Setter
private String deviceAddr; // 从 final 移除，因为在注册后才设置
@Getter
@Setter
private String productKey;
@Getter
@Setter
private String deviceName;
@Getter
@Setter
private Long deviceId;
@Getter
private NetSocket socket;
@Getter
@Setter
private RecordParser parser;
@Getter
private final long keepAliveTimeoutMs;
private volatile long lastKeepAliveTime;
private final AtomicBoolean authenticated = new AtomicBoolean(false);
private final AtomicBoolean closed = new AtomicBoolean(false);
/**
* 构造函数
*
* @param clientId 客户端 ID，全局唯一
* @param keepAliveTimeoutMs 心跳超时时间（毫秒），从配置中读取
*/
public TcpDeviceClient(String clientId, long keepAliveTimeoutMs) {
this.clientId = clientId;
this.keepAliveTimeoutMs = keepAliveTimeoutMs;
this.lastKeepAliveTime = System.currentTimeMillis();
}
/**
* 绑定网络套接字，并设置相关处理器。
* 此方法应在 Vert.x Event Loop 线程中调用
*
* @param socket 网络套接字
*/
public void setSocket(NetSocket socket) {
// 无需 synchronized，Vert.x 保证了同一个 socket 的事件在同一个 Event Loop 中处理
if (this.socket != null && this.socket != socket) {
log.warn("[setSocket][客户端({}) 正在用新的 socket 替换旧的，旧 socket 将被关闭]", clientId);
this.socket.close();
}
this.socket = socket;
// 注册处理器
if (socket != null) {
// 1. 设置关闭处理器
socket.closeHandler(v -> {
log.info("[setSocket][设备客户端({})的连接已由远端关闭]", clientId);
shutdown(); // 统一调用 shutdown 进行资源清理
});
// 2. 设置异常处理器
socket.exceptionHandler(e -> {
log.error("[setSocket][设备客户端({})连接出现异常]", clientId, e);
shutdown(); // 出现异常时也关闭连接
});
// 3. 设置数据处理器
socket.handler(buffer -> {
// 任何数据往来都表示连接是活跃的
keepAlive();
if (parser != null) {
parser.handle(buffer);
} else {
log.warn("[setSocket][设备客户端({}) 未设置解析器(parser)，原始数据被忽略: {}]", clientId, buffer.toString());
}
});
}
}
/**
* 更新心跳时间，表示设备仍然活跃
*/
public void keepAlive() {
this.lastKeepAliveTime = System.currentTimeMillis();
}
/**
* 检查连接是否在线
* 判断标准：未被主动关闭、socket 存在、且在心跳超时时间内
*
* @return 是否在线
*/
public boolean isOnline() {
if (closed.get() || socket == null) {
return false;
}
long idleTime = System.currentTimeMillis() - lastKeepAliveTime;
return idleTime < keepAliveTimeoutMs;
}
// TODO @haohao：1）是不是简化下：productKey 和 deviceName 非空，就认为是已认证；2）如果是的话，productKey 和 deviceName 搞成一个设置方法？setAuthenticated（productKey、deviceName）
public boolean isAuthenticated() {
return authenticated.get();
}
public void setAuthenticated(boolean authenticated) {
this.authenticated.set(authenticated);
}
/**
* 向设备发送消息
*
* @param buffer 消息内容
*/
public void sendMessage(Buffer buffer) {
if (closed.get() || socket == null) {
log.warn("[sendMessage][设备客户端({})连接已关闭，无法发送消息]", clientId);
return;
}
// Vert.x 的 write 是异步的，不会阻塞
socket.write(buffer, result -> {
// 发送失败可能意味着连接已断开，主动关闭
if (!result.succeeded()) {
log.error("[sendMessage][设备客户端({})发送消息失败]", clientId, result.cause());
shutdown();
return;
}
// 发送成功也更新心跳，表示连接活跃
if (log.isDebugEnabled()) {
log.debug("[sendMessage][设备客户端({})发送消息成功]", clientId);
}
keepAlive();
});
}
// TODO @haohao：是不是叫 close 好点？或者问问大模型
/**
* 关闭客户端连接并清理资源。
* 这是一个幂等操作，可以被多次安全调用。
*/
public void shutdown() {
// 使用原子操作保证只执行一次关闭逻辑
if (closed.getAndSet(true)) {
return;
}
log.info("[shutdown][正在关闭设备客户端连接: {}]", clientId);
// 先将 socket 引用置空，再关闭，避免并发问题
NetSocket socketToClose = this.socket;
this.socket = null;
if (socketToClose != null) {
try {
// close 是异步的，但我们在这里不关心其结果，因为我们已经将客户端标记为关闭
socketToClose.close();
} catch (Exception e) {
log.warn("[shutdown][关闭TCP连接时出现异常，可能已被关闭]", e);
}
}
// 重置认证状态
authenticated.set(false);
}
public String getConnectionInfo() {
NetSocket currentSocket = this.socket;
if (currentSocket != null && currentSocket.remoteAddress() != null) {
return currentSocket.remoteAddress().toString();
}
return "disconnected";
}
@Override
public String toString() {
return "TcpDeviceClient{" +
"clientId='" + clientId + '\'' +
", deviceAddr='" + deviceAddr + '\'' +
", deviceId=" + deviceId +
", authenticated=" + authenticated.get() +
", online=" + isOnline() +
", connection=" + getConnectionInfo() +
'}';
}
}

506
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/manager/TcpDeviceConnectionManager.java
View File

@@ -1,506 +0,0 @@
package cn.iocoder.yudao.module.iot.gateway.protocol.tcp.manager;
import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.client.TcpDeviceClient;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.net.NetSocket;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantReadWriteLock;
/**
* TCP 设备连接管理器
* <p>
* 参考 EMQX 设计理念：
* 1. 高性能连接管理
* 2. 连接池和资源管理
* 3. 流量控制 TODO @haohao：这个要不先去掉
* 4. 监控统计 TODO @haohao：这个要不先去掉
* 5. 自动清理和容错
*
* @author 芋道源码
*/
@Component
@Slf4j
public class TcpDeviceConnectionManager {
// ==================== 连接存储 ====================
/**
* 设备客户端映射
* Key: 设备地址, Value: 设备客户端
*/
private final ConcurrentMap<String, TcpDeviceClient> clientMap = new ConcurrentHashMap<>();
/**
* 设备ID到设备地址的映射
* Key: 设备ID, Value: 设备地址
*/
private final ConcurrentMap<Long, String> deviceIdToAddrMap = new ConcurrentHashMap<>();
/**
* 套接字到客户端的映射，用于快速查找
* Key: NetSocket, Value: 设备地址
*/
private final ConcurrentMap<NetSocket, String> socketToAddrMap = new ConcurrentHashMap<>();
// ==================== 读写锁 ====================
private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
private final ReentrantReadWriteLock.ReadLock readLock = lock.readLock();
private final ReentrantReadWriteLock.WriteLock writeLock = lock.writeLock();
// ==================== 定时任务 ====================
/**
* 定时任务执行器
*/
private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(3, r -> {
Thread t = new Thread(r, "tcp-connection-manager");
t.setDaemon(true);
return t;
});
// ==================== 统计信息 ====================
private final AtomicLong totalConnections = new AtomicLong(0);
private final AtomicLong totalDisconnections = new AtomicLong(0);
private final AtomicLong totalMessages = new AtomicLong(0);
private final AtomicLong totalFailedMessages = new AtomicLong(0);
private final AtomicLong totalBytes = new AtomicLong(0);
// ==================== 配置参数 ====================
private static final int MAX_CONNECTIONS = 10000;
private static final int HEARTBEAT_CHECK_INTERVAL = 30; // 秒
private static final int CONNECTION_CLEANUP_INTERVAL = 60; // 秒
private static final int STATS_LOG_INTERVAL = 300; // 秒
/**
* 构造函数，启动定时任务
*/
public TcpDeviceConnectionManager() {
startScheduledTasks();
}
/**
* 启动定时任务
*/
private void startScheduledTasks() {
// 心跳检查任务
scheduler.scheduleAtFixedRate(this::checkHeartbeat,
HEARTBEAT_CHECK_INTERVAL, HEARTBEAT_CHECK_INTERVAL, TimeUnit.SECONDS);
// 连接清理任务
scheduler.scheduleAtFixedRate(this::cleanupConnections,
CONNECTION_CLEANUP_INTERVAL, CONNECTION_CLEANUP_INTERVAL, TimeUnit.SECONDS);
// 统计日志任务
scheduler.scheduleAtFixedRate(this::logStatistics,
STATS_LOG_INTERVAL, STATS_LOG_INTERVAL, TimeUnit.SECONDS);
}
/**
* 添加设备客户端
*/
public boolean addClient(String deviceAddr, TcpDeviceClient client) {
// TODO @haohao：这个要不去掉；目前看着没做 result 的处理；
if (clientMap.size() >= MAX_CONNECTIONS) {
log.warn("[addClient][连接数已达上限({})，拒绝新连接: {}]", MAX_CONNECTIONS, deviceAddr);
return false;
}
writeLock.lock();
try {
log.info("[addClient][添加设备客户端: {}]", deviceAddr);
// 关闭之前的连接（如果存在）
TcpDeviceClient existingClient = clientMap.get(deviceAddr);
if (existingClient != null) {
log.warn("[addClient][设备({})已存在连接，关闭旧连接]", deviceAddr);
removeClientInternal(deviceAddr, existingClient);
}
// 添加新连接
clientMap.put(deviceAddr, client);
// 添加套接字映射
if (client.getSocket() != null) {
socketToAddrMap.put(client.getSocket(), deviceAddr);
}
// 如果客户端已设置设备 ID，更新映射
if (client.getDeviceId() != null) {
deviceIdToAddrMap.put(client.getDeviceId(), deviceAddr);
}
totalConnections.incrementAndGet();
return true;
} finally {
writeLock.unlock();
}
}
/**
* 移除设备客户端
*/
public void removeClient(String deviceAddr) {
writeLock.lock();
try {
TcpDeviceClient client = clientMap.get(deviceAddr);
if (client != null) {
removeClientInternal(deviceAddr, client);
}
} finally {
writeLock.unlock();
}
}
/**
* 内部移除客户端方法（无锁）
*/
private void removeClientInternal(String deviceAddr, TcpDeviceClient client) {
log.info("[removeClient][移除设备客户端: {}]", deviceAddr);
// 从映射中移除
clientMap.remove(deviceAddr);
// 移除套接字映射
if (client.getSocket() != null) {
socketToAddrMap.remove(client.getSocket());
}
// 移除设备ID映射
if (client.getDeviceId() != null) {
deviceIdToAddrMap.remove(client.getDeviceId());
}
// 关闭连接
client.shutdown();
totalDisconnections.incrementAndGet();
}
/**
* 通过设备地址获取客户端
*/
public TcpDeviceClient getClient(String deviceAddr) {
readLock.lock();
try {
return clientMap.get(deviceAddr);
} finally {
readLock.unlock();
}
}
/**
* 通过设备 ID 获取客户端
*/
public TcpDeviceClient getClientByDeviceId(Long deviceId) {
readLock.lock();
try {
String deviceAddr = deviceIdToAddrMap.get(deviceId);
return deviceAddr != null ? clientMap.get(deviceAddr) : null;
} finally {
readLock.unlock();
}
}
// TODO @haohao：getClientBySocket、isDeviceOnline、sendMessage、sendMessageByDeviceId、broadcastMessage 用不到的方法，要不先暂时不提供？保持简洁、更容易理解哈。
/**
* 通过网络连接获取客户端
*/
public TcpDeviceClient getClientBySocket(NetSocket socket) {
readLock.lock();
try {
String deviceAddr = socketToAddrMap.get(socket);
return deviceAddr != null ? clientMap.get(deviceAddr) : null;
} finally {
readLock.unlock();
}
}
/**
* 检查设备是否在线
*/
public boolean isDeviceOnline(Long deviceId) {
TcpDeviceClient client = getClientByDeviceId(deviceId);
return client != null && client.isOnline();
}
/**
* 设置设备 ID 映射
*/
public void setDeviceIdMapping(String deviceAddr, Long deviceId) {
writeLock.lock();
try {
TcpDeviceClient client = clientMap.get(deviceAddr);
if (client != null) {
client.setDeviceId(deviceId);
deviceIdToAddrMap.put(deviceId, deviceAddr);
log.debug("[setDeviceIdMapping][设置设备ID映射: {} -> {}]", deviceAddr, deviceId);
}
} finally {
writeLock.unlock();
}
}
/**
* 发送消息给设备
*/
public boolean sendMessage(String deviceAddr, Buffer buffer) {
TcpDeviceClient client = getClient(deviceAddr);
if (client != null && client.isOnline()) {
try {
client.sendMessage(buffer);
totalMessages.incrementAndGet();
totalBytes.addAndGet(buffer.length());
return true;
} catch (Exception e) {
totalFailedMessages.incrementAndGet();
log.error("[sendMessage][发送消息失败] 设备地址: {}", deviceAddr, e);
return false;
}
}
log.warn("[sendMessage][设备({})不在线，无法发送消息]", deviceAddr);
return false;
}
/**
* 通过设备ID发送消息
*/
public boolean sendMessageByDeviceId(Long deviceId, Buffer buffer) {
TcpDeviceClient client = getClientByDeviceId(deviceId);
if (client != null && client.isOnline()) {
try {
client.sendMessage(buffer);
totalMessages.incrementAndGet();
totalBytes.addAndGet(buffer.length());
return true;
} catch (Exception e) {
totalFailedMessages.incrementAndGet();
log.error("[sendMessageByDeviceId][发送消息失败] 设备ID: {}", deviceId, e);
return false;
}
}
log.warn("[sendMessageByDeviceId][设备ID({})不在线，无法发送消息]", deviceId);
return false;
}
/**
* 广播消息给所有在线设备
*/
public int broadcastMessage(Buffer buffer) {
int successCount = 0;
readLock.lock();
try {
for (TcpDeviceClient client : clientMap.values()) {
if (client.isOnline()) {
try {
client.sendMessage(buffer);
successCount++;
} catch (Exception e) {
log.error("[broadcastMessage][广播消息失败] 设备: {}", client.getDeviceAddr(), e);
}
}
}
} finally {
readLock.unlock();
}
totalMessages.addAndGet(successCount);
totalBytes.addAndGet((long) successCount * buffer.length());
return successCount;
}
/**
* 获取在线设备数量
*/
public int getOnlineCount() {
readLock.lock();
try {
return (int) clientMap.values().stream()
.filter(TcpDeviceClient::isOnline)
.count();
} finally {
readLock.unlock();
}
}
/**
* 获取总连接数
*/
public int getTotalCount() {
return clientMap.size();
}
/**
* 获取认证设备数量
*/
public int getAuthenticatedCount() {
readLock.lock();
try {
return (int) clientMap.values().stream()
.filter(TcpDeviceClient::isAuthenticated)
.count();
} finally {
readLock.unlock();
}
}
// TODO @haohao：心跳超时，需要 close 么？
/**
* 心跳检查任务
*/
private void checkHeartbeat() {
try {
int offlineCount = 0;
readLock.lock();
try {
for (TcpDeviceClient client : clientMap.values()) {
if (!client.isOnline()) {
offlineCount++;
}
}
} finally {
readLock.unlock();
}
if (offlineCount > 0) {
log.info("[checkHeartbeat][发现 {} 个离线设备，将在清理任务中处理]", offlineCount);
}
} catch (Exception e) {
log.error("[checkHeartbeat][心跳检查任务异常]", e);
}
}
/**
* 连接清理任务
*/
private void cleanupConnections() {
try {
int beforeSize = clientMap.size();
writeLock.lock();
try {
clientMap.entrySet().removeIf(entry -> {
TcpDeviceClient client = entry.getValue();
if (!client.isOnline()) {
log.debug("[cleanupConnections][清理离线连接: {}]", entry.getKey());
// 清理相关映射
if (client.getSocket() != null) {
socketToAddrMap.remove(client.getSocket());
}
if (client.getDeviceId() != null) {
deviceIdToAddrMap.remove(client.getDeviceId());
}
client.shutdown();
totalDisconnections.incrementAndGet();
return true;
}
return false;
});
} finally {
writeLock.unlock();
}
int afterSize = clientMap.size();
if (beforeSize != afterSize) {
log.info("[cleanupConnections][清理完成] 连接数: {} -> {}, 清理数: {}",
beforeSize, afterSize, beforeSize - afterSize);
}
} catch (Exception e) {
log.error("[cleanupConnections][连接清理任务异常]", e);
}
}
/**
* 统计日志任务
*/
private void logStatistics() {
try {
long totalConn = totalConnections.get();
long totalDisconnections = this.totalDisconnections.get();
long totalMsg = totalMessages.get();
long totalFailedMsg = totalFailedMessages.get();
long totalBytesValue = totalBytes.get();
log.info("[logStatistics][连接统计] 总连接: {}, 总断开: {}, 当前在线: {}, 认证设备: {}, " +
"总消息: {}, 失败消息: {}, 总字节: {}",
totalConn, totalDisconnections, getOnlineCount(), getAuthenticatedCount(),
totalMsg, totalFailedMsg, totalBytesValue);
} catch (Exception e) {
log.error("[logStatistics][统计日志任务异常]", e);
}
}
/**
* 关闭连接管理器
*/
public void shutdown() {
log.info("[shutdown][关闭TCP连接管理器]");
// 关闭定时任务
scheduler.shutdown();
try {
if (!scheduler.awaitTermination(10, TimeUnit.SECONDS)) {
scheduler.shutdownNow();
}
} catch (InterruptedException e) {
scheduler.shutdownNow();
Thread.currentThread().interrupt();
}
// 关闭所有连接
writeLock.lock();
try {
clientMap.values().forEach(TcpDeviceClient::shutdown);
clientMap.clear();
deviceIdToAddrMap.clear();
socketToAddrMap.clear();
} finally {
writeLock.unlock();
}
}
/**
* 获取连接状态信息
*/
public String getConnectionStatus() {
return String.format("总连接数: %d, 在线设备: %d, 认证设备: %d, 成功率: %.2f%%",
getTotalCount(), getOnlineCount(), getAuthenticatedCount(),
totalMessages.get() > 0
? (double) (totalMessages.get() - totalFailedMessages.get()) / totalMessages.get() * 100
: 0.0);
}
/**
* 获取详细统计信息
*/
public String getDetailedStatistics() {
return String.format(
"TCP连接管理器统计:\n" +
"- 当前连接数: %d\n" +
"- 在线设备数: %d\n" +
"- 认证设备数: %d\n" +
"- 历史总连接: %d\n" +
"- 历史总断开: %d\n" +
"- 总消息数: %d\n" +
"- 失败消息数: %d\n" +
"- 总字节数: %d\n" +
"- 消息成功率: %.2f%%",
getTotalCount(), getOnlineCount(), getAuthenticatedCount(),
totalConnections.get(), totalDisconnections.get(),
totalMessages.get(), totalFailedMessages.get(), totalBytes.get(),
totalMessages.get() > 0
? (double) (totalMessages.get() - totalFailedMessages.get()) / totalMessages.get() * 100
: 0.0);
}
}

98
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/protocol/TcpDataDecoder.java
View File

@@ -1,98 +0,0 @@
package cn.iocoder.yudao.module.iot.gateway.protocol.tcp.protocol;
import io.vertx.core.buffer.Buffer;
import lombok.extern.slf4j.Slf4j;
// TODO @haohao：“设备地址长度”是不是不需要。
/**
* TCP 数据解码器
* <p>
* 负责将字节流解码为 TcpDataPackage 对象
* <p>
* 数据包格式：
* 包头(4 字节长度) | 设备地址长度(2 字节) | 设备地址(不定长) | 功能码(2 字节) | 消息序号(2 字节) | 包体(不定长)
*
* @author 芋道源码
*/
@Slf4j
public class TcpDataDecoder {
/**
* 解码数据包
*
* @param buffer 数据缓冲区
* @return 解码后的数据包
* @throws IllegalArgumentException 如果数据包格式不正确
*/
public static TcpDataPackage decode(Buffer buffer) {
if (buffer == null || buffer.length() < 8) {
throw new IllegalArgumentException("数据包长度不足");
}
try {
int index = 0;
// 1.1 获取设备地址长度（2字节）
short addrLength = buffer.getShort(index);
index += 2;
// 1.2 校验数据包长度
int expectedLength = 2 + addrLength + 2 + 2; // 地址长度 + 地址 + 功能码 + 消息序号
if (buffer.length() < expectedLength) {
throw new IllegalArgumentException("数据包长度不足，期望至少 " + expectedLength + " 字节");
}
// 1.3 获取设备地址
String addr = buffer.getBuffer(index, index + addrLength).toString();
index += addrLength;
// 1.4 获取功能码（2字节）
short code = buffer.getShort(index);
index += 2;
// 1.5 获取消息序号（2字节）
short mid = buffer.getShort(index);
index += 2;
// 1.6 获取包体数据
String payload = "";
if (index < buffer.length()) {
payload = buffer.getString(index, buffer.length());
}
// 2. 构建数据包对象
TcpDataPackage dataPackage = TcpDataPackage.builder()
.addrLength((int) addrLength)
.addr(addr)
.code(code)
.mid(mid)
.payload(payload)
.build();
log.debug("[decode][解码成功] 设备地址: {}, 功能码: {}, 消息序号: {}, 包体长度: {}",
addr, dataPackage.getCodeDescription(), mid, payload.length());
return dataPackage;
} catch (Exception e) {
log.error("[decode][解码失败] 数据: {}", buffer.toString(), e);
throw new IllegalArgumentException("数据包解码失败: " + e.getMessage(), e);
}
}
// TODO @haohao：这个要不去掉，暂时没用到；
/**
* 校验数据包格式
*
* @param buffer 数据缓冲区
* @return 校验结果
*/
public static boolean validate(Buffer buffer) {
try {
decode(buffer);
return true;
} catch (Exception e) {
log.warn("[validate][数据包格式校验失败] 数据: {}, 错误: {}", buffer.toString(), e.getMessage());
return false;
}
}
}

159
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/protocol/TcpDataEncoder.java
View File

@@ -1,159 +0,0 @@
package cn.iocoder.yudao.module.iot.gateway.protocol.tcp.protocol;
import io.vertx.core.buffer.Buffer;
import lombok.extern.slf4j.Slf4j;
/**
* TCP 数据编码器
* <p>
* 负责将 TcpDataPackage 对象编码为字节流
* <p>
* 数据包格式：
* 包头(4字节长度) | 设备地址长度(2字节) | 设备地址(不定长) | 功能码(2字节) | 消息序号(2字节) | 包体(不定长)
*
* @author 芋道源码
*/
@Slf4j
public class TcpDataEncoder {
/**
* 编码数据包
*
* @param dataPackage 数据包对象
* @return 编码后的字节流
* @throws IllegalArgumentException 如果数据包对象不正确
*/
public static Buffer encode(TcpDataPackage dataPackage) {
if (dataPackage == null) {
throw new IllegalArgumentException("数据包对象不能为空");
}
if (dataPackage.getAddr() == null || dataPackage.getAddr().isEmpty()) {
throw new IllegalArgumentException("设备地址不能为空");
}
if (dataPackage.getPayload() == null) {
dataPackage.setPayload("");
}
try {
Buffer buffer = Buffer.buffer();
// 1. 计算包体长度（除了包头 4 字节）
int payloadLength = dataPackage.getPayload().getBytes().length;
int totalLength = 2 + dataPackage.getAddr().length() + 2 + 2 + payloadLength;
// 2.1 写入包头：总长度（4 字节）
buffer.appendInt(totalLength);
// 2.2 写入设备地址长度（2 字节）
buffer.appendShort((short) dataPackage.getAddr().length());
// 2.3 写入设备地址（不定长）
buffer.appendBytes(dataPackage.getAddr().getBytes());
// 2.4 写入功能码（2 字节）
buffer.appendShort(dataPackage.getCode());
// 2.5 写入消息序号（2 字节）
buffer.appendShort(dataPackage.getMid());
// 2.6 写入包体数据（不定长）
buffer.appendBytes(dataPackage.getPayload().getBytes());
log.debug("[encode][编码成功] 设备地址: {}, 功能码: {}, 消息序号: {}, 总长度: {}",
dataPackage.getAddr(), dataPackage.getCodeDescription(),
dataPackage.getMid(), buffer.length());
return buffer;
} catch (Exception e) {
log.error("[encode][编码失败] 数据包: {}", dataPackage, e);
throw new IllegalArgumentException("数据包编码失败: " + e.getMessage(), e);
}
}
/**
* 创建注册回复数据包
*
* @param addr 设备地址
* @param mid 消息序号
* @param success 是否成功
* @return 编码后的数据包
*/
public static Buffer createRegisterReply(String addr, short mid, boolean success) {
// TODO @haohao：payload 默认成功、失败，最好讴有个枚举
String payload = success ? "0" : "1"; // 0 表示成功，1 表示失败
TcpDataPackage dataPackage = TcpDataPackage.builder()
.addr(addr)
.code(TcpDataPackage.CODE_REGISTER_REPLY)
.mid(mid)
.payload(payload)
.build();
return encode(dataPackage);
}
/**
* 创建数据下发数据包
*
* @param addr 设备地址
* @param mid 消息序号
* @param data 下发数据
* @return 编码后的数据包
*/
public static Buffer createDataDownPackage(String addr, short mid, String data) {
TcpDataPackage dataPackage = TcpDataPackage.builder()
.addr(addr)
.code(TcpDataPackage.CODE_DATA_DOWN)
.mid(mid)
.payload(data)
.build();
return encode(dataPackage);
}
/**
* 创建服务调用数据包
*
* @param addr 设备地址
* @param mid 消息序号
* @param serviceData 服务数据
* @return 编码后的数据包
*/
public static Buffer createServiceInvokePackage(String addr, short mid, String serviceData) {
TcpDataPackage dataPackage = TcpDataPackage.builder()
.addr(addr)
.code(TcpDataPackage.CODE_SERVICE_INVOKE)
.mid(mid)
.payload(serviceData)
.build();
return encode(dataPackage);
}
/**
* 创建属性设置数据包
*
* @param addr 设备地址
* @param mid 消息序号
* @param propertyData 属性数据
* @return 编码后的数据包
*/
public static Buffer createPropertySetPackage(String addr, short mid, String propertyData) {
TcpDataPackage dataPackage = TcpDataPackage.builder()
.addr(addr)
.code(TcpDataPackage.CODE_PROPERTY_SET)
.mid(mid)
.payload(propertyData)
.build();
return encode(dataPackage);
}
/**
* 创建属性获取数据包
*
* @param addr 设备地址
* @param mid 消息序号
* @param propertyNames 属性名称列表
* @return 编码后的数据包
*/
public static Buffer createPropertyGetPackage(String addr, short mid, String propertyNames) {
TcpDataPackage dataPackage = TcpDataPackage.builder()
.addr(addr)
.code(TcpDataPackage.CODE_PROPERTY_GET)
.mid(mid)
.payload(propertyNames)
.build();
return encode(dataPackage);
}
}

160
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/protocol/TcpDataPackage.java
View File

@@ -1,160 +0,0 @@
package cn.iocoder.yudao.module.iot.gateway.protocol.tcp.protocol;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
/**
* TCP 数据包协议定义
* <p>
* 数据包格式：
* 包头(4 字节长度) | 设备地址长度(2 字节) | 设备地址(不定长) | 功能码(2 字节) | 消息序号(2 字节) | 包体(不定长)
*
* @author 芋道源码
*/
@Data
@AllArgsConstructor
@NoArgsConstructor
@Builder
public class TcpDataPackage {
// ==================== 功能码定义 ====================
/**
* 设备注册
*/
public static final short CODE_REGISTER = 10;
/**
* 注册回复
*/
public static final short CODE_REGISTER_REPLY = 11;
// TODO @haohao：【重要】一般心跳，服务端会回复一条；回复要搞独立的 code 码，还是继续用原来的，因为 requestId 可以映射；
/**
* 心跳
*/
public static final short CODE_HEARTBEAT = 20;
// TODO @haohao：【重要】下面的，是不是融合成消息上行（client -> server），消息下行（server -> client）；然后把 method 放到 body 里？
/**
* 数据上报
*/
public static final short CODE_DATA_UP = 30;
/**
* 事件上报
*/
public static final short CODE_EVENT_UP = 40;
/**
* 数据下发
*/
public static final short CODE_DATA_DOWN = 50;
/**
* 服务调用
*/
public static final short CODE_SERVICE_INVOKE = 60;
/**
* 属性设置
*/
public static final short CODE_PROPERTY_SET = 70;
/**
* 属性获取
*/
public static final short CODE_PROPERTY_GET = 80;
// ==================== 数据包字段 ====================
// TODO @haohao：设备 addrLength、addr 是不是非必要呀？
/**
* 设备地址长度
*/
private Integer addrLength;
/**
* 设备地址
*/
private String addr;
/**
* 功能码
*/
private short code;
/**
* 消息序号
*/
private short mid;
/**
* 包体数据
*/
private String payload;
// ==================== 辅助方法 ====================
// TODO @haohao：用不到的方法，可以清理掉哈；
/**
* 是否为注册消息
*/
public boolean isRegisterMessage() {
return code == CODE_REGISTER;
}
/**
* 是否为心跳消息
*/
public boolean isHeartbeatMessage() {
return code == CODE_HEARTBEAT;
}
/**
* 是否为数据上报消息
*/
public boolean isDataUpMessage() {
return code == CODE_DATA_UP;
}
/**
* 是否为事件上报消息
*/
public boolean isEventUpMessage() {
return code == CODE_EVENT_UP;
}
/**
* 是否为下行消息
*/
public boolean isDownstreamMessage() {
return code == CODE_DATA_DOWN || code == CODE_SERVICE_INVOKE ||
code == CODE_PROPERTY_SET || code == CODE_PROPERTY_GET;
}
// TODO @haohao：这个是不是去掉呀？多了一些维护成本；
/**
* 获取功能码描述
*/
public String getCodeDescription() {
switch (code) {
case CODE_REGISTER:
return "设备注册";
case CODE_REGISTER_REPLY:
return "注册回复";
case CODE_HEARTBEAT:
return "心跳";
case CODE_DATA_UP:
return "数据上报";
case CODE_EVENT_UP:
return "事件上报";
case CODE_DATA_DOWN:
return "数据下发";
case CODE_SERVICE_INVOKE:
return "服务调用";
case CODE_PROPERTY_SET:
return "属性设置";
case CODE_PROPERTY_GET:
return "属性获取";
default:
return "未知功能码";
}
}
}

162
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/protocol/TcpDataReader.java
View File

@@ -1,162 +0,0 @@
package cn.iocoder.yudao.module.iot.gateway.protocol.tcp.protocol;
import io.vertx.core.Handler;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.parsetools.RecordParser;
import lombok.extern.slf4j.Slf4j;
import java.util.function.Consumer;
/**
* TCP 数据读取器
* <p>
* 负责从 TCP 流中读取完整的数据包
* <p>
* 数据包格式：
* 包头(4 字节长度) | 设备地址长度(2 字节) | 设备地址(不定长) | 功能码(2 字节) | 消息序号(2 字节) | 包体(不定长)
*
* @author 芋道源码
*/
@Slf4j
public class TcpDataReader {
/**
* 创建数据包解析器
*
* @param receiveHandler 接收处理器
* @return RecordParser 解析器
*/
public static RecordParser createParser(Consumer<Buffer> receiveHandler) {
// 首先读取 4 字节的长度信息
RecordParser parser = RecordParser.newFixed(4);
// 设置处理器
parser.setOutput(new Handler<Buffer>() {
// 当前数据包的长度，-1 表示还没有读取到长度信息
private int dataLength = -1;
@Override
public void handle(Buffer buffer) {
try {
// 如果还没有读取到长度信息
if (dataLength == -1) {
// 从包头中读取数据长度
dataLength = buffer.getInt(0);
// 校验数据长度（最大 1 MB）
// TODO @haohao：1m 蛮多地方在写死，最好配置管理下。或者有个全局的枚举；
if (dataLength <= 0 || dataLength > 1024 * 1024) {
log.error("[handle][无效的数据包长度: {}]", dataLength);
reset();
return;
}
// 切换到读取数据模式
parser.fixedSizeMode(dataLength);
log.debug("[handle][读取到数据包长度: {}]", dataLength);
} else {
// 读取到完整的数据包
log.debug("[handle][读取到完整数据包，长度: {}]", buffer.length());
// 处理数据包
try {
receiveHandler.accept(buffer);
} catch (Exception e) {
log.error("[handle][处理数据包失败]", e);
}
// 重置状态，准备读取下一个数据包
reset();
}
} catch (Exception e) {
log.error("[handle][数据包处理异常]", e);
reset();
}
}
/**
* 重置解析器状态
*/
private void reset() {
dataLength = -1;
parser.fixedSizeMode(4);
}
});
return parser;
}
// TODO @haohao：用不到的方法，可以清理掉哈；
/**
* 创建带异常处理的数据包解析器
*
* @param receiveHandler 接收处理器
* @param exceptionHandler 异常处理器
* @return RecordParser 解析器
*/
public static RecordParser createParserWithExceptionHandler(
Consumer<Buffer> receiveHandler,
Consumer<Throwable> exceptionHandler) {
RecordParser parser = RecordParser.newFixed(4);
parser.setOutput(new Handler<Buffer>() {
private int dataLength = -1;
@Override
public void handle(Buffer buffer) {
try {
if (dataLength == -1) {
dataLength = buffer.getInt(0);
if (dataLength <= 0 || dataLength > 1024 * 1024) {
throw new IllegalArgumentException("无效的数据包长度: " + dataLength);
}
parser.fixedSizeMode(dataLength);
log.debug("[handle][读取到数据包长度: {}]", dataLength);
} else {
log.debug("[handle][读取到完整数据包，长度: {}]", buffer.length());
try {
receiveHandler.accept(buffer);
} catch (Exception e) {
exceptionHandler.accept(e);
}
reset();
}
} catch (Exception e) {
exceptionHandler.accept(e);
reset();
}
}
private void reset() {
dataLength = -1;
parser.fixedSizeMode(4);
}
});
return parser;
}
/**
* 创建简单的数据包解析器（用于测试）
*
* @param receiveHandler 接收处理器
* @return RecordParser 解析器
*/
public static RecordParser createSimpleParser(Consumer<TcpDataPackage> receiveHandler) {
return createParser(buffer -> {
try {
TcpDataPackage dataPackage = TcpDataDecoder.decode(buffer);
receiveHandler.accept(dataPackage);
} catch (Exception e) {
log.error("[createSimpleParser][解码数据包失败]", e);
}
});
}
}

336
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/router/IotTcpDownstreamHandler.java
View File

@@ -1,15 +1,8 @@
package cn.iocoder.yudao.module.iot.gateway.protocol.tcp.router;
import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.client.TcpDeviceClient;
import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.manager.TcpDeviceConnectionManager;
import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.protocol.TcpDataDecoder;
import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.protocol.TcpDataEncoder;
import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.protocol.TcpDataPackage;
import cn.iocoder.yudao.module.iot.gateway.codec.tcp.IotTcpDeviceMessageCodec;
import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
import com.alibaba.fastjson.JSON;
import io.vertx.core.buffer.Buffer;
import lombok.extern.slf4j.Slf4j;
/**
@@ -21,20 +14,21 @@ import lombok.extern.slf4j.Slf4j;
* 3. 属性获取
* 4. 配置下发
* 5. OTA 升级
* <p>
* 注意：由于移除了连接管理器，此处理器主要负责消息的编码和日志记录
*
* @author 芋道源码
*/
@Slf4j
public class IotTcpDownstreamHandler {
private final TcpDeviceConnectionManager connectionManager;
private final IotDeviceMessageService messageService;
public IotTcpDownstreamHandler(TcpDeviceConnectionManager connectionManager,
IotDeviceMessageService messageService) {
this.connectionManager = connectionManager;
private final IotTcpDeviceMessageCodec codec;
public IotTcpDownstreamHandler(IotDeviceMessageService messageService) {
this.messageService = messageService;
this.codec = new IotTcpDeviceMessageCodec();
}
/**
@@ -47,315 +41,19 @@ public class IotTcpDownstreamHandler {
log.info("[handle][处理下行消息] 设备ID: {}, 方法: {}, 消息ID: {}",
message.getDeviceId(), message.getMethod(), message.getId());
// 1. 获取设备连接
TcpDeviceClient client = connectionManager.getClientByDeviceId(message.getDeviceId());
if (client == null || !client.isOnline()) {
log.error("[handle][设备({})不在线，无法发送下行消息]", message.getDeviceId());
return;
}
// 编码消息用于日志记录和验证
byte[] encodedMessage = codec.encode(message);
log.debug("[handle][消息编码成功] 设备ID: {}, 编码后长度: {} 字节",
message.getDeviceId(), encodedMessage.length);
// 记录下行消息处理
log.info("[handle][下行消息处理完成] 设备ID: {}, 方法: {}, 消息内容: {}",
message.getDeviceId(), message.getMethod(), message.getParams());
// 2. 根据消息方法处理不同类型的下行消息
// TODO @芋艿、@haohao：看看有没什么办法，减少这样的编码。拓展新消息类型，成本高；
switch (message.getMethod()) {
case "thing.property.set":
handlePropertySet(client, message);
break;
case "thing.property.get":
handlePropertyGet(client, message);
break;
case "thing.service.invoke":
handleServiceInvoke(client, message);
break;
case "thing.config.push":
handleConfigPush(client, message);
break;
case "thing.ota.upgrade":
handleOtaUpgrade(client, message);
break;
default:
log.warn("[handle][未知的下行消息方法: {}]", message.getMethod());
break;
}
} catch (Exception e) {
// TODO @haohao：最好消息的内容，打印下；
log.error("[handle][处理下行消息失败]", e);
log.error("[handle][处理下行消息失败] 设备ID: {}, 方法: {}, 消息内容: {}",
message.getDeviceId(), message.getMethod(), message.getParams(), e);
}
}
/**
* 处理属性设置
*
* @param client 设备客户端
* @param message 设备消息
*/
private void handlePropertySet(TcpDeviceClient client, IotDeviceMessage message) {
try {
log.info("[handlePropertySet][属性设置] 设备地址: {}, 属性: {}",
client.getDeviceAddr(), message.getParams());
// 使用编解码器发送消息，降级处理使用原始编码
sendMessageWithCodec(client, message, "handlePropertySet", () -> {
String payload = JSON.toJSONString(message.getParams());
short mid = generateMessageId();
Buffer buffer = TcpDataEncoder.createPropertySetPackage(
client.getDeviceAddr(), mid, payload);
client.sendMessage(buffer);
log.debug("[handlePropertySet][属性设置消息已发送(降级)] 设备地址: {}, 消息序号: {}",
client.getDeviceAddr(), mid);
});
} catch (Exception e) {
log.error("[handlePropertySet][属性设置失败]", e);
}
}
/**
* 处理属性获取
*
* @param client 设备客户端
* @param message 设备消息
*/
private void handlePropertyGet(TcpDeviceClient client, IotDeviceMessage message) {
try {
log.info("[handlePropertyGet][属性获取] 设备地址: {}, 属性列表: {}",
client.getDeviceAddr(), message.getParams());
// 使用编解码器发送消息，降级处理使用原始编码
sendMessageWithCodec(client, message, "handlePropertyGet", () -> {
String payload = JSON.toJSONString(message.getParams());
short mid = generateMessageId();
Buffer buffer = TcpDataEncoder.createPropertyGetPackage(
client.getDeviceAddr(), mid, payload);
client.sendMessage(buffer);
log.debug("[handlePropertyGet][属性获取消息已发送(降级)] 设备地址: {}, 消息序号: {}",
client.getDeviceAddr(), mid);
});
} catch (Exception e) {
log.error("[handlePropertyGet][属性获取失败]", e);
}
}
/**
* 处理服务调用
*
* @param client 设备客户端
* @param message 设备消息
*/
private void handleServiceInvoke(TcpDeviceClient client, IotDeviceMessage message) {
try {
log.info("[handleServiceInvoke][服务调用] 设备地址: {}, 服务参数: {}",
client.getDeviceAddr(), message.getParams());
// 1. 构建服务调用数据包
String payload = JSON.toJSONString(message.getParams());
short mid = generateMessageId();
Buffer buffer = TcpDataEncoder.createServiceInvokePackage(
client.getDeviceAddr(), mid, payload);
// 2. 发送消息
client.sendMessage(buffer);
log.debug("[handleServiceInvoke][服务调用消息已发送] 设备地址: {}, 消息序号: {}",
client.getDeviceAddr(), mid);
} catch (Exception e) {
log.error("[handleServiceInvoke][服务调用失败]", e);
}
}
/**
* 处理配置推送
*
* @param client 设备客户端
* @param message 设备消息
*/
private void handleConfigPush(TcpDeviceClient client, IotDeviceMessage message) {
try {
log.info("[handleConfigPush][配置推送] 设备地址: {}, 配置: {}",
client.getDeviceAddr(), message.getParams());
// 1. 构建配置推送数据包
String payload = JSON.toJSONString(message.getParams());
short mid = generateMessageId();
Buffer buffer = TcpDataEncoder.createDataDownPackage(
client.getDeviceAddr(), mid, payload);
// 2. 发送消息
client.sendMessage(buffer);
log.debug("[handleConfigPush][配置推送消息已发送] 设备地址: {}, 消息序号: {}",
client.getDeviceAddr(), mid);
} catch (Exception e) {
log.error("[handleConfigPush][配置推送失败]", e);
}
}
/**
* 处理 OTA 升级
*
* @param client 设备客户端
* @param message 设备消息
*/
private void handleOtaUpgrade(TcpDeviceClient client, IotDeviceMessage message) {
try {
log.info("[handleOtaUpgrade][OTA升级] 设备地址: {}, 升级信息: {}",
client.getDeviceAddr(), message.getParams());
// 1. 构建 OTA 升级数据包
String payload = JSON.toJSONString(message.getParams());
short mid = generateMessageId();
Buffer buffer = TcpDataEncoder.createDataDownPackage(
client.getDeviceAddr(), mid, payload);
// 2. 发送消息
client.sendMessage(buffer);
log.debug("[handleOtaUpgrade][OTA升级消息已发送] 设备地址: {}, 消息序号: {}",
client.getDeviceAddr(), mid);
} catch (Exception e) {
log.error("[handleOtaUpgrade][OTA升级失败]", e);
}
}
/**
* 处理自定义下行消息
*
* @param client 设备客户端
* @param message 设备消息
* @param code 功能码
*/
private void handleCustomMessage(TcpDeviceClient client, IotDeviceMessage message, short code) {
try {
log.info("[handleCustomMessage][自定义消息] 设备地址: {}, 功能码: {}, 数据: {}",
client.getDeviceAddr(), code, message.getParams());
// 1. 构建自定义数据包
String payload = JSON.toJSONString(message.getParams());
short mid = generateMessageId();
TcpDataPackage dataPackage = TcpDataPackage.builder()
.addr(client.getDeviceAddr())
.code(code)
.mid(mid)
.payload(payload)
.build();
Buffer buffer = TcpDataEncoder.encode(dataPackage);
// 2. 发送消息
client.sendMessage(buffer);
log.debug("[handleCustomMessage][自定义消息已发送] 设备地址: {}, 功能码: {}, 消息序号: {}",
client.getDeviceAddr(), code, mid);
} catch (Exception e) {
log.error("[handleCustomMessage][自定义消息发送失败]", e);
}
}
// TODO @haohao：用不到的，要不暂时不提供；
/**
* 批量发送下行消息
*
* @param deviceIds 设备ID列表
* @param message 设备消息
*/
public void broadcastMessage(Long[] deviceIds, IotDeviceMessage message) {
try {
log.info("[broadcastMessage][批量发送消息] 设备数量: {}, 方法: {}",
deviceIds.length, message.getMethod());
for (Long deviceId : deviceIds) {
// 创建副本消息（避免消息ID冲突）
IotDeviceMessage copyMessage = IotDeviceMessage.of(
message.getRequestId(),
message.getMethod(),
message.getParams(),
message.getData(),
message.getCode(),
message.getMsg());
copyMessage.setDeviceId(deviceId);
// 处理单个设备消息
handle(copyMessage);
}
} catch (Exception e) {
log.error("[broadcastMessage][批量发送消息失败]", e);
}
}
/**
* 检查设备是否支持指定方法
*
* @param client 设备客户端
* @param method 消息方法
* @return 是否支持
*/
private boolean isMethodSupported(TcpDeviceClient client, String method) {
// TODO: 可以根据设备类型或产品信息判断是否支持特定方法
return IotDeviceMessageMethodEnum.of(method) != null;
}
/**
* 生成消息序号
*
* @return 消息序号
*/
private short generateMessageId() {
return (short) (System.currentTimeMillis() % Short.MAX_VALUE);
}
/**
* 使用编解码器发送消息
*
* @param client 设备客户端
* @param message 设备消息
* @param methodName 方法名称
* @param fallbackAction 降级处理逻辑
*/
private void sendMessageWithCodec(TcpDeviceClient client, IotDeviceMessage message,
String methodName, Runnable fallbackAction) {
try {
// 1. 使用编解码器编码消息
byte[] messageBytes = messageService.encodeDeviceMessage(
message, client.getProductKey(), client.getDeviceName());
// 2. 解析编码后的数据包并设置设备地址和消息序号
Buffer buffer = Buffer.buffer(messageBytes);
TcpDataPackage dataPackage = TcpDataDecoder.decode(buffer);
dataPackage.setAddr(client.getDeviceAddr());
dataPackage.setMid(generateMessageId());
// 3. 重新编码并发送
Buffer finalBuffer = TcpDataEncoder.encode(dataPackage);
client.sendMessage(finalBuffer);
log.debug("[{}][消息已发送] 设备地址: {}, 消息序号: {}",
methodName, client.getDeviceAddr(), dataPackage.getMid());
} catch (Exception e) {
log.warn("[{}][使用编解码器编码失败，降级使用原始编码] 错误: {}",
methodName, e.getMessage());
// 执行降级处理
if (fallbackAction != null) {
fallbackAction.run();
}
}
}
// TODO @haohao：看看这个要不要删除掉
/**
* 获取连接统计信息
*
* @return 连接统计信息
*/
public String getHandlerStatistics() {
return String.format("TCP下游处理器 - %s", connectionManager.getConnectionStatus());
}
}

389
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/router/IotTcpUpstreamHandler.java
View File

@@ -1,385 +1,110 @@
package cn.iocoder.yudao.module.iot.gateway.protocol.tcp.router;
import cn.hutool.core.util.IdUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import cn.iocoder.yudao.module.iot.core.biz.IotDeviceCommonApi;
import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceRespDTO;
import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
import cn.iocoder.yudao.module.iot.gateway.config.IotGatewayProperties;
import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.client.TcpDeviceClient;
import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.manager.TcpDeviceConnectionManager;
import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.protocol.TcpDataDecoder;
import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.protocol.TcpDataEncoder;
import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.protocol.TcpDataPackage;
import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.protocol.TcpDataReader;
import cn.iocoder.yudao.module.iot.gateway.service.device.IotDeviceService;
import cn.iocoder.yudao.module.iot.gateway.codec.tcp.IotTcpCodecManager;
import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.IotTcpUpstreamProtocol;
import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
import io.vertx.core.Handler;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.net.NetSocket;
import io.vertx.core.parsetools.RecordParser;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
/**
* IoT 网关 TCP 上行消息处理器
* <p>
* 核心负责：
* 1. 【设备注册】设备连接后发送注册消息，注册成功后可以进行通信
* 2. 【心跳处理】定期接收设备心跳消息，维持连接状态
* 3. 【数据上报】接收设备数据上报和事件上报
* 4. 【连接管理】管理连接的建立、维护和清理
*
* @author 芋道源码
*/
@RequiredArgsConstructor
@Slf4j
public class IotTcpUpstreamHandler implements Handler<NetSocket> {
private final IotGatewayProperties.TcpProperties tcpConfig;
// TODO @haohao：可以把 TcpDeviceConnectionManager 能力放大一点：1）handle 里的 client 初始化，可以拿到 TcpDeviceConnectionManager 里；2）handleDeviceRegister 也是；
private final TcpDeviceConnectionManager connectionManager;
private final IotDeviceService deviceService;
private final IotDeviceMessageService messageService;
private final IotDeviceCommonApi deviceApi;
private final IotDeviceMessageService deviceMessageService;
private final String serverId;
private final IotTcpCodecManager codecManager;
public IotTcpUpstreamHandler(IotTcpUpstreamProtocol protocol, IotDeviceMessageService deviceMessageService,
IotTcpCodecManager codecManager) {
this.deviceMessageService = deviceMessageService;
this.serverId = protocol.getServerId();
this.codecManager = codecManager;
}
@Override
public void handle(NetSocket socket) {
log.info("[handle][收到设备连接: {}]", socket.remoteAddress());
// 生成客户端ID用于日志标识
String clientId = IdUtil.simpleUUID();
log.info("[handle][收到设备连接] clientId: {}, address: {}", clientId, socket.remoteAddress());
// 创建客户端 ID 和设备客户端
// TODO @haohao：clientid 给 TcpDeviceClient 生成会简洁一点；减少 upsteramhanlder 的非核心逻辑；
String clientId = IdUtil.simpleUUID() + "_" + socket.remoteAddress();
TcpDeviceClient client = new TcpDeviceClient(clientId, tcpConfig.getKeepAliveTimeoutMs());
// 设置解析器
RecordParser parser = RecordParser.newFixed(1024, buffer -> {
try {
handleDataPackage(clientId, buffer);
} catch (Exception e) {
log.error("[handle][处理数据包异常] clientId: {}", clientId, e);
}
});
try {
// 设置连接异常和关闭处理
socket.exceptionHandler(ex -> {
// TODO @haohao：这里的日志，可能把 clientid 都打上？因为 address 会重复么？
log.error("[handle][连接({})异常]", socket.remoteAddress(), ex);
handleConnectionClose(client);
});
socket.closeHandler(v -> {
log.info("[handle][连接({})关闭]", socket.remoteAddress());
handleConnectionClose(client);
});
client.setSocket(socket);
// 设置异常处理
socket.exceptionHandler(ex -> {
log.error("[handle][连接异常] clientId: {}, address: {}", clientId, socket.remoteAddress(), ex);
});
// 设置解析器
RecordParser parser = TcpDataReader.createParser(buffer -> {
try {
handleDataPackage(client, buffer);
} catch (Exception e) {
log.error("[handle][处理数据包异常]", e);
}
});
client.setParser(parser);
socket.closeHandler(v -> {
log.info("[handle][连接关闭] clientId: {}, address: {}", clientId, socket.remoteAddress());
});
// TODO @haohao：socket.remoteAddress()) 打印进去
log.info("[handle][设备连接处理器初始化完成: {}]", clientId);
} catch (Exception e) {
// TODO @haohao：socket.remoteAddress()) 打印进去
log.error("[handle][初始化连接处理器失败]", e);
client.shutdown();
}
// 设置数据处理器
socket.handler(parser);
}
/**
* 处理数据包
*
* @param client 设备客户端
* @param buffer 数据缓冲区
*/
private void handleDataPackage(TcpDeviceClient client, io.vertx.core.buffer.Buffer buffer) {
private void handleDataPackage(String clientId, Buffer buffer) {
try {
// 解码数据包
TcpDataPackage dataPackage = TcpDataDecoder.decode(buffer);
log.info("[handleDataPackage][接收数据包] 设备地址: {}, 功能码: {}, 消息序号: {}",
dataPackage.getAddr(), dataPackage.getCodeDescription(), dataPackage.getMid());
// 使用编解码器管理器自动检测协议并解码消息
IotDeviceMessage message = codecManager.decode(buffer.getBytes());
log.info("[handleDataPackage][接收数据包] clientId: {}, 方法: {}, 设备ID: {}",
clientId, message.getMethod(), message.getDeviceId());
// 根据功能码处理不同类型的消息
switch (dataPackage.getCode()) {
// TODO @haohao：【重要】code 要不要改成 opCode。这样和 data 里的 code 好区分；
case TcpDataPackage.CODE_REGISTER:
handleDeviceRegister(client, dataPackage);
break;
case TcpDataPackage.CODE_HEARTBEAT:
handleHeartbeat(client, dataPackage);
break;
case TcpDataPackage.CODE_DATA_UP:
handleDataUp(client, dataPackage);
break;
case TcpDataPackage.CODE_EVENT_UP:
handleEventUp(client, dataPackage);
break;
default:
log.warn("[handleDataPackage][未知功能码: {}]", dataPackage.getCode());
break;
}
// 处理上行消息
handleUpstreamMessage(clientId, message);
} catch (Exception e) {
// TODO @haohao：最好有 client 标识；
log.error("[handleDataPackage][处理数据包失败]", e);
log.error("[handleDataPackage][处理数据包失败] clientId: {}", clientId, e);
}
}
/**
* 处理设备注册
*
* @param client 设备客户端
* @param dataPackage 数据包
* 处理上行消息
*/
private void handleDeviceRegister(TcpDeviceClient client, TcpDataPackage dataPackage) {
private void handleUpstreamMessage(String clientId, IotDeviceMessage message) {
try {
String deviceAddr = dataPackage.getAddr();
String productKey = dataPackage.getPayload();
log.info("[handleDeviceRegister][设备注册] 设备地址: {}, 产品密钥: {}", deviceAddr, productKey);
log.info("[handleUpstreamMessage][上行消息] clientId: {}, 方法: {}, 设备ID: {}",
clientId, message.getMethod(), message.getDeviceId());
// 获取设备信息
IotDeviceRespDTO device = deviceService.getDeviceFromCache(productKey, deviceAddr);
if (device == null) {
log.error("[handleDeviceRegister][设备不存在: {} - {}]", productKey, deviceAddr);
sendRegisterReply(client, dataPackage, false);
return;
}
// 解析设备信息（简化处理）
String deviceId = String.valueOf(message.getDeviceId());
String productKey = extractProductKey(deviceId);
String deviceName = deviceId;
// 更新客户端信息
// TODO @haohao：一个 set 方法，统一处理掉会好点哈；
client.setProductKey(productKey);
client.setDeviceName(deviceAddr);
client.setDeviceId(device.getId());
client.setAuthenticated(true);
// 添加到连接管理器
connectionManager.addClient(deviceAddr, client);
connectionManager.setDeviceIdMapping(deviceAddr, device.getId());
// 发送设备上线消息
IotDeviceMessage onlineMessage = IotDeviceMessage.buildStateUpdateOnline();
messageService.sendDeviceMessage(onlineMessage, productKey, deviceAddr, serverId);
// 发送注册成功回复
sendRegisterReply(client, dataPackage, true);
log.info("[handleDeviceRegister][设备注册成功] 设备地址: {}, 设备ID: {}", deviceAddr, device.getId());
// 发送消息到队列
deviceMessageService.sendDeviceMessage(message, productKey, deviceName, serverId);
} catch (Exception e) {
log.error("[handleDeviceRegister][设备注册失败]", e);
sendRegisterReply(client, dataPackage, false);
log.error("[handleUpstreamMessage][处理上行消息失败] clientId: {}", clientId, e);
}
}
/**
* 处理心跳
*
* @param client 设备客户端
* @param dataPackage 数据包
* 从设备ID中提取产品密钥（简化实现）
*/
private void handleHeartbeat(TcpDeviceClient client, TcpDataPackage dataPackage) {
try {
String deviceAddr = dataPackage.getAddr();
log.debug("[handleHeartbeat][收到心跳] 设备地址: {}", deviceAddr);
// 更新心跳时间
client.keepAlive();
// 发送心跳回复（可选）
// sendHeartbeatReply(client, dataPackage);
} catch (Exception e) {
log.error("[handleHeartbeat][处理心跳失败]", e);
}
}
/**
* 处理数据上报
*
* @param client 设备客户端
* @param dataPackage 数据包
*/
private void handleDataUp(TcpDeviceClient client, TcpDataPackage dataPackage) {
try {
String deviceAddr = dataPackage.getAddr();
String payload = dataPackage.getPayload();
log.info("[handleDataUp][数据上报] 设备地址: {}, 数据: {}", deviceAddr, payload);
// 检查设备是否已认证
if (!client.isAuthenticated()) {
log.warn("[handleDataUp][设备未认证，忽略数据上报: {}]", deviceAddr);
return;
}
// 使用 IotDeviceMessageService 解码消息
try {
// 1. 将 TCP 数据包重新编码为字节数组
Buffer buffer = TcpDataEncoder.encode(dataPackage);
byte[] messageBytes = buffer.getBytes();
// 2. 使用 messageService 解码消息
IotDeviceMessage message = messageService.decodeDeviceMessage(
messageBytes, client.getProductKey(), client.getDeviceName());
// 3. 发送解码后的消息
messageService.sendDeviceMessage(message, client.getProductKey(), client.getDeviceName(), serverId);
} catch (Exception e) {
log.warn("[handleDataUp][使用编解码器解码失败，降级使用原始解析] 错误: {}", e.getMessage());
// 降级处理：使用原始方式解析数据
JSONObject dataJson = JSONUtil.parseObj(payload);
IotDeviceMessage message = IotDeviceMessage.requestOf("thing.property.post", dataJson);
messageService.sendDeviceMessage(message, client.getProductKey(), client.getDeviceName(), serverId);
}
// 发送数据上报回复
sendDataUpReply(client, dataPackage);
} catch (Exception e) {
log.error("[handleDataUp][处理数据上报失败]", e);
}
}
/**
* 处理事件上报
*
* @param client 设备客户端
* @param dataPackage 数据包
*/
private void handleEventUp(TcpDeviceClient client, TcpDataPackage dataPackage) {
try {
String deviceAddr = dataPackage.getAddr();
String payload = dataPackage.getPayload();
log.info("[handleEventUp][事件上报] 设备地址: {}, 数据: {}", deviceAddr, payload);
// 检查设备是否已认证
if (!client.isAuthenticated()) {
log.warn("[handleEventUp][设备未认证，忽略事件上报: {}]", deviceAddr);
return;
}
// 使用 IotDeviceMessageService 解码消息
try {
// 1. 将 TCP 数据包重新编码为字节数组
Buffer buffer = TcpDataEncoder.encode(dataPackage);
byte[] messageBytes = buffer.getBytes();
// 2. 使用 messageService 解码消息
IotDeviceMessage message = messageService.decodeDeviceMessage(
messageBytes, client.getProductKey(), client.getDeviceName());
// 3. 发送解码后的消息
messageService.sendDeviceMessage(message, client.getProductKey(), client.getDeviceName(), serverId);
} catch (Exception e) {
log.warn("[handleEventUp][使用编解码器解码失败，降级使用原始解析] 错误: {}", e.getMessage());
// 降级处理：使用原始方式解析数据
// TODO @芋艿：降级处理逻辑；
JSONObject eventJson = JSONUtil.parseObj(payload);
IotDeviceMessage message = IotDeviceMessage.requestOf("thing.event.post", eventJson);
messageService.sendDeviceMessage(message, client.getProductKey(), client.getDeviceName(), serverId);
}
// 发送事件上报回复
sendEventUpReply(client, dataPackage);
} catch (Exception e) {
log.error("[handleEventUp][处理事件上报失败]", e);
}
}
/**
* 发送注册回复
*
* @param client 设备客户端
* @param dataPackage 原始数据包
* @param success 是否成功
*/
private void sendRegisterReply(TcpDeviceClient client, TcpDataPackage dataPackage, boolean success) {
try {
io.vertx.core.buffer.Buffer replyBuffer = TcpDataEncoder.createRegisterReply(
dataPackage.getAddr(), dataPackage.getMid(), success);
client.sendMessage(replyBuffer);
log.debug("[sendRegisterReply][发送注册回复] 设备地址: {}, 结果: {}",
dataPackage.getAddr(), success ? "成功" : "失败");
} catch (Exception e) {
log.error("[sendRegisterReply][发送注册回复失败]", e);
}
}
/**
* 发送数据上报回复
*
* @param client 设备客户端
* @param dataPackage 原始数据包
*/
private void sendDataUpReply(TcpDeviceClient client, TcpDataPackage dataPackage) {
try {
TcpDataPackage replyPackage = TcpDataPackage.builder()
.addr(dataPackage.getAddr())
.code(TcpDataPackage.CODE_DATA_UP)
.mid(dataPackage.getMid())
.payload("0") // 0 表示成功 TODO @haohao：最好枚举到 TcpDataPackage 里？
.build();
io.vertx.core.buffer.Buffer replyBuffer = TcpDataEncoder.encode(replyPackage);
client.sendMessage(replyBuffer);
} catch (Exception e) {
// TODO @haohao：可以有个 client id
log.error("[sendDataUpReply][发送数据上报回复失败]", e);
}
}
/**
* 发送事件上报回复
*
* @param client 设备客户端
* @param dataPackage 原始数据包
*/
private void sendEventUpReply(TcpDeviceClient client, TcpDataPackage dataPackage) {
try {
TcpDataPackage replyPackage = TcpDataPackage.builder()
.addr(dataPackage.getAddr())
.code(TcpDataPackage.CODE_EVENT_UP)
.mid(dataPackage.getMid())
.payload("0") // 0 表示成功
.build();
io.vertx.core.buffer.Buffer replyBuffer = TcpDataEncoder.encode(replyPackage);
client.sendMessage(replyBuffer);
} catch (Exception e) {
log.error("[sendEventUpReply][发送事件上报回复失败]", e);
}
}
/**
* 处理连接关闭
*
* @param client 设备客户端
*/
private void handleConnectionClose(TcpDeviceClient client) {
try {
String deviceAddr = client.getDeviceAddr();
// 发送设备离线消息
if (client.isAuthenticated()) {
IotDeviceMessage offlineMessage = IotDeviceMessage.buildStateOffline();
messageService.sendDeviceMessage(offlineMessage,
client.getProductKey(), client.getDeviceName(), serverId);
}
// 从连接管理器移除
if (deviceAddr != null) {
connectionManager.removeClient(deviceAddr);
}
log.info("[handleConnectionClose][处理连接关闭完成] 设备地址: {}", deviceAddr);
} catch (Exception e) {
log.error("[handleConnectionClose][处理连接关闭失败]", e);
private String extractProductKey(String deviceId) {
// 简化实现：假设设备ID格式为 "productKey_deviceName"
if (deviceId != null && deviceId.contains("_")) {
return deviceId.split("_")[0];
}
return "default_product";
}
}

219
yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/codec/tcp/TcpBinaryDataPacketExamples.java Normal file
View File

@@ -0,0 +1,219 @@
package cn.iocoder.yudao.module.iot.gateway.codec.tcp;
import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
import lombok.extern.slf4j.Slf4j;
import java.util.HashMap;
import java.util.Map;
/**
* TCP二进制格式数据包示例
*
* 演示如何使用二进制协议创建和解析TCP上报数据包和心跳包
*
* 二进制协议格式：
* 包头(4字节) | 地址长度(2字节) | 设备地址(变长) | 功能码(2字节) | 消息序号(2字节) | 包体数据(变长)
*
* @author 芋道源码
*/
@Slf4j
public class TcpBinaryDataPacketExamples {
public static void main(String[] args) {
IotTcpBinaryDeviceMessageCodec codec = new IotTcpBinaryDeviceMessageCodec();
// 1. 数据上报包示例
demonstrateDataReport(codec);
// 2. 心跳包示例
demonstrateHeartbeat(codec);
// 3. 复杂数据上报示例
demonstrateComplexDataReport(codec);
}
/**
* 演示二进制格式数据上报包
*/
private static void demonstrateDataReport(IotTcpBinaryDeviceMessageCodec codec) {
log.info("=== 二进制格式数据上报包示例 ===");
// 创建传感器数据
Map<String, Object> sensorData = new HashMap<>();
sensorData.put("temperature", 25.5);
sensorData.put("humidity", 60.2);
sensorData.put("pressure", 1013.25);
sensorData.put("battery", 85);
// 创建设备消息
IotDeviceMessage message = IotDeviceMessage.requestOf("thing.property.post", sensorData);
message.setDeviceId(123456L);
// 编码
byte[] packet = codec.encode(message);
log.info("编码后数据包长度: {} 字节", packet.length);
log.info("编码后数据包(HEX): {}", bytesToHex(packet));
// 解码验证
IotDeviceMessage decoded = codec.decode(packet);
log.info("解码后消息ID: {}", decoded.getId());
log.info("解码后请求ID: {}", decoded.getRequestId());
log.info("解码后方法: {}", decoded.getMethod());
log.info("解码后设备ID: {}", decoded.getDeviceId());
log.info("解码后服务ID: {}", decoded.getServerId());
log.info("解码后上报时间: {}", decoded.getReportTime());
log.info("解码后参数: {}", decoded.getParams());
System.out.println();
}
/**
* 演示二进制格式心跳包
*/
private static void demonstrateHeartbeat(IotTcpBinaryDeviceMessageCodec codec) {
log.info("=== 二进制格式心跳包示例 ===");
// 创建心跳消息
IotDeviceMessage heartbeat = IotDeviceMessage.requestOf("thing.state.online", null);
heartbeat.setDeviceId(123456L);
// 编码
byte[] packet = codec.encode(heartbeat);
log.info("心跳包长度: {} 字节", packet.length);
log.info("心跳包(HEX): {}", bytesToHex(packet));
// 解码验证
IotDeviceMessage decoded = codec.decode(packet);
log.info("解码后消息ID: {}", decoded.getId());
log.info("解码后请求ID: {}", decoded.getRequestId());
log.info("解码后方法: {}", decoded.getMethod());
log.info("解码后设备ID: {}", decoded.getDeviceId());
log.info("解码后服务ID: {}", decoded.getServerId());
log.info("解码后参数: {}", decoded.getParams());
System.out.println();
}
/**
* 演示二进制格式复杂数据上报
*/
private static void demonstrateComplexDataReport(IotTcpBinaryDeviceMessageCodec codec) {
log.info("=== 二进制格式复杂数据上报示例 ===");
// 创建复杂设备数据
Map<String, Object> deviceData = new HashMap<>();
// 环境数据
Map<String, Object> environment = new HashMap<>();
environment.put("temperature", 23.8);
environment.put("humidity", 55.0);
environment.put("co2", 420);
deviceData.put("environment", environment);
// GPS数据
Map<String, Object> location = new HashMap<>();
location.put("latitude", 39.9042);
location.put("longitude", 116.4074);
location.put("altitude", 43.5);
deviceData.put("location", location);
// 设备状态
Map<String, Object> status = new HashMap<>();
status.put("battery", 78);
status.put("signal", -65);
status.put("online", true);
deviceData.put("status", status);
// 创建设备消息
IotDeviceMessage message = IotDeviceMessage.requestOf("thing.property.post", deviceData);
message.setDeviceId(789012L);
// 编码
byte[] packet = codec.encode(message);
log.info("复杂数据包长度: {} 字节", packet.length);
log.info("复杂数据包(HEX): {}", bytesToHex(packet));
// 解码验证
IotDeviceMessage decoded = codec.decode(packet);
log.info("解码后消息ID: {}", decoded.getId());
log.info("解码后请求ID: {}", decoded.getRequestId());
log.info("解码后方法: {}", decoded.getMethod());
log.info("解码后设备ID: {}", decoded.getDeviceId());
log.info("解码后服务ID: {}", decoded.getServerId());
log.info("解码后参数: {}", decoded.getParams());
System.out.println();
}
/**
* 字节数组转十六进制字符串
*/
private static String bytesToHex(byte[] bytes) {
StringBuilder result = new StringBuilder();
for (byte b : bytes) {
result.append(String.format("%02X ", b));
}
return result.toString().trim();
}
/**
* 演示数据包结构分析
*/
public static void analyzePacketStructure(byte[] packet) {
if (packet.length < 8) {
log.error("数据包长度不足");
return;
}
int index = 0;
// 解析包头(4字节) - 后续数据长度
int totalLength = ((packet[index] & 0xFF) << 24) |
((packet[index + 1] & 0xFF) << 16) |
((packet[index + 2] & 0xFF) << 8) |
(packet[index + 3] & 0xFF);
index += 4;
log.info("包头 - 后续数据长度: {} 字节", totalLength);
// 解析设备地址长度(2字节)
int addrLength = ((packet[index] & 0xFF) << 8) | (packet[index + 1] & 0xFF);
index += 2;
log.info("设备地址长度: {} 字节", addrLength);
// 解析设备地址
String deviceAddr = new String(packet, index, addrLength);
index += addrLength;
log.info("设备地址: {}", deviceAddr);
// 解析功能码(2字节)
int functionCode = ((packet[index] & 0xFF) << 8) | (packet[index + 1] & 0xFF);
index += 2;
log.info("功能码: {} ({})", functionCode, getFunctionCodeName(functionCode));
// 解析消息序号(2字节)
int messageId = ((packet[index] & 0xFF) << 8) | (packet[index + 1] & 0xFF);
index += 2;
log.info("消息序号: {}", messageId);
// 解析包体数据
if (index < packet.length) {
String payload = new String(packet, index, packet.length - index);
log.info("包体数据: {}", payload);
}
}
/**
* 获取功能码名称
*/
private static String getFunctionCodeName(int code) {
switch (code) {
case 10: return "设备注册";
case 11: return "注册回复";
case 20: return "心跳请求";
case 21: return "心跳回复";
case 30: return "消息上行";
case 40: return "消息下行";
default: return "未知功能码";
}
}
}

253
yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/codec/tcp/TcpJsonDataPacketExamples.java Normal file
View File

@@ -0,0 +1,253 @@
package cn.iocoder.yudao.module.iot.gateway.codec.tcp;
import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
import lombok.extern.slf4j.Slf4j;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
/**
* TCP JSON格式数据包示例
*
* 演示如何使用新的JSON格式进行TCP消息编解码
*
* @author 芋道源码
*/
@Slf4j
public class TcpJsonDataPacketExamples {
public static void main(String[] args) {
IotTcpJsonDeviceMessageCodec codec = new IotTcpJsonDeviceMessageCodec();
// 1. 数据上报示例
demonstrateDataReport(codec);
// 2. 心跳示例
demonstrateHeartbeat(codec);
// 3. 事件上报示例
demonstrateEventReport(codec);
// 4. 复杂数据上报示例
demonstrateComplexDataReport(codec);
// 5. 便捷方法示例
demonstrateConvenienceMethods();
// 6. EMQX兼容性示例
demonstrateEmqxCompatibility();
}
/**
* 演示数据上报
*/
private static void demonstrateDataReport(IotTcpJsonDeviceMessageCodec codec) {
log.info("=== JSON格式数据上报示例 ===");
// 创建传感器数据
Map<String, Object> sensorData = new HashMap<>();
sensorData.put("temperature", 25.5);
sensorData.put("humidity", 60.2);
sensorData.put("pressure", 1013.25);
sensorData.put("battery", 85);
// 创建设备消息
IotDeviceMessage message = IotDeviceMessage.requestOf("thing.property.post", sensorData);
message.setDeviceId(123456L);
// 编码
byte[] packet = codec.encode(message);
String jsonString = new String(packet, StandardCharsets.UTF_8);
log.info("编码后JSON: {}", jsonString);
log.info("数据包长度: {} 字节", packet.length);
// 解码验证
IotDeviceMessage decoded = codec.decode(packet);
log.info("解码后消息ID: {}", decoded.getId());
log.info("解码后方法: {}", decoded.getMethod());
log.info("解码后设备ID: {}", decoded.getDeviceId());
log.info("解码后服务ID: {}", decoded.getServerId());
log.info("解码后参数: {}", decoded.getParams());
System.out.println();
}
/**
* 演示心跳
*/
private static void demonstrateHeartbeat(IotTcpJsonDeviceMessageCodec codec) {
log.info("=== JSON格式心跳示例 ===");
// 创建心跳消息
IotDeviceMessage heartbeat = IotDeviceMessage.requestOf("thing.state.online", null);
heartbeat.setDeviceId(123456L);
// 编码
byte[] packet = codec.encode(heartbeat);
String jsonString = new String(packet, StandardCharsets.UTF_8);
log.info("编码后JSON: {}", jsonString);
log.info("心跳包长度: {} 字节", packet.length);
// 解码验证
IotDeviceMessage decoded = codec.decode(packet);
log.info("解码后消息ID: {}", decoded.getId());
log.info("解码后方法: {}", decoded.getMethod());
log.info("解码后设备ID: {}", decoded.getDeviceId());
log.info("解码后服务ID: {}", decoded.getServerId());
System.out.println();
}
/**
* 演示事件上报
*/
private static void demonstrateEventReport(IotTcpJsonDeviceMessageCodec codec) {
log.info("=== JSON格式事件上报示例 ===");
// 创建事件数据
Map<String, Object> eventData = new HashMap<>();
eventData.put("eventType", "alarm");
eventData.put("level", "warning");
eventData.put("description", "温度过高");
eventData.put("value", 45.8);
// 创建事件消息
IotDeviceMessage event = IotDeviceMessage.requestOf("thing.event.post", eventData);
event.setDeviceId(123456L);
// 编码
byte[] packet = codec.encode(event);
String jsonString = new String(packet, StandardCharsets.UTF_8);
log.info("编码后JSON: {}", jsonString);
log.info("事件包长度: {} 字节", packet.length);
// 解码验证
IotDeviceMessage decoded = codec.decode(packet);
log.info("解码后消息ID: {}", decoded.getId());
log.info("解码后方法: {}", decoded.getMethod());
log.info("解码后设备ID: {}", decoded.getDeviceId());
log.info("解码后参数: {}", decoded.getParams());
System.out.println();
}
/**
* 演示复杂数据上报
*/
private static void demonstrateComplexDataReport(IotTcpJsonDeviceMessageCodec codec) {
log.info("=== JSON格式复杂数据上报示例 ===");
// 创建复杂设备数据（类似EMQX格式）
Map<String, Object> deviceData = new HashMap<>();
// 环境数据
Map<String, Object> environment = new HashMap<>();
environment.put("temperature", 23.8);
environment.put("humidity", 55.0);
environment.put("co2", 420);
environment.put("pm25", 35);
deviceData.put("environment", environment);
// GPS数据
Map<String, Object> location = new HashMap<>();
location.put("latitude", 39.9042);
location.put("longitude", 116.4074);
location.put("altitude", 43.5);
location.put("speed", 0.0);
deviceData.put("location", location);
// 设备状态
Map<String, Object> status = new HashMap<>();
status.put("battery", 78);
status.put("signal", -65);
status.put("online", true);
status.put("version", "1.2.3");
deviceData.put("status", status);
// 创建设备消息
IotDeviceMessage message = IotDeviceMessage.requestOf("thing.property.post", deviceData);
message.setDeviceId(789012L);
// 编码
byte[] packet = codec.encode(message);
String jsonString = new String(packet, StandardCharsets.UTF_8);
log.info("编码后JSON: {}", jsonString);
log.info("复杂数据包长度: {} 字节", packet.length);
// 解码验证
IotDeviceMessage decoded = codec.decode(packet);
log.info("解码后消息ID: {}", decoded.getId());
log.info("解码后方法: {}", decoded.getMethod());
log.info("解码后设备ID: {}", decoded.getDeviceId());
log.info("解码后参数: {}", decoded.getParams());
System.out.println();
}
/**
* 演示便捷方法
*/
private static void demonstrateConvenienceMethods() {
log.info("=== 便捷方法示例 ===");
IotTcpJsonDeviceMessageCodec codec = new IotTcpJsonDeviceMessageCodec();
// 使用便捷方法编码数据上报
Map<String, Object> sensorData = Map.of(
"temperature", 26.5,
"humidity", 58.3
);
byte[] dataPacket = codec.encodeDataReport(sensorData, 123456L, "smart_sensor", "device_001");
log.info("便捷方法编码数据上报: {}", new String(dataPacket, StandardCharsets.UTF_8));
// 使用便捷方法编码心跳
byte[] heartbeatPacket = codec.encodeHeartbeat(123456L, "smart_sensor", "device_001");
log.info("便捷方法编码心跳: {}", new String(heartbeatPacket, StandardCharsets.UTF_8));
// 使用便捷方法编码事件
Map<String, Object> eventData = Map.of(
"eventType", "maintenance",
"description", "定期维护提醒"
);
byte[] eventPacket = codec.encodeEventReport(eventData, 123456L, "smart_sensor", "device_001");
log.info("便捷方法编码事件: {}", new String(eventPacket, StandardCharsets.UTF_8));
System.out.println();
}
/**
* 演示与EMQX格式的兼容性
*/
private static void demonstrateEmqxCompatibility() {
log.info("=== EMQX格式兼容性示例 ===");
// 模拟EMQX风格的消息格式
String emqxStyleJson = """
{
"id": "msg_001",
"method": "thing.property.post",
"deviceId": 123456,
"params": {
"temperature": 25.5,
"humidity": 60.2
},
"timestamp": 1642781234567
}
""";
IotTcpJsonDeviceMessageCodec codec = new IotTcpJsonDeviceMessageCodec();
// 解码EMQX风格的消息
byte[] emqxBytes = emqxStyleJson.getBytes(StandardCharsets.UTF_8);
IotDeviceMessage decoded = codec.decode(emqxBytes);
log.info("EMQX风格消息解码成功:");
log.info("消息ID: {}", decoded.getId());
log.info("方法: {}", decoded.getMethod());
log.info("设备ID: {}", decoded.getDeviceId());
log.info("参数: {}", decoded.getParams());
System.out.println();
}
}

222
yudao-module-iot/yudao-module-iot-gateway/src/test/resources/tcp-binary-packet-examples.md Normal file
View File

@@ -0,0 +1,222 @@
# TCP二进制协议数据包格式说明和示例
## 1. 二进制协议概述
TCP二进制协议是一种高效的自定义协议格式，适用于对带宽和性能要求较高的场景。
## 2. 数据包格式
### 2.1 整体结构
```
+----------+----------+----------+----------+----------+----------+
| 包头 | 地址长度 | 设备地址 | 功能码 | 消息序号 | 包体数据 |
| 4字节 | 2字节 | 变长 | 2字节 | 2字节 | 变长 |
+----------+----------+----------+----------+----------+----------+
```
### 2.2 字段说明
| 字段 | 长度 | 类型 | 说明 |
|----------|--------|--------|--------------------------------|
| 包头 | 4字节 | int | 后续数据的总长度（不包含包头） |
| 地址长度 | 2字节 | short | 设备地址的字节长度 |
| 设备地址 | 变长 | string | 设备标识符 |
| 功能码 | 2字节 | short | 消息类型标识 |
| 消息序号 | 2字节 | short | 消息唯一标识 |
| 包体数据 | 变长 | string | JSON格式的消息内容 |
### 2.3 功能码定义
| 功能码 | 名称 | 说明 |
|--------|----------|--------------------------------|
| 10 | 设备注册 | 设备首次连接时的注册请求 |
| 11 | 注册回复 | 服务器对注册请求的回复 |
| 20 | 心跳请求 | 设备发送的心跳包 |
| 21 | 心跳回复 | 服务器对心跳的回复 |
| 30 | 消息上行 | 设备向服务器发送的数据 |
| 40 | 消息下行 | 服务器向设备发送的指令 |
## 3. 二进制数据上报包示例
### 3.1 温度传感器数据上报
**原始数据：**
```json
{
"method": "thing.property.post",
"params": {
"temperature": 25.5,
"humidity": 60.2,
"pressure": 1013.25
},
"timestamp": 1642781234567
}
```
**数据包结构：**
```
包头: 0x00000045 (69字节)
地址长度: 0x0006 (6字节)
设备地址: "123456"
功能码: 0x001E (30 - 消息上行)
消息序号: 0x1234 (4660)
包体: JSON字符串
```
**完整十六进制数据包：**
```
00 00 00 45 00 06 31 32 33 34 35 36 00 1E 12 34
7B 22 6D 65 74 68 6F 64 22 3A 22 74 68 69 6E 67
2E 70 72 6F 70 65 72 74 79 2E 70 6F 73 74 22 2C
22 70 61 72 61 6D 73 22 3A 7B 22 74 65 6D 70 65
72 61 74 75 72 65 22 3A 32 35 2E 35 2C 22 68 75
6D 69 64 69 74 79 22 3A 36 30 2E 32 2C 22 70 72
65 73 73 75 72 65 22 3A 31 30 31 33 2E 32 35 7D
2C 22 74 69 6D 65 73 74 61 6D 70 22 3A 31 36 34
32 37 38 31 32 33 34 35 36 37 7D
```
### 2.2 GPS定位数据上报
**原始数据：**
```json
{
"method": "thing.property.post",
"params": {
"latitude": 39.9042,
"longitude": 116.4074,
"altitude": 43.5,
"speed": 0.0
},
"timestamp": 1642781234567
}
```
## 3. 心跳包示例
### 3.1 标准心跳包
**原始数据：**
```json
{
"method": "thing.state.online",
"timestamp": 1642781234567
}
```
**数据包结构：**
```
包头: 0x00000028 (40字节)
地址长度: 0x0006 (6字节)
设备地址: "123456"
功能码: 0x0014 (20 - 心跳请求)
消息序号: 0x5678 (22136)
包体: JSON字符串
```
**完整十六进制数据包：**
```
00 00 00 28 00 06 31 32 33 34 35 36 00 14 56 78
7B 22 6D 65 74 68 6F 64 22 3A 22 74 68 69 6E 67
2E 73 74 61 74 65 2E 6F 6E 6C 69 6E 65 22 2C 22
74 69 6D 65 73 74 61 6D 70 22 3A 31 36 34 32 37
38 31 32 33 34 35 36 37 7D
```
## 4. 复杂数据上报示例
### 4.1 多传感器综合数据
**原始数据：**
```json
{
"method": "thing.property.post",
"params": {
"environment": {
"temperature": 23.8,
"humidity": 55.0,
"co2": 420
},
"location": {
"latitude": 39.9042,
"longitude": 116.4074,
"altitude": 43.5
},
"status": {
"battery": 78,
"signal": -65,
"online": true
}
},
"timestamp": 1642781234567
}
```
## 5. 数据包解析步骤
### 5.1 解析流程
1. **读取包头（4字节）**
- 获取后续数据的总长度
- 验证数据包完整性
2. **读取设备地址长度（2字节）**
- 确定设备地址的字节数
3. **读取设备地址（变长）**
- 根据地址长度读取设备标识
4. **读取功能码（2字节）**
- 确定消息类型
5. **读取消息序号（2字节）**
- 获取消息唯一标识
6. **读取包体数据（变长）**
- 解析JSON格式的消息内容
### 5.2 Java解析示例
```java
public TcpDataPackage parsePacket(byte[] packet) {
int index = 0;
// 1. 解析包头
int totalLength = ByteBuffer.wrap(packet, index, 4).getInt();
index += 4;
// 2. 解析设备地址长度
short addrLength = ByteBuffer.wrap(packet, index, 2).getShort();
index += 2;
// 3. 解析设备地址
String deviceAddr = new String(packet, index, addrLength);
index += addrLength;
// 4. 解析功能码
short functionCode = ByteBuffer.wrap(packet, index, 2).getShort();
index += 2;
// 5. 解析消息序号
short messageId = ByteBuffer.wrap(packet, index, 2).getShort();
index += 2;
// 6. 解析包体数据
String payload = new String(packet, index, packet.length - index);
return TcpDataPackage.builder()
.addr(deviceAddr)
.code(functionCode)
.mid(messageId)
.payload(payload)
.build();
}
```
## 6. 注意事项
1. **字节序**：所有多字节数据使用大端序（Big-Endian）
2. **字符编码**：字符串数据使用UTF-8编码
3. **JSON格式**：包体数据必须是有效的JSON格式
4. **长度限制**：单个数据包建议不超过1MB
5. **错误处理**：解析失败时应返回相应的错误码

286
yudao-module-iot/yudao-module-iot-gateway/src/test/resources/tcp-json-packet-examples.md Normal file
View File

@@ -0,0 +1,286 @@
# TCP JSON格式协议说明
## 1. 协议概述
TCP JSON格式协议采用纯JSON格式进行数据传输，参考了EMQX和HTTP模块的数据格式设计，具有以下优势：
- **标准化**：使用标准JSON格式，易于解析和处理
- **可读性**：人类可读，便于调试和维护
- **扩展性**：可以轻松添加新字段，向后兼容
- **统一性**：与HTTP模块保持一致的数据格式
## 2. 消息格式
### 2.1 基础消息结构
```json
{
"id": "消息唯一标识",
"method": "消息方法",
"deviceId": "设备ID",
"params": {
// 消息参数
},
"timestamp": 时间戳
}
```
### 2.2 字段说明
| 字段名 | 类型 | 必填 | 说明 |
|--------|------|------|------|
| id | String | 是 | 消息唯一标识，UUID格式 |
| method | String | 是 | 消息方法，如 thing.property.post |
| deviceId | Long | 是 | 设备ID |
| params | Object | 否 | 消息参数，具体内容根据method而定 |
| timestamp | Long | 是 | 时间戳（毫秒） |
| code | Integer | 否 | 响应码（下行消息使用） |
| message | String | 否 | 响应消息（下行消息使用） |
## 3. 消息类型
### 3.1 数据上报 (thing.property.post)
设备向服务器上报属性数据。
**示例：**
```json
{
"id": "8ac6a1db91e64aa9996143fdbac2cbfe",
"method": "thing.property.post",
"deviceId": 123456,
"params": {
"temperature": 25.5,
"humidity": 60.2,
"pressure": 1013.25,
"battery": 85
},
"timestamp": 1753111026437
}
```
### 3.2 心跳 (thing.state.online)
设备向服务器发送心跳保活。
**示例：**
```json
{
"id": "7db8c4e6408b40f8b2549ddd94f6bb02",
"method": "thing.state.online",
"deviceId": 123456,
"timestamp": 1753111026467
}
```
### 3.3 事件上报 (thing.event.post)
设备向服务器上报事件信息。
**示例：**
```json
{
"id": "9e7d72731b854916b1baa5088bd6a907",
"method": "thing.event.post",
"deviceId": 123456,
"params": {
"eventType": "alarm",
"level": "warning",
"description": "温度过高",
"value": 45.8
},
"timestamp": 1753111026468
}
```
### 3.4 属性设置 (thing.property.set)
服务器向设备下发属性设置指令。
**示例：**
```json
{
"id": "cmd_001",
"method": "thing.property.set",
"deviceId": 123456,
"params": {
"targetTemperature": 22.0,
"mode": "auto"
},
"timestamp": 1753111026469
}
```
### 3.5 服务调用 (thing.service.invoke)
服务器向设备调用服务。
**示例：**
```json
{
"id": "service_001",
"method": "thing.service.invoke",
"deviceId": 123456,
"params": {
"service": "restart",
"args": {
"delay": 5
}
},
"timestamp": 1753111026470
}
```
## 4. 复杂数据示例
### 4.1 多传感器综合数据
```json
{
"id": "complex_001",
"method": "thing.property.post",
"deviceId": 789012,
"params": {
"environment": {
"temperature": 23.8,
"humidity": 55.0,
"co2": 420,
"pm25": 35
},
"location": {
"latitude": 39.9042,
"longitude": 116.4074,
"altitude": 43.5,
"speed": 0.0
},
"status": {
"battery": 78,
"signal": -65,
"online": true,
"version": "1.2.3"
}
},
"timestamp": 1753111026471
}
```
## 5. 与EMQX格式的兼容性
本协议设计参考了EMQX的消息格式，具有良好的兼容性：
### 5.1 EMQX标准格式
```json
{
"id": "msg_001",
"method": "thing.property.post",
"deviceId": 123456,
"params": {
"temperature": 25.5,
"humidity": 60.2
},
"timestamp": 1642781234567
}
```
### 5.2 兼容性说明
- ✅ **字段名称**：与EMQX保持一致
- ✅ **数据类型**：完全兼容
- ✅ **消息结构**：结构相同
- ✅ **扩展字段**：支持自定义扩展
## 6. 使用示例
### 6.1 Java编码示例
```java
// 创建编解码器
IotTcpJsonDeviceMessageCodec codec = new IotTcpJsonDeviceMessageCodec();
// 创建数据上报消息
Map<String, Object> sensorData = Map.of(
"temperature", 25.5,
"humidity", 60.2
);
IotDeviceMessage message = IotDeviceMessage.requestOf("thing.property.post", sensorData);
message.setDeviceId(123456L);
// 编码为字节数组
byte[] jsonBytes = codec.encode(message);
// 解码
IotDeviceMessage decoded = codec.decode(jsonBytes);
```
### 6.2 便捷方法示例
```java
// 快速编码数据上报
byte[] dataPacket = codec.encodeDataReport(sensorData, 123456L, "product_key", "device_name");
// 快速编码心跳
byte[] heartbeatPacket = codec.encodeHeartbeat(123456L, "product_key", "device_name");
// 快速编码事件
byte[] eventPacket = codec.encodeEventReport(eventData, 123456L, "product_key", "device_name");
```
## 7. 协议优势
### 7.1 与原TCP二进制协议对比
| 特性 | 二进制协议 | JSON协议 |
|------|------------|----------|
| 可读性 | 差 | 优秀 |
| 调试难度 | 高 | 低 |
| 扩展性 | 差 | 优秀 |
| 解析复杂度 | 高 | 低 |
| 数据大小 | 小 | 稍大 |
| 标准化程度 | 低 | 高 |
### 7.2 适用场景
- ✅ **开发调试**：JSON格式便于查看和调试
- ✅ **快速集成**：标准JSON格式，集成简单
- ✅ **协议扩展**：可以轻松添加新字段
- ✅ **多语言支持**：JSON格式支持所有主流语言
- ✅ **云平台对接**：与主流IoT云平台格式兼容
## 8. 最佳实践
### 8.1 消息设计建议
1. **保持简洁**：避免过深的嵌套结构
2. **字段命名**：使用驼峰命名法，保持一致性
3. **数据类型**：使用合适的数据类型，避免字符串表示数字
4. **时间戳**：统一使用毫秒级时间戳
### 8.2 性能优化
1. **批量上报**：可以在params中包含多个数据点
2. **压缩传输**：对于大数据量可以考虑gzip压缩
3. **缓存机制**：客户端可以缓存消息，批量发送
### 8.3 错误处理
1. **格式验证**：确保JSON格式正确
2. **字段检查**：验证必填字段是否存在
3. **异常处理**：提供详细的错误信息
## 9. 迁移指南
### 9.1 从二进制协议迁移
1. **保持兼容**：可以同时支持两种协议
2. **逐步迁移**：按设备类型逐步迁移
3. **测试验证**：充分测试新协议的稳定性
### 9.2 配置变更
```java
// 在设备配置中指定编解码器类型
device.setCodecType("TCP_JSON");
```
这样就完成了TCP协议向JSON格式的升级，提供了更好的可读性、扩展性和兼容性。