fix:【IoT 物联网】code review tcp 实现

This commit is contained in:
YunaiV 2025-07-23 19:17:33 +08:00
parent bd8052f56b
commit f70f578ac5
7 changed files with 133 additions and 101 deletions

View File

@ -10,11 +10,12 @@ import lombok.Data;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
// TODO @haohao设备地址(变长) 是不是非必要哈因为认证后不需要每次都带呀
/** /**
* TCP 二进制格式 {@link IotDeviceMessage} 编解码器 * TCP 二进制格式 {@link IotDeviceMessage} 编解码器
* *
* 使用自定义二进制协议格式 * 使用自定义二进制协议格式
* 包头(4字节) | 地址长度(2字节) | 设备地址(变长) | 功能码(2字节) | 消息序号(2字节) | 包体数据(变长) * 包头(4 字节) | 地址长度(2 字节) | 设备地址(变长) | 功能码(2 字节) | 消息序号(2 字节) | 包体数据(变长)
* *
* @author 芋道源码 * @author 芋道源码
*/ */
@ -27,6 +28,7 @@ public class IotTcpBinaryDeviceMessageCodec implements IotDeviceMessageCodec {
*/ */
public static final String TYPE = "TCP_BINARY"; public static final String TYPE = "TCP_BINARY";
// TODO @haohao这个注释不太对
// ==================== 常量定义 ==================== // ==================== 常量定义 ====================
@Override @Override
@ -67,10 +69,11 @@ public class IotTcpBinaryDeviceMessageCodec implements IotDeviceMessageCodec {
TcpDataPackage dataPackage = decodeTcpDataPackage(Buffer.buffer(bytes)); TcpDataPackage dataPackage = decodeTcpDataPackage(Buffer.buffer(bytes));
// 2. 根据功能码确定方法 // 2. 根据功能码确定方法
// TODO @haohao会不会有事件上报哈
String method = (dataPackage.getCode() == TcpDataPackage.CODE_HEARTBEAT) ? String method = (dataPackage.getCode() == TcpDataPackage.CODE_HEARTBEAT) ?
MessageMethod.STATE_ONLINE : MessageMethod.PROPERTY_POST; MessageMethod.STATE_ONLINE : MessageMethod.PROPERTY_POST;
// 3. 解析负载数据和请求ID // 3. 解析负载数据和请求 ID
PayloadInfo payloadInfo = parsePayloadInfo(dataPackage.getPayload()); PayloadInfo payloadInfo = parsePayloadInfo(dataPackage.getPayload());
// 4. 构建 IoT 设备消息设置完整的必要参数 // 4. 构建 IoT 设备消息设置完整的必要参数
@ -78,13 +81,16 @@ public class IotTcpBinaryDeviceMessageCodec implements IotDeviceMessageCodec {
payloadInfo.getRequestId(), method, payloadInfo.getParams()); payloadInfo.getRequestId(), method, payloadInfo.getParams());
// 5. 设置设备相关信息 // 5. 设置设备相关信息
// TODO @haohaoserverId 不是这里解析的哈
Long deviceId = parseDeviceId(dataPackage.getAddr()); Long deviceId = parseDeviceId(dataPackage.getAddr());
message.setDeviceId(deviceId); message.setDeviceId(deviceId);
// 6. 设置TCP协议相关信息 // 6. 设置 TCP 协议相关信息
// TODO @haohaoserverId 不是这里解析的哈
message.setServerId(generateServerId(dataPackage)); message.setServerId(generateServerId(dataPackage));
// 7. 设置租户IDTODO: 后续可以从设备信息中获取 // 7. 设置租户 IDTODO: 后续可以从设备信息中获取
// TODO @haohao租户 id 不是这里解析的哈
// message.setTenantId(getTenantIdByDeviceId(deviceId)); // message.setTenantId(getTenantIdByDeviceId(deviceId));
if (log.isDebugEnabled()) { if (log.isDebugEnabled()) {
@ -104,6 +110,7 @@ public class IotTcpBinaryDeviceMessageCodec implements IotDeviceMessageCodec {
return TYPE; return TYPE;
} }
// TODO @haohao这种简单解析中间不用空格哈
/** /**
* 构建完整负载 * 构建完整负载
*/ */
@ -130,12 +137,10 @@ public class IotTcpBinaryDeviceMessageCodec implements IotDeviceMessageCodec {
return payload.toString(); return payload.toString();
} }
// ==================== 编解码方法 ==================== // ==================== 编解码方法 ====================
/** /**
* 解析负载信息包含requestId和params * 解析负载信息包含 requestId params
*/ */
private PayloadInfo parsePayloadInfo(String payload) { private PayloadInfo parsePayloadInfo(String payload) {
if (StrUtil.isEmpty(payload)) { if (StrUtil.isEmpty(payload)) {
@ -143,6 +148,7 @@ public class IotTcpBinaryDeviceMessageCodec implements IotDeviceMessageCodec {
} }
try { try {
// TODO @haohao使用 jsonUtils
JSONObject jsonObject = JSONUtil.parseObj(payload); JSONObject jsonObject = JSONUtil.parseObj(payload);
String requestId = jsonObject.getStr(PayloadField.REQUEST_ID); String requestId = jsonObject.getStr(PayloadField.REQUEST_ID);
if (StrUtil.isEmpty(requestId)) { if (StrUtil.isEmpty(requestId)) {
@ -185,7 +191,7 @@ public class IotTcpBinaryDeviceMessageCodec implements IotDeviceMessageCodec {
* @return 服务ID * @return 服务ID
*/ */
private String generateServerId(TcpDataPackage dataPackage) { private String generateServerId(TcpDataPackage dataPackage) {
// 使用协议类型 + 设备地址 + 消息序号生成唯一的服务ID // 使用协议类型 + 设备地址 + 消息序号生成唯一的服务 ID
return String.format("tcp_%s_%d", dataPackage.getAddr(), dataPackage.getMid()); return String.format("tcp_%s_%d", dataPackage.getAddr(), dataPackage.getMid());
} }
@ -300,23 +306,28 @@ public class IotTcpBinaryDeviceMessageCodec implements IotDeviceMessageCodec {
* 消息方法常量 * 消息方法常量
*/ */
public static class MessageMethod { public static class MessageMethod {
public static final String PROPERTY_POST = "thing.property.post"; // 数据上报 public static final String PROPERTY_POST = "thing.property.post"; // 数据上报
public static final String STATE_ONLINE = "thing.state.online"; // 心跳 public static final String STATE_ONLINE = "thing.state.online"; // 心跳
} }
/** /**
* 负载字段名 * 负载字段名
*/ */
private static class PayloadField { private static class PayloadField {
public static final String METHOD = "method"; public static final String METHOD = "method";
public static final String PARAMS = "params"; public static final String PARAMS = "params";
public static final String TIMESTAMP = "timestamp"; public static final String TIMESTAMP = "timestamp";
public static final String REQUEST_ID = "requestId"; public static final String REQUEST_ID = "requestId";
public static final String MESSAGE_ID = "msgId"; public static final String MESSAGE_ID = "msgId";
} }
// ==================== TCP 数据包编解码方法 ==================== // ==================== TCP 数据包编解码方法 ====================
// TODO @haohaolombok 简化
/** /**
* 负载信息类 * 负载信息类
*/ */
@ -361,11 +372,13 @@ public class IotTcpBinaryDeviceMessageCodec implements IotDeviceMessageCodec {
// ==================== 自定义异常 ==================== // ==================== 自定义异常 ====================
// TODO @haohao可以搞个全局的
/** /**
* TCP 编解码异常 * TCP 编解码异常
*/ */
public static class TcpCodecException extends RuntimeException { public static class TcpCodecException extends RuntimeException {
// TODO @haohao非必要构造方法可以去掉哈
public TcpCodecException(String message) { public TcpCodecException(String message) {
super(message); super(message);
} }

View File

@ -8,11 +8,11 @@ import org.springframework.stereotype.Component;
/** /**
* TCP编解码器管理器简化版 * TCP编解码器管理器简化版
* *
* 核心功能 * 核心功能
* - 自动协议检测二进制 vs JSON * - 自动协议检测二进制 vs JSON
* - 统一编解码接口 * - 统一编解码接口
* - 默认使用JSON协议 * - 默认使用 JSON 协议
* *
* @author 芋道源码 * @author 芋道源码
*/ */
@ -22,6 +22,8 @@ public class IotTcpCodecManager implements IotDeviceMessageCodec {
public static final String TYPE = "TCP"; public static final String TYPE = "TCP";
// TODO @haohao@Resource
@Autowired @Autowired
private IotTcpBinaryDeviceMessageCodec binaryCodec; private IotTcpBinaryDeviceMessageCodec binaryCodec;
@ -40,21 +42,22 @@ public class IotTcpCodecManager implements IotDeviceMessageCodec {
@Override @Override
public byte[] encode(IotDeviceMessage message) { public byte[] encode(IotDeviceMessage message) {
// 默认使用JSON协议编码 // 默认使用 JSON 协议编码
return jsonCodec.encode(message); return jsonCodec.encode(message);
} }
// TODO @haohao要不还是不自动检测用户手动配置哈简化一些
@Override @Override
public IotDeviceMessage decode(byte[] bytes) { public IotDeviceMessage decode(byte[] bytes) {
// 自动检测协议类型并解码 // 自动检测协议类型并解码
if (isJsonFormat(bytes)) { if (isJsonFormat(bytes)) {
if (log.isDebugEnabled()) { if (log.isDebugEnabled()) {
log.debug("[decode][检测到JSON协议] 数据长度: {}字节", bytes.length); log.debug("[decode][检测到 JSON 协议,数据长度: {} 字节]", bytes.length);
} }
return jsonCodec.decode(bytes); return jsonCodec.decode(bytes);
} else { } else {
if (log.isDebugEnabled()) { if (log.isDebugEnabled()) {
log.debug("[decode][检测到二进制协议] 数据长度: {}字节", bytes.length); log.debug("[decode][检测到二进制协议,数据长度: {} 字节]", bytes.length);
} }
return binaryCodec.decode(bytes); return binaryCodec.decode(bytes);
} }
@ -63,7 +66,7 @@ public class IotTcpCodecManager implements IotDeviceMessageCodec {
// ==================== 便捷方法 ==================== // ==================== 便捷方法 ====================
/** /**
* 使用JSON协议编码 * 使用 JSON 协议编码
*/ */
public byte[] encodeJson(IotDeviceMessage message) { public byte[] encodeJson(IotDeviceMessage message) {
return jsonCodec.encode(message); return jsonCodec.encode(message);
@ -95,42 +98,46 @@ public class IotTcpCodecManager implements IotDeviceMessageCodec {
/** /**
* 检测是否为JSON格式 * 检测是否为JSON格式
* *
* 检测规则 * 检测规则
* 1. 数据以 '{' 开头 * 1. 数据以 '{' 开头
* 2. 包含 "method" "id" 字段 * 2. 包含 "method" "id" 字段
*/ */
private boolean isJsonFormat(byte[] bytes) { private boolean isJsonFormat(byte[] bytes) {
// TODO @haohaoArrayUtil.isEmpty(bytes) 可以简化下
if (bytes == null || bytes.length == 0) { if (bytes == null || bytes.length == 0) {
return useJsonByDefault; return useJsonByDefault;
} }
try { try {
// 检测JSON格式 '{' 开头 // 检测 JSON 格式 '{' 开头
if (bytes[0] == '{') { if (bytes[0] == '{') {
// 进一步验证是否为有效JSON // TODO @haohao不一定按照顺序写这个可能要看下
// 进一步验证是否为有效 JSON
String jsonStr = new String(bytes, 0, Math.min(bytes.length, 100)); String jsonStr = new String(bytes, 0, Math.min(bytes.length, 100));
return jsonStr.contains("\"method\"") || jsonStr.contains("\"id\""); return jsonStr.contains("\"method\"") || jsonStr.contains("\"id\"");
} }
// 检测二进制格式长度 >= 8 且符合二进制协议结构 // 检测二进制格式长度 >= 8 且符合二进制协议结构
if (bytes.length >= 8) { if (bytes.length >= 8) {
// 读取包头4字节表示后续数据长度 // 读取包头 4 字节表示后续数据长度
int expectedLength = ((bytes[0] & 0xFF) << 24) | int expectedLength = ((bytes[0] & 0xFF) << 24) |
((bytes[1] & 0xFF) << 16) | ((bytes[1] & 0xFF) << 16) |
((bytes[2] & 0xFF) << 8) | ((bytes[2] & 0xFF) << 8) |
(bytes[3] & 0xFF); (bytes[3] & 0xFF);
// 验证长度是否合理 // 验证长度是否合理
// TODO @haohaoexpectedLength > 0 多余的貌似
if (expectedLength == bytes.length - 4 && expectedLength > 0 && expectedLength < 1024 * 1024) { if (expectedLength == bytes.length - 4 && expectedLength > 0 && expectedLength < 1024 * 1024) {
return false; // 二进制格式 return false; // 二进制格式
} }
} }
} catch (Exception e) { } catch (Exception e) {
log.warn("[isJsonFormat][协议检测异常] 使用默认协议: {}", getDefaultProtocol(), e); log.warn("[isJsonFormat][协议检测异常,使用默认协议: {}]", getDefaultProtocol(), e);
} }
// 默认使用当前设置的协议类型 // 默认使用当前设置的协议类型
return useJsonByDefault; return useJsonByDefault;
} }
} }

View File

@ -12,16 +12,16 @@ import org.springframework.stereotype.Component;
import java.nio.charset.StandardCharsets; import java.nio.charset.StandardCharsets;
/** /**
* TCP JSON格式 {@link IotDeviceMessage} 编解码器 * TCP JSON 格式 {@link IotDeviceMessage} 编解码器
* *
* 采用纯JSON格式传输参考EMQX和HTTP模块的数据格式 * 采用纯 JSON 格式传输参考 EMQX HTTP 模块的数据格式
* *
* JSON消息格式 * JSON消息格式
* { * {
* "id": "消息ID", * "id": "消息 ID",
* "method": "消息方法", * "method": "消息方法",
* "deviceId": "设备ID", * "deviceId": "设备 ID",
* "productKey": "产品Key", * "productKey": "产品 Key",
* "deviceName": "设备名称", * "deviceName": "设备名称",
* "params": {...}, * "params": {...},
* "timestamp": 时间戳 * "timestamp": 时间戳
@ -35,6 +35,7 @@ public class IotTcpJsonDeviceMessageCodec implements IotDeviceMessageCodec {
public static final String TYPE = "TCP_JSON"; public static final String TYPE = "TCP_JSON";
// TODO @haohao变量不太对
// ==================== 常量定义 ==================== // ==================== 常量定义 ====================
@Override @Override
@ -77,14 +78,15 @@ public class IotTcpJsonDeviceMessageCodec implements IotDeviceMessageCodec {
} }
try { try {
// 转换为JSON字符串 // 转换为 JSON 字符串
String jsonString = new String(bytes, StandardCharsets.UTF_8); String jsonString = new String(bytes, StandardCharsets.UTF_8);
if (log.isDebugEnabled()) { if (log.isDebugEnabled()) {
log.debug("[decode][开始解码] JSON长度: {}字节, 内容: {}", bytes.length, jsonString); log.debug("[decode][开始解码] JSON长度: {}字节, 内容: {}", bytes.length, jsonString);
} }
// 解析JSON消息 // 解析 JSON 消息
// TODO @haohaoJsonUtils
JSONObject jsonMessage = JSONUtil.parseObj(jsonString); JSONObject jsonMessage = JSONUtil.parseObj(jsonString);
// 构建IoT设备消息 // 构建IoT设备消息
@ -129,7 +131,7 @@ public class IotTcpJsonDeviceMessageCodec implements IotDeviceMessageCodec {
} }
/** /**
* 构建JSON消息 * 构建 JSON 消息
*/ */
private JSONObject buildJsonMessage(IotDeviceMessage message) { private JSONObject buildJsonMessage(IotDeviceMessage message) {
JSONObject jsonMessage = new JSONObject(); JSONObject jsonMessage = new JSONObject();
@ -189,7 +191,7 @@ public class IotTcpJsonDeviceMessageCodec implements IotDeviceMessageCodec {
message.setMsg(msg); message.setMsg(msg);
} }
// 设置服务ID基于JSON格式 // 设置服务 ID基于 JSON 格式
message.setServerId(generateServerId(jsonMessage)); message.setServerId(generateServerId(jsonMessage));
return message; return message;
@ -216,22 +218,26 @@ public class IotTcpJsonDeviceMessageCodec implements IotDeviceMessageCodec {
StrUtil.isNotEmpty(id) ? id.substring(0, Math.min(8, id.length())) : "noId"); StrUtil.isNotEmpty(id) ? id.substring(0, Math.min(8, id.length())) : "noId");
} }
// TODO @haohao注释格式不对
/** /**
* 消息方法常量 * 消息方法常量
*/ */
public static class MessageMethod { public static class MessageMethod {
public static final String PROPERTY_POST = "thing.property.post"; // 数据上报 public static final String PROPERTY_POST = "thing.property.post"; // 数据上报
public static final String STATE_ONLINE = "thing.state.online"; // 心跳 public static final String STATE_ONLINE = "thing.state.online"; // 心跳
public static final String EVENT_POST = "thing.event.post"; // 事件上报 public static final String EVENT_POST = "thing.event.post"; // 事件上报
public static final String PROPERTY_SET = "thing.property.set"; // 属性设置 public static final String PROPERTY_SET = "thing.property.set"; // 属性设置
public static final String PROPERTY_GET = "thing.property.get"; // 属性获取 public static final String PROPERTY_GET = "thing.property.get"; // 属性获取
public static final String SERVICE_INVOKE = "thing.service.invoke"; // 服务调用 public static final String SERVICE_INVOKE = "thing.service.invoke"; // 服务调用
} }
/** /**
* JSON字段名参考EMQX和HTTP模块格式 * JSON字段名参考EMQX和HTTP模块格式
*/ */
private static class JsonField { private static class JsonField {
public static final String ID = "id"; public static final String ID = "id";
public static final String METHOD = "method"; public static final String METHOD = "method";
public static final String DEVICE_ID = "deviceId"; public static final String DEVICE_ID = "deviceId";
@ -241,5 +247,6 @@ public class IotTcpJsonDeviceMessageCodec implements IotDeviceMessageCodec {
public static final String TIMESTAMP = "timestamp"; public static final String TIMESTAMP = "timestamp";
public static final String CODE = "code"; public static final String CODE = "code";
public static final String MESSAGE = "message"; public static final String MESSAGE = "message";
} }
} }

View File

@ -30,6 +30,7 @@ public class IotTcpUpstreamProtocol {
private final IotDeviceMessageService messageService; private final IotDeviceMessageService messageService;
// TODO @haohao不用的变量可以删除
private final IotDeviceCommonApi deviceApi; private final IotDeviceCommonApi deviceApi;
private final IotTcpCodecManager codecManager; private final IotTcpCodecManager codecManager;
@ -58,6 +59,7 @@ public class IotTcpUpstreamProtocol {
@PostConstruct @PostConstruct
public void start() { public void start() {
// TODO @haohao类似下面 62 75 是处理 options 因为中间写了注释其实可以不用空行然后 77 91 可以中间空喊去掉更紧凑一点
// 创建服务器选项 // 创建服务器选项
NetServerOptions options = new NetServerOptions() NetServerOptions options = new NetServerOptions()
.setPort(tcpProperties.getPort()) .setPort(tcpProperties.getPort())

View File

@ -1,7 +1,6 @@
package cn.iocoder.yudao.module.iot.gateway.protocol.tcp.router; package cn.iocoder.yudao.module.iot.gateway.protocol.tcp.router;
import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
import cn.iocoder.yudao.module.iot.gateway.codec.tcp.IotTcpDeviceMessageCodec;
import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService; import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
@ -24,11 +23,12 @@ public class IotTcpDownstreamHandler {
private final IotDeviceMessageService messageService; private final IotDeviceMessageService messageService;
private final IotTcpDeviceMessageCodec codec; // TODO @haohao代码没提交全有报错
// private final IotTcpDeviceMessageCodec codec;
public IotTcpDownstreamHandler(IotDeviceMessageService messageService) { public IotTcpDownstreamHandler(IotDeviceMessageService messageService) {
this.messageService = messageService; this.messageService = messageService;
this.codec = new IotTcpDeviceMessageCodec(); // this.codec = new IotTcpDeviceMessageCodec();
} }
/** /**
@ -42,7 +42,8 @@ public class IotTcpDownstreamHandler {
message.getDeviceId(), message.getMethod(), message.getId()); message.getDeviceId(), message.getMethod(), message.getId());
// 编码消息用于日志记录和验证 // 编码消息用于日志记录和验证
byte[] encodedMessage = codec.encode(message); byte[] encodedMessage = null;
// codec.encode(message);
log.debug("[handle][消息编码成功] 设备ID: {}, 编码后长度: {} 字节", log.debug("[handle][消息编码成功] 设备ID: {}, 编码后长度: {} 字节",
message.getDeviceId(), encodedMessage.length); message.getDeviceId(), encodedMessage.length);

View File

@ -6,6 +6,7 @@ import lombok.extern.slf4j.Slf4j;
import java.util.HashMap; import java.util.HashMap;
import java.util.Map; import java.util.Map;
// TODO @haohao这种写成单测会好点
/** /**
* TCP二进制格式数据包示例 * TCP二进制格式数据包示例
* *
@ -21,13 +22,13 @@ public class TcpBinaryDataPacketExamples {
public static void main(String[] args) { public static void main(String[] args) {
IotTcpBinaryDeviceMessageCodec codec = new IotTcpBinaryDeviceMessageCodec(); IotTcpBinaryDeviceMessageCodec codec = new IotTcpBinaryDeviceMessageCodec();
// 1. 数据上报包示例 // 1. 数据上报包示例
demonstrateDataReport(codec); demonstrateDataReport(codec);
// 2. 心跳包示例 // 2. 心跳包示例
demonstrateHeartbeat(codec); demonstrateHeartbeat(codec);
// 3. 复杂数据上报示例 // 3. 复杂数据上报示例
demonstrateComplexDataReport(codec); demonstrateComplexDataReport(codec);
} }
@ -37,23 +38,23 @@ public class TcpBinaryDataPacketExamples {
*/ */
private static void demonstrateDataReport(IotTcpBinaryDeviceMessageCodec codec) { private static void demonstrateDataReport(IotTcpBinaryDeviceMessageCodec codec) {
log.info("=== 二进制格式数据上报包示例 ==="); log.info("=== 二进制格式数据上报包示例 ===");
// 创建传感器数据 // 创建传感器数据
Map<String, Object> sensorData = new HashMap<>(); Map<String, Object> sensorData = new HashMap<>();
sensorData.put("temperature", 25.5); sensorData.put("temperature", 25.5);
sensorData.put("humidity", 60.2); sensorData.put("humidity", 60.2);
sensorData.put("pressure", 1013.25); sensorData.put("pressure", 1013.25);
sensorData.put("battery", 85); sensorData.put("battery", 85);
// 创建设备消息 // 创建设备消息
IotDeviceMessage message = IotDeviceMessage.requestOf("thing.property.post", sensorData); IotDeviceMessage message = IotDeviceMessage.requestOf("thing.property.post", sensorData);
message.setDeviceId(123456L); message.setDeviceId(123456L);
// 编码 // 编码
byte[] packet = codec.encode(message); byte[] packet = codec.encode(message);
log.info("编码后数据包长度: {} 字节", packet.length); log.info("编码后数据包长度: {} 字节", packet.length);
log.info("编码后数据包(HEX): {}", bytesToHex(packet)); log.info("编码后数据包(HEX): {}", bytesToHex(packet));
// 解码验证 // 解码验证
IotDeviceMessage decoded = codec.decode(packet); IotDeviceMessage decoded = codec.decode(packet);
log.info("解码后消息ID: {}", decoded.getId()); log.info("解码后消息ID: {}", decoded.getId());
@ -63,7 +64,7 @@ public class TcpBinaryDataPacketExamples {
log.info("解码后服务ID: {}", decoded.getServerId()); log.info("解码后服务ID: {}", decoded.getServerId());
log.info("解码后上报时间: {}", decoded.getReportTime()); log.info("解码后上报时间: {}", decoded.getReportTime());
log.info("解码后参数: {}", decoded.getParams()); log.info("解码后参数: {}", decoded.getParams());
System.out.println(); System.out.println();
} }
@ -72,16 +73,16 @@ public class TcpBinaryDataPacketExamples {
*/ */
private static void demonstrateHeartbeat(IotTcpBinaryDeviceMessageCodec codec) { private static void demonstrateHeartbeat(IotTcpBinaryDeviceMessageCodec codec) {
log.info("=== 二进制格式心跳包示例 ==="); log.info("=== 二进制格式心跳包示例 ===");
// 创建心跳消息 // 创建心跳消息
IotDeviceMessage heartbeat = IotDeviceMessage.requestOf("thing.state.online", null); IotDeviceMessage heartbeat = IotDeviceMessage.requestOf("thing.state.online", null);
heartbeat.setDeviceId(123456L); heartbeat.setDeviceId(123456L);
// 编码 // 编码
byte[] packet = codec.encode(heartbeat); byte[] packet = codec.encode(heartbeat);
log.info("心跳包长度: {} 字节", packet.length); log.info("心跳包长度: {} 字节", packet.length);
log.info("心跳包(HEX): {}", bytesToHex(packet)); log.info("心跳包(HEX): {}", bytesToHex(packet));
// 解码验证 // 解码验证
IotDeviceMessage decoded = codec.decode(packet); IotDeviceMessage decoded = codec.decode(packet);
log.info("解码后消息ID: {}", decoded.getId()); log.info("解码后消息ID: {}", decoded.getId());
@ -90,7 +91,7 @@ public class TcpBinaryDataPacketExamples {
log.info("解码后设备ID: {}", decoded.getDeviceId()); log.info("解码后设备ID: {}", decoded.getDeviceId());
log.info("解码后服务ID: {}", decoded.getServerId()); log.info("解码后服务ID: {}", decoded.getServerId());
log.info("解码后参数: {}", decoded.getParams()); log.info("解码后参数: {}", decoded.getParams());
System.out.println(); System.out.println();
} }
@ -99,40 +100,40 @@ public class TcpBinaryDataPacketExamples {
*/ */
private static void demonstrateComplexDataReport(IotTcpBinaryDeviceMessageCodec codec) { private static void demonstrateComplexDataReport(IotTcpBinaryDeviceMessageCodec codec) {
log.info("=== 二进制格式复杂数据上报示例 ==="); log.info("=== 二进制格式复杂数据上报示例 ===");
// 创建复杂设备数据 // 创建复杂设备数据
Map<String, Object> deviceData = new HashMap<>(); Map<String, Object> deviceData = new HashMap<>();
// 环境数据 // 环境数据
Map<String, Object> environment = new HashMap<>(); Map<String, Object> environment = new HashMap<>();
environment.put("temperature", 23.8); environment.put("temperature", 23.8);
environment.put("humidity", 55.0); environment.put("humidity", 55.0);
environment.put("co2", 420); environment.put("co2", 420);
deviceData.put("environment", environment); deviceData.put("environment", environment);
// GPS数据 // GPS数据
Map<String, Object> location = new HashMap<>(); Map<String, Object> location = new HashMap<>();
location.put("latitude", 39.9042); location.put("latitude", 39.9042);
location.put("longitude", 116.4074); location.put("longitude", 116.4074);
location.put("altitude", 43.5); location.put("altitude", 43.5);
deviceData.put("location", location); deviceData.put("location", location);
// 设备状态 // 设备状态
Map<String, Object> status = new HashMap<>(); Map<String, Object> status = new HashMap<>();
status.put("battery", 78); status.put("battery", 78);
status.put("signal", -65); status.put("signal", -65);
status.put("online", true); status.put("online", true);
deviceData.put("status", status); deviceData.put("status", status);
// 创建设备消息 // 创建设备消息
IotDeviceMessage message = IotDeviceMessage.requestOf("thing.property.post", deviceData); IotDeviceMessage message = IotDeviceMessage.requestOf("thing.property.post", deviceData);
message.setDeviceId(789012L); message.setDeviceId(789012L);
// 编码 // 编码
byte[] packet = codec.encode(message); byte[] packet = codec.encode(message);
log.info("复杂数据包长度: {} 字节", packet.length); log.info("复杂数据包长度: {} 字节", packet.length);
log.info("复杂数据包(HEX): {}", bytesToHex(packet)); log.info("复杂数据包(HEX): {}", bytesToHex(packet));
// 解码验证 // 解码验证
IotDeviceMessage decoded = codec.decode(packet); IotDeviceMessage decoded = codec.decode(packet);
log.info("解码后消息ID: {}", decoded.getId()); log.info("解码后消息ID: {}", decoded.getId());
@ -141,7 +142,7 @@ public class TcpBinaryDataPacketExamples {
log.info("解码后设备ID: {}", decoded.getDeviceId()); log.info("解码后设备ID: {}", decoded.getDeviceId());
log.info("解码后服务ID: {}", decoded.getServerId()); log.info("解码后服务ID: {}", decoded.getServerId());
log.info("解码后参数: {}", decoded.getParams()); log.info("解码后参数: {}", decoded.getParams());
System.out.println(); System.out.println();
} }
@ -164,9 +165,9 @@ public class TcpBinaryDataPacketExamples {
log.error("数据包长度不足"); log.error("数据包长度不足");
return; return;
} }
int index = 0; int index = 0;
// 解析包头(4字节) - 后续数据长度 // 解析包头(4字节) - 后续数据长度
int totalLength = ((packet[index] & 0xFF) << 24) | int totalLength = ((packet[index] & 0xFF) << 24) |
((packet[index + 1] & 0xFF) << 16) | ((packet[index + 1] & 0xFF) << 16) |
@ -174,27 +175,27 @@ public class TcpBinaryDataPacketExamples {
(packet[index + 3] & 0xFF); (packet[index + 3] & 0xFF);
index += 4; index += 4;
log.info("包头 - 后续数据长度: {} 字节", totalLength); log.info("包头 - 后续数据长度: {} 字节", totalLength);
// 解析设备地址长度(2字节) // 解析设备地址长度(2字节)
int addrLength = ((packet[index] & 0xFF) << 8) | (packet[index + 1] & 0xFF); int addrLength = ((packet[index] & 0xFF) << 8) | (packet[index + 1] & 0xFF);
index += 2; index += 2;
log.info("设备地址长度: {} 字节", addrLength); log.info("设备地址长度: {} 字节", addrLength);
// 解析设备地址 // 解析设备地址
String deviceAddr = new String(packet, index, addrLength); String deviceAddr = new String(packet, index, addrLength);
index += addrLength; index += addrLength;
log.info("设备地址: {}", deviceAddr); log.info("设备地址: {}", deviceAddr);
// 解析功能码(2字节) // 解析功能码(2字节)
int functionCode = ((packet[index] & 0xFF) << 8) | (packet[index + 1] & 0xFF); int functionCode = ((packet[index] & 0xFF) << 8) | (packet[index + 1] & 0xFF);
index += 2; index += 2;
log.info("功能码: {} ({})", functionCode, getFunctionCodeName(functionCode)); log.info("功能码: {} ({})", functionCode, getFunctionCodeName(functionCode));
// 解析消息序号(2字节) // 解析消息序号(2字节)
int messageId = ((packet[index] & 0xFF) << 8) | (packet[index + 1] & 0xFF); int messageId = ((packet[index] & 0xFF) << 8) | (packet[index + 1] & 0xFF);
index += 2; index += 2;
log.info("消息序号: {}", messageId); log.info("消息序号: {}", messageId);
// 解析包体数据 // 解析包体数据
if (index < packet.length) { if (index < packet.length) {
String payload = new String(packet, index, packet.length - index); String payload = new String(packet, index, packet.length - index);

View File

@ -7,9 +7,10 @@ import java.nio.charset.StandardCharsets;
import java.util.HashMap; import java.util.HashMap;
import java.util.Map; import java.util.Map;
// TODO @haohao这种写成单测会好点
/** /**
* TCP JSON格式数据包示例 * TCP JSON格式数据包示例
* *
* 演示如何使用新的JSON格式进行TCP消息编解码 * 演示如何使用新的JSON格式进行TCP消息编解码
* *
* @author 芋道源码 * @author 芋道源码
@ -19,22 +20,22 @@ public class TcpJsonDataPacketExamples {
public static void main(String[] args) { public static void main(String[] args) {
IotTcpJsonDeviceMessageCodec codec = new IotTcpJsonDeviceMessageCodec(); IotTcpJsonDeviceMessageCodec codec = new IotTcpJsonDeviceMessageCodec();
// 1. 数据上报示例 // 1. 数据上报示例
demonstrateDataReport(codec); demonstrateDataReport(codec);
// 2. 心跳示例 // 2. 心跳示例
demonstrateHeartbeat(codec); demonstrateHeartbeat(codec);
// 3. 事件上报示例 // 3. 事件上报示例
demonstrateEventReport(codec); demonstrateEventReport(codec);
// 4. 复杂数据上报示例 // 4. 复杂数据上报示例
demonstrateComplexDataReport(codec); demonstrateComplexDataReport(codec);
// 5. 便捷方法示例 // 5. 便捷方法示例
demonstrateConvenienceMethods(); demonstrateConvenienceMethods();
// 6. EMQX兼容性示例 // 6. EMQX兼容性示例
demonstrateEmqxCompatibility(); demonstrateEmqxCompatibility();
} }
@ -44,24 +45,24 @@ public class TcpJsonDataPacketExamples {
*/ */
private static void demonstrateDataReport(IotTcpJsonDeviceMessageCodec codec) { private static void demonstrateDataReport(IotTcpJsonDeviceMessageCodec codec) {
log.info("=== JSON格式数据上报示例 ==="); log.info("=== JSON格式数据上报示例 ===");
// 创建传感器数据 // 创建传感器数据
Map<String, Object> sensorData = new HashMap<>(); Map<String, Object> sensorData = new HashMap<>();
sensorData.put("temperature", 25.5); sensorData.put("temperature", 25.5);
sensorData.put("humidity", 60.2); sensorData.put("humidity", 60.2);
sensorData.put("pressure", 1013.25); sensorData.put("pressure", 1013.25);
sensorData.put("battery", 85); sensorData.put("battery", 85);
// 创建设备消息 // 创建设备消息
IotDeviceMessage message = IotDeviceMessage.requestOf("thing.property.post", sensorData); IotDeviceMessage message = IotDeviceMessage.requestOf("thing.property.post", sensorData);
message.setDeviceId(123456L); message.setDeviceId(123456L);
// 编码 // 编码
byte[] packet = codec.encode(message); byte[] packet = codec.encode(message);
String jsonString = new String(packet, StandardCharsets.UTF_8); String jsonString = new String(packet, StandardCharsets.UTF_8);
log.info("编码后JSON: {}", jsonString); log.info("编码后JSON: {}", jsonString);
log.info("数据包长度: {} 字节", packet.length); log.info("数据包长度: {} 字节", packet.length);
// 解码验证 // 解码验证
IotDeviceMessage decoded = codec.decode(packet); IotDeviceMessage decoded = codec.decode(packet);
log.info("解码后消息ID: {}", decoded.getId()); log.info("解码后消息ID: {}", decoded.getId());
@ -69,7 +70,7 @@ public class TcpJsonDataPacketExamples {
log.info("解码后设备ID: {}", decoded.getDeviceId()); log.info("解码后设备ID: {}", decoded.getDeviceId());
log.info("解码后服务ID: {}", decoded.getServerId()); log.info("解码后服务ID: {}", decoded.getServerId());
log.info("解码后参数: {}", decoded.getParams()); log.info("解码后参数: {}", decoded.getParams());
System.out.println(); System.out.println();
} }
@ -78,24 +79,24 @@ public class TcpJsonDataPacketExamples {
*/ */
private static void demonstrateHeartbeat(IotTcpJsonDeviceMessageCodec codec) { private static void demonstrateHeartbeat(IotTcpJsonDeviceMessageCodec codec) {
log.info("=== JSON格式心跳示例 ==="); log.info("=== JSON格式心跳示例 ===");
// 创建心跳消息 // 创建心跳消息
IotDeviceMessage heartbeat = IotDeviceMessage.requestOf("thing.state.online", null); IotDeviceMessage heartbeat = IotDeviceMessage.requestOf("thing.state.online", null);
heartbeat.setDeviceId(123456L); heartbeat.setDeviceId(123456L);
// 编码 // 编码
byte[] packet = codec.encode(heartbeat); byte[] packet = codec.encode(heartbeat);
String jsonString = new String(packet, StandardCharsets.UTF_8); String jsonString = new String(packet, StandardCharsets.UTF_8);
log.info("编码后JSON: {}", jsonString); log.info("编码后JSON: {}", jsonString);
log.info("心跳包长度: {} 字节", packet.length); log.info("心跳包长度: {} 字节", packet.length);
// 解码验证 // 解码验证
IotDeviceMessage decoded = codec.decode(packet); IotDeviceMessage decoded = codec.decode(packet);
log.info("解码后消息ID: {}", decoded.getId()); log.info("解码后消息ID: {}", decoded.getId());
log.info("解码后方法: {}", decoded.getMethod()); log.info("解码后方法: {}", decoded.getMethod());
log.info("解码后设备ID: {}", decoded.getDeviceId()); log.info("解码后设备ID: {}", decoded.getDeviceId());
log.info("解码后服务ID: {}", decoded.getServerId()); log.info("解码后服务ID: {}", decoded.getServerId());
System.out.println(); System.out.println();
} }
@ -104,31 +105,31 @@ public class TcpJsonDataPacketExamples {
*/ */
private static void demonstrateEventReport(IotTcpJsonDeviceMessageCodec codec) { private static void demonstrateEventReport(IotTcpJsonDeviceMessageCodec codec) {
log.info("=== JSON格式事件上报示例 ==="); log.info("=== JSON格式事件上报示例 ===");
// 创建事件数据 // 创建事件数据
Map<String, Object> eventData = new HashMap<>(); Map<String, Object> eventData = new HashMap<>();
eventData.put("eventType", "alarm"); eventData.put("eventType", "alarm");
eventData.put("level", "warning"); eventData.put("level", "warning");
eventData.put("description", "温度过高"); eventData.put("description", "温度过高");
eventData.put("value", 45.8); eventData.put("value", 45.8);
// 创建事件消息 // 创建事件消息
IotDeviceMessage event = IotDeviceMessage.requestOf("thing.event.post", eventData); IotDeviceMessage event = IotDeviceMessage.requestOf("thing.event.post", eventData);
event.setDeviceId(123456L); event.setDeviceId(123456L);
// 编码 // 编码
byte[] packet = codec.encode(event); byte[] packet = codec.encode(event);
String jsonString = new String(packet, StandardCharsets.UTF_8); String jsonString = new String(packet, StandardCharsets.UTF_8);
log.info("编码后JSON: {}", jsonString); log.info("编码后JSON: {}", jsonString);
log.info("事件包长度: {} 字节", packet.length); log.info("事件包长度: {} 字节", packet.length);
// 解码验证 // 解码验证
IotDeviceMessage decoded = codec.decode(packet); IotDeviceMessage decoded = codec.decode(packet);
log.info("解码后消息ID: {}", decoded.getId()); log.info("解码后消息ID: {}", decoded.getId());
log.info("解码后方法: {}", decoded.getMethod()); log.info("解码后方法: {}", decoded.getMethod());
log.info("解码后设备ID: {}", decoded.getDeviceId()); log.info("解码后设备ID: {}", decoded.getDeviceId());
log.info("解码后参数: {}", decoded.getParams()); log.info("解码后参数: {}", decoded.getParams());
System.out.println(); System.out.println();
} }
@ -137,10 +138,10 @@ public class TcpJsonDataPacketExamples {
*/ */
private static void demonstrateComplexDataReport(IotTcpJsonDeviceMessageCodec codec) { private static void demonstrateComplexDataReport(IotTcpJsonDeviceMessageCodec codec) {
log.info("=== JSON格式复杂数据上报示例 ==="); log.info("=== JSON格式复杂数据上报示例 ===");
// 创建复杂设备数据类似EMQX格式 // 创建复杂设备数据类似EMQX格式
Map<String, Object> deviceData = new HashMap<>(); Map<String, Object> deviceData = new HashMap<>();
// 环境数据 // 环境数据
Map<String, Object> environment = new HashMap<>(); Map<String, Object> environment = new HashMap<>();
environment.put("temperature", 23.8); environment.put("temperature", 23.8);
@ -148,7 +149,7 @@ public class TcpJsonDataPacketExamples {
environment.put("co2", 420); environment.put("co2", 420);
environment.put("pm25", 35); environment.put("pm25", 35);
deviceData.put("environment", environment); deviceData.put("environment", environment);
// GPS数据 // GPS数据
Map<String, Object> location = new HashMap<>(); Map<String, Object> location = new HashMap<>();
location.put("latitude", 39.9042); location.put("latitude", 39.9042);
@ -156,7 +157,7 @@ public class TcpJsonDataPacketExamples {
location.put("altitude", 43.5); location.put("altitude", 43.5);
location.put("speed", 0.0); location.put("speed", 0.0);
deviceData.put("location", location); deviceData.put("location", location);
// 设备状态 // 设备状态
Map<String, Object> status = new HashMap<>(); Map<String, Object> status = new HashMap<>();
status.put("battery", 78); status.put("battery", 78);
@ -164,24 +165,24 @@ public class TcpJsonDataPacketExamples {
status.put("online", true); status.put("online", true);
status.put("version", "1.2.3"); status.put("version", "1.2.3");
deviceData.put("status", status); deviceData.put("status", status);
// 创建设备消息 // 创建设备消息
IotDeviceMessage message = IotDeviceMessage.requestOf("thing.property.post", deviceData); IotDeviceMessage message = IotDeviceMessage.requestOf("thing.property.post", deviceData);
message.setDeviceId(789012L); message.setDeviceId(789012L);
// 编码 // 编码
byte[] packet = codec.encode(message); byte[] packet = codec.encode(message);
String jsonString = new String(packet, StandardCharsets.UTF_8); String jsonString = new String(packet, StandardCharsets.UTF_8);
log.info("编码后JSON: {}", jsonString); log.info("编码后JSON: {}", jsonString);
log.info("复杂数据包长度: {} 字节", packet.length); log.info("复杂数据包长度: {} 字节", packet.length);
// 解码验证 // 解码验证
IotDeviceMessage decoded = codec.decode(packet); IotDeviceMessage decoded = codec.decode(packet);
log.info("解码后消息ID: {}", decoded.getId()); log.info("解码后消息ID: {}", decoded.getId());
log.info("解码后方法: {}", decoded.getMethod()); log.info("解码后方法: {}", decoded.getMethod());
log.info("解码后设备ID: {}", decoded.getDeviceId()); log.info("解码后设备ID: {}", decoded.getDeviceId());
log.info("解码后参数: {}", decoded.getParams()); log.info("解码后参数: {}", decoded.getParams());
System.out.println(); System.out.println();
} }
@ -190,9 +191,9 @@ public class TcpJsonDataPacketExamples {
*/ */
private static void demonstrateConvenienceMethods() { private static void demonstrateConvenienceMethods() {
log.info("=== 便捷方法示例 ==="); log.info("=== 便捷方法示例 ===");
IotTcpJsonDeviceMessageCodec codec = new IotTcpJsonDeviceMessageCodec(); IotTcpJsonDeviceMessageCodec codec = new IotTcpJsonDeviceMessageCodec();
// 使用便捷方法编码数据上报 // 使用便捷方法编码数据上报
Map<String, Object> sensorData = Map.of( Map<String, Object> sensorData = Map.of(
"temperature", 26.5, "temperature", 26.5,
@ -200,11 +201,11 @@ public class TcpJsonDataPacketExamples {
); );
byte[] dataPacket = codec.encodeDataReport(sensorData, 123456L, "smart_sensor", "device_001"); byte[] dataPacket = codec.encodeDataReport(sensorData, 123456L, "smart_sensor", "device_001");
log.info("便捷方法编码数据上报: {}", new String(dataPacket, StandardCharsets.UTF_8)); log.info("便捷方法编码数据上报: {}", new String(dataPacket, StandardCharsets.UTF_8));
// 使用便捷方法编码心跳 // 使用便捷方法编码心跳
byte[] heartbeatPacket = codec.encodeHeartbeat(123456L, "smart_sensor", "device_001"); byte[] heartbeatPacket = codec.encodeHeartbeat(123456L, "smart_sensor", "device_001");
log.info("便捷方法编码心跳: {}", new String(heartbeatPacket, StandardCharsets.UTF_8)); log.info("便捷方法编码心跳: {}", new String(heartbeatPacket, StandardCharsets.UTF_8));
// 使用便捷方法编码事件 // 使用便捷方法编码事件
Map<String, Object> eventData = Map.of( Map<String, Object> eventData = Map.of(
"eventType", "maintenance", "eventType", "maintenance",
@ -212,7 +213,7 @@ public class TcpJsonDataPacketExamples {
); );
byte[] eventPacket = codec.encodeEventReport(eventData, 123456L, "smart_sensor", "device_001"); byte[] eventPacket = codec.encodeEventReport(eventData, 123456L, "smart_sensor", "device_001");
log.info("便捷方法编码事件: {}", new String(eventPacket, StandardCharsets.UTF_8)); log.info("便捷方法编码事件: {}", new String(eventPacket, StandardCharsets.UTF_8));
System.out.println(); System.out.println();
} }
@ -221,7 +222,7 @@ public class TcpJsonDataPacketExamples {
*/ */
private static void demonstrateEmqxCompatibility() { private static void demonstrateEmqxCompatibility() {
log.info("=== EMQX格式兼容性示例 ==="); log.info("=== EMQX格式兼容性示例 ===");
// 模拟EMQX风格的消息格式 // 模拟EMQX风格的消息格式
String emqxStyleJson = """ String emqxStyleJson = """
{ {
@ -235,19 +236,19 @@ public class TcpJsonDataPacketExamples {
"timestamp": 1642781234567 "timestamp": 1642781234567
} }
"""; """;
IotTcpJsonDeviceMessageCodec codec = new IotTcpJsonDeviceMessageCodec(); IotTcpJsonDeviceMessageCodec codec = new IotTcpJsonDeviceMessageCodec();
// 解码EMQX风格的消息 // 解码EMQX风格的消息
byte[] emqxBytes = emqxStyleJson.getBytes(StandardCharsets.UTF_8); byte[] emqxBytes = emqxStyleJson.getBytes(StandardCharsets.UTF_8);
IotDeviceMessage decoded = codec.decode(emqxBytes); IotDeviceMessage decoded = codec.decode(emqxBytes);
log.info("EMQX风格消息解码成功:"); log.info("EMQX风格消息解码成功:");
log.info("消息ID: {}", decoded.getId()); log.info("消息ID: {}", decoded.getId());
log.info("方法: {}", decoded.getMethod()); log.info("方法: {}", decoded.getMethod());
log.info("设备ID: {}", decoded.getDeviceId()); log.info("设备ID: {}", decoded.getDeviceId());
log.info("参数: {}", decoded.getParams()); log.info("参数: {}", decoded.getParams());
System.out.println(); System.out.println();
} }
} }