feat(iot): 实现设备 API 接口及更新网关处理逻辑
This commit is contained in:
@@ -1,388 +1,399 @@
|
||||
package com.viewsh.module.iot.gateway.protocol.tcp.router;
|
||||
|
||||
import cn.hutool.core.util.IdUtil;
|
||||
import cn.hutool.core.util.StrUtil;
|
||||
import com.viewsh.module.iot.core.biz.dto.IotDeviceRespDTO;
|
||||
import com.viewsh.module.iot.core.mq.message.IotDeviceMessage;
|
||||
import com.viewsh.module.iot.gateway.codec.jt808.IotJt808DeviceMessageCodec;
|
||||
import com.viewsh.module.iot.gateway.codec.tcp.IotTcpBinaryDeviceMessageCodec;
|
||||
import com.viewsh.module.iot.gateway.codec.tcp.IotTcpJsonDeviceMessageCodec;
|
||||
import com.viewsh.module.iot.gateway.protocol.tcp.IotTcpUpstreamProtocol;
|
||||
import com.viewsh.module.iot.gateway.protocol.tcp.handler.AuthResult;
|
||||
import com.viewsh.module.iot.gateway.protocol.tcp.handler.ProtocolHandler;
|
||||
import com.viewsh.module.iot.gateway.protocol.tcp.manager.IotTcpConnectionManager;
|
||||
import com.viewsh.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 lombok.extern.slf4j.Slf4j;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* TCP 上行消息处理器（重构版）
|
||||
* <p>
|
||||
* 职责：
|
||||
* 1. 管理 TCP 连接生命周期（连接建立、异常、关闭）
|
||||
* 2. 检测消息格式类型（JSON/Binary/JT808）
|
||||
* 3. 解码设备消息
|
||||
* 4. 路由到对应的协议处理器
|
||||
* 5. 管理设备认证状态
|
||||
* 6. 发送设备上线/离线消息
|
||||
* <p>
|
||||
* 设计原则：
|
||||
* - 主处理器只负责路由，不包含协议特定逻辑
|
||||
* - 协议处理器通过 Spring 自动注入，实现插件化
|
||||
* - 认证成功后，统一注册连接和发送上线消息
|
||||
* - 支持多步认证（如 JT808 的注册+鉴权）
|
||||
*
|
||||
* @author 芋道源码
|
||||
*/
|
||||
@Slf4j
|
||||
public class IotTcpUpstreamHandler implements Handler<NetSocket> {
|
||||
|
||||
private static final String CODEC_TYPE_JSON = IotTcpJsonDeviceMessageCodec.TYPE;
|
||||
private static final String CODEC_TYPE_BINARY = IotTcpBinaryDeviceMessageCodec.TYPE;
|
||||
private static final String CODEC_TYPE_JT808 = IotJt808DeviceMessageCodec.TYPE;
|
||||
|
||||
private final IotDeviceMessageService deviceMessageService;
|
||||
private final IotTcpConnectionManager connectionManager;
|
||||
private final List<ProtocolHandler> protocolHandlers;
|
||||
private final String serverId;
|
||||
|
||||
public IotTcpUpstreamHandler(IotTcpUpstreamProtocol protocol,
|
||||
IotDeviceMessageService deviceMessageService,
|
||||
IotTcpConnectionManager connectionManager,
|
||||
List<ProtocolHandler> protocolHandlers) {
|
||||
this.deviceMessageService = deviceMessageService;
|
||||
this.connectionManager = connectionManager;
|
||||
this.protocolHandlers = protocolHandlers;
|
||||
this.serverId = protocol.getServerId();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void handle(NetSocket socket) {
|
||||
String clientId = IdUtil.simpleUUID();
|
||||
log.debug("[handle][设备连接，客户端 ID: {}，地址: {}]", clientId, socket.remoteAddress());
|
||||
|
||||
// 设置异常和关闭处理器
|
||||
socket.exceptionHandler(ex -> {
|
||||
log.warn("[handle][连接异常，客户端 ID: {}，地址: {}]", clientId, socket.remoteAddress(), ex);
|
||||
cleanupConnection(socket);
|
||||
});
|
||||
|
||||
socket.closeHandler(v -> {
|
||||
log.debug("[handle][连接关闭，客户端 ID: {}，地址: {}]", clientId, socket.remoteAddress());
|
||||
cleanupConnection(socket);
|
||||
});
|
||||
|
||||
// 设置消息处理器
|
||||
socket.handler(buffer -> {
|
||||
try {
|
||||
processMessage(clientId, buffer, socket);
|
||||
} catch (Exception e) {
|
||||
log.error("[handle][消息处理失败，断开连接，客户端 ID: {}，地址: {}，错误: {}]",
|
||||
clientId, socket.remoteAddress(), e.getMessage(), e);
|
||||
cleanupConnection(socket);
|
||||
socket.close();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* 处理消息
|
||||
* <p>
|
||||
* 流程：
|
||||
* 1. 检测消息格式类型（JSON/Binary/JT808）
|
||||
* 2. 解码消息
|
||||
* 3. 查找协议处理器
|
||||
* 4. 判断是否为认证消息
|
||||
* 5. 路由到协议处理器处理
|
||||
*
|
||||
* @param clientId 客户端 ID
|
||||
* @param buffer 消息
|
||||
* @param socket 网络连接
|
||||
* @throws Exception 消息解码失败时抛出异常
|
||||
*/
|
||||
private void processMessage(String clientId, Buffer buffer, NetSocket socket) throws Exception {
|
||||
// 1. 基础检查
|
||||
if (buffer == null || buffer.length() == 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
// 2. 获取消息格式类型
|
||||
String codecType = getMessageCodecType(buffer, socket);
|
||||
|
||||
// 3. 解码消息
|
||||
IotDeviceMessage message;
|
||||
try {
|
||||
message = deviceMessageService.decodeDeviceMessage(buffer.getBytes(), codecType);
|
||||
if (message == null) {
|
||||
throw new Exception("解码后消息为空");
|
||||
}
|
||||
} catch (Exception e) {
|
||||
// 消息格式错误时抛出异常，由上层处理连接断开
|
||||
throw new Exception("消息解码失败: " + e.getMessage(), e);
|
||||
}
|
||||
|
||||
// 4. 查找协议处理器
|
||||
ProtocolHandler handler = findProtocolHandler(message, codecType);
|
||||
if (handler == null) {
|
||||
log.warn("[processMessage][未找到协议处理器，codecType: {}, method: {}]",
|
||||
codecType, message.getMethod());
|
||||
return;
|
||||
}
|
||||
|
||||
// 5. 判断是否为认证消息
|
||||
if (isAuthenticationMessage(message)) {
|
||||
handleAuthenticationWithProtocol(clientId, message, codecType, socket, handler);
|
||||
} else {
|
||||
handleBusinessWithProtocol(clientId, message, codecType, socket, handler);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 使用协议处理器处理认证消息
|
||||
* <p>
|
||||
* 认证结果处理：
|
||||
* - SUCCESS：注册连接，发送上线消息
|
||||
* - PENDING：等待后续认证步骤（如 JT808 注册后等待鉴权）
|
||||
* - FAILURE：认证失败，不做处理（协议处理器已发送失败响应）
|
||||
*
|
||||
* @param clientId 客户端 ID
|
||||
* @param message 认证消息
|
||||
* @param codecType 消息编解码类型
|
||||
* @param socket 网络连接
|
||||
* @param handler 协议处理器
|
||||
*/
|
||||
private void handleAuthenticationWithProtocol(String clientId, IotDeviceMessage message,
|
||||
String codecType, NetSocket socket,
|
||||
ProtocolHandler handler) {
|
||||
try {
|
||||
// 委托给协议处理器
|
||||
AuthResult result = handler.handleAuthentication(clientId, message, codecType, socket);
|
||||
|
||||
// 根据认证结果处理
|
||||
if (result.isSuccess()) {
|
||||
// 认证成功：注册连接并发送上线消息
|
||||
registerConnection(socket, result.getDevice(), clientId, codecType);
|
||||
sendOnlineMessage(result.getDevice());
|
||||
|
||||
log.info("[handleAuthentication][认证成功，设备: {}, 协议: {}]",
|
||||
result.getDevice().getDeviceName(), handler.getProtocolType());
|
||||
|
||||
} else if (result.isPending()) {
|
||||
// 认证待定：等待后续认证步骤（如 JT808 注册后等待鉴权）
|
||||
log.info("[handleAuthentication][认证待定，设备: {}, 协议: {}, 消息: {}]",
|
||||
result.getDevice() != null ? result.getDevice().getDeviceName() : "unknown",
|
||||
handler.getProtocolType(), result.getMessage());
|
||||
|
||||
} else {
|
||||
// 认证失败：协议处理器已发送失败响应，这里只记录日志
|
||||
log.warn("[handleAuthentication][认证失败，clientId: {}, 协议: {}, 原因: {}]",
|
||||
clientId, handler.getProtocolType(), result.getMessage());
|
||||
}
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("[handleAuthentication][认证异常，clientId: {}, 协议: {}]",
|
||||
clientId, handler.getProtocolType(), e);
|
||||
handler.sendResponse(socket, message, false, "认证异常", codecType);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 使用协议处理器处理业务消息
|
||||
* <p>
|
||||
* 前置条件：设备已认证
|
||||
*
|
||||
* @param clientId 客户端 ID
|
||||
* @param message 业务消息
|
||||
* @param codecType 消息编解码类型
|
||||
* @param socket 网络连接
|
||||
* @param handler 协议处理器
|
||||
*/
|
||||
private void handleBusinessWithProtocol(String clientId, IotDeviceMessage message,
|
||||
String codecType, NetSocket socket,
|
||||
ProtocolHandler handler) {
|
||||
try {
|
||||
// 1. 检查认证状态
|
||||
if (connectionManager.isNotAuthenticated(socket)) {
|
||||
log.warn("[handleBusinessMessage][设备未认证，clientId: {}]", clientId);
|
||||
handler.sendResponse(socket, message, false, "请先进行认证", codecType);
|
||||
return;
|
||||
}
|
||||
|
||||
// 2. 获取连接信息
|
||||
IotTcpConnectionManager.ConnectionInfo connectionInfo = connectionManager.getConnectionInfo(socket);
|
||||
if (connectionInfo == null) {
|
||||
log.error("[handleBusinessMessage][连接信息不存在，clientId: {}]", clientId);
|
||||
return;
|
||||
}
|
||||
|
||||
// 3. 委托给协议处理器处理业务消息
|
||||
handler.handleBusinessMessage(
|
||||
clientId,
|
||||
message,
|
||||
codecType,
|
||||
socket,
|
||||
connectionInfo.getProductKey(),
|
||||
connectionInfo.getDeviceName(),
|
||||
serverId
|
||||
);
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("[handleBusinessMessage][业务消息处理异常，clientId: {}]", clientId, e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取消息编解码类型
|
||||
* <p>
|
||||
* 检测优先级：
|
||||
* 1. 如果已认证，使用缓存的编解码类型
|
||||
* 2. 未认证时，通过消息格式自动检测：
|
||||
* - JT808：首尾标识符 0x7e
|
||||
* - Binary：魔术字 0x7E
|
||||
* - JSON：默认
|
||||
*
|
||||
* @param buffer 消息
|
||||
* @param socket 网络连接
|
||||
* @return 消息编解码类型
|
||||
*/
|
||||
private String getMessageCodecType(Buffer buffer, NetSocket socket) {
|
||||
// 1. 如果已认证，优先使用缓存的编解码类型
|
||||
IotTcpConnectionManager.ConnectionInfo connectionInfo = connectionManager.getConnectionInfo(socket);
|
||||
if (connectionInfo != null && connectionInfo.isAuthenticated() &&
|
||||
StrUtil.isNotBlank(connectionInfo.getCodecType())) {
|
||||
return connectionInfo.getCodecType();
|
||||
}
|
||||
|
||||
// 2. 未认证时检测消息格式类型
|
||||
byte[] data = buffer.getBytes();
|
||||
|
||||
// 2.1 检测是否为 JT808 格式（首尾标识符 0x7e）
|
||||
if (IotJt808DeviceMessageCodec.isJt808Format(data)) {
|
||||
return CODEC_TYPE_JT808;
|
||||
}
|
||||
|
||||
// 2.2 检测是否为自定义二进制格式（魔术字 0x7E）
|
||||
if (IotTcpBinaryDeviceMessageCodec.isBinaryFormatQuick(data)) {
|
||||
return CODEC_TYPE_BINARY;
|
||||
}
|
||||
|
||||
// 2.3 默认为 JSON 格式
|
||||
return CODEC_TYPE_JSON;
|
||||
}
|
||||
|
||||
/**
|
||||
* 查找协议处理器
|
||||
* <p>
|
||||
* 遍历所有协议处理器，返回第一个能处理该消息的处理器
|
||||
*
|
||||
* @param message 设备消息
|
||||
* @param codecType 消息编解码类型
|
||||
* @return 协议处理器，未找到返回 null
|
||||
*/
|
||||
private ProtocolHandler findProtocolHandler(IotDeviceMessage message, String codecType) {
|
||||
return protocolHandlers.stream()
|
||||
.filter(handler -> handler.canHandle(message, codecType))
|
||||
.findFirst()
|
||||
.orElse(null);
|
||||
}
|
||||
|
||||
/**
|
||||
* 判断是否为认证消息
|
||||
* <p>
|
||||
* 认证消息包括：
|
||||
* - 标准认证：auth
|
||||
* - JT808 注册：jt808.terminal.register
|
||||
* - JT808 鉴权：jt808.terminal.auth
|
||||
*
|
||||
* @param message 设备消息
|
||||
* @return true-是认证消息，false-不是
|
||||
*/
|
||||
private boolean isAuthenticationMessage(IotDeviceMessage message) {
|
||||
String method = message.getMethod();
|
||||
return "auth".equals(method)
|
||||
|| "jt808.terminal.register".equals(method)
|
||||
|| "jt808.terminal.auth".equals(method);
|
||||
}
|
||||
|
||||
/**
|
||||
* 注册连接信息
|
||||
* <p>
|
||||
* 将设备连接信息注册到连接管理器，包括：
|
||||
* - 设备 ID
|
||||
* - 产品 Key
|
||||
* - 设备名称
|
||||
* - 客户端 ID
|
||||
* - 编解码类型
|
||||
* - 认证状态
|
||||
*
|
||||
* @param socket 网络连接
|
||||
* @param device 设备信息
|
||||
* @param clientId 客户端 ID
|
||||
* @param codecType 消息编解码类型
|
||||
*/
|
||||
private void registerConnection(NetSocket socket, IotDeviceRespDTO device,
|
||||
String clientId, String codecType) {
|
||||
IotTcpConnectionManager.ConnectionInfo connectionInfo = new IotTcpConnectionManager.ConnectionInfo()
|
||||
.setDeviceId(device.getId())
|
||||
.setProductKey(device.getProductKey())
|
||||
.setDeviceName(device.getDeviceName())
|
||||
.setClientId(clientId)
|
||||
.setCodecType(codecType)
|
||||
.setAuthenticated(true);
|
||||
|
||||
// 注册连接（如果设备已有其他连接，会自动断开旧连接）
|
||||
connectionManager.registerConnection(socket, device.getId(), connectionInfo);
|
||||
}
|
||||
|
||||
/**
|
||||
* 发送设备上线消息
|
||||
* <p>
|
||||
* 设备认证成功后，发送上线消息到消息总线，通知业务层设备已上线
|
||||
*
|
||||
* @param device 设备信息
|
||||
*/
|
||||
private void sendOnlineMessage(IotDeviceRespDTO device) {
|
||||
try {
|
||||
IotDeviceMessage onlineMessage = IotDeviceMessage.buildStateUpdateOnline();
|
||||
deviceMessageService.sendDeviceMessage(onlineMessage, device.getProductKey(),
|
||||
device.getDeviceName(), serverId);
|
||||
|
||||
log.debug("[sendOnlineMessage][发送上线消息成功，设备: {}]", device.getDeviceName());
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("[sendOnlineMessage][发送上线消息失败，设备: {}]", device.getDeviceName(), e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 清理连接
|
||||
* <p>
|
||||
* 连接关闭或异常时，清理连接信息并发送离线消息
|
||||
*
|
||||
* @param socket 网络连接
|
||||
*/
|
||||
private void cleanupConnection(NetSocket socket) {
|
||||
try {
|
||||
// 1. 发送离线消息（如果已认证）
|
||||
IotTcpConnectionManager.ConnectionInfo connectionInfo = connectionManager.getConnectionInfo(socket);
|
||||
if (connectionInfo != null && connectionInfo.isAuthenticated()) {
|
||||
IotDeviceMessage offlineMessage = IotDeviceMessage.buildStateOffline();
|
||||
deviceMessageService.sendDeviceMessage(offlineMessage, connectionInfo.getProductKey(),
|
||||
connectionInfo.getDeviceName(), serverId);
|
||||
|
||||
log.debug("[cleanupConnection][发送离线消息成功，设备: {}]", connectionInfo.getDeviceName());
|
||||
}
|
||||
|
||||
// 2. 注销连接
|
||||
connectionManager.unregisterConnection(socket);
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("[cleanupConnection][清理连接失败]", e);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
package com.viewsh.module.iot.gateway.protocol.tcp.router;
|
||||
|
||||
import cn.hutool.core.util.IdUtil;
|
||||
import cn.hutool.core.util.StrUtil;
|
||||
import com.viewsh.module.iot.core.biz.dto.IotDeviceRespDTO;
|
||||
import com.viewsh.module.iot.core.mq.message.IotDeviceMessage;
|
||||
import com.viewsh.module.iot.gateway.codec.jt808.IotJt808DeviceMessageCodec;
|
||||
import com.viewsh.module.iot.gateway.codec.tcp.IotTcpBinaryDeviceMessageCodec;
|
||||
import com.viewsh.module.iot.gateway.codec.tcp.IotTcpJsonDeviceMessageCodec;
|
||||
import com.viewsh.module.iot.gateway.protocol.tcp.IotTcpUpstreamProtocol;
|
||||
import com.viewsh.module.iot.gateway.protocol.tcp.handler.AuthResult;
|
||||
import com.viewsh.module.iot.gateway.protocol.tcp.handler.ProtocolHandler;
|
||||
import com.viewsh.module.iot.gateway.protocol.tcp.manager.IotTcpConnectionManager;
|
||||
import com.viewsh.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 lombok.extern.slf4j.Slf4j;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* TCP 上行消息处理器（重构版）
|
||||
* <p>
|
||||
* 职责：
|
||||
* 1. 管理 TCP 连接生命周期（连接建立、异常、关闭）
|
||||
* 2. 检测消息格式类型（JSON/Binary/JT808）
|
||||
* 3. 解码设备消息
|
||||
* 4. 路由到对应的协议处理器
|
||||
* 5. 管理设备认证状态
|
||||
* 6. 发送设备上线/离线消息
|
||||
* <p>
|
||||
* 设计原则：
|
||||
* - 主处理器只负责路由，不包含协议特定逻辑
|
||||
* - 协议处理器通过 Spring 自动注入，实现插件化
|
||||
* - 认证成功后，统一注册连接和发送上线消息
|
||||
* - 支持多步认证（如 JT808 的注册+鉴权）
|
||||
*
|
||||
* @author 芋道源码
|
||||
*/
|
||||
@Slf4j
|
||||
public class IotTcpUpstreamHandler implements Handler<NetSocket> {
|
||||
|
||||
private static final String CODEC_TYPE_JSON = IotTcpJsonDeviceMessageCodec.TYPE;
|
||||
private static final String CODEC_TYPE_BINARY = IotTcpBinaryDeviceMessageCodec.TYPE;
|
||||
private static final String CODEC_TYPE_JT808 = IotJt808DeviceMessageCodec.TYPE;
|
||||
|
||||
private final IotDeviceMessageService deviceMessageService;
|
||||
private final IotTcpConnectionManager connectionManager;
|
||||
private final List<ProtocolHandler> protocolHandlers;
|
||||
private final String serverId;
|
||||
|
||||
public IotTcpUpstreamHandler(IotTcpUpstreamProtocol protocol,
|
||||
IotDeviceMessageService deviceMessageService,
|
||||
IotTcpConnectionManager connectionManager,
|
||||
List<ProtocolHandler> protocolHandlers) {
|
||||
this.deviceMessageService = deviceMessageService;
|
||||
this.connectionManager = connectionManager;
|
||||
this.protocolHandlers = protocolHandlers;
|
||||
this.serverId = protocol.getServerId();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void handle(NetSocket socket) {
|
||||
String clientId = IdUtil.simpleUUID();
|
||||
log.debug("[handle][设备连接，客户端 ID: {}，地址: {}]", clientId, socket.remoteAddress());
|
||||
|
||||
// 设置异常和关闭处理器
|
||||
socket.exceptionHandler(ex -> {
|
||||
log.warn("[handle][连接异常，客户端 ID: {}，地址: {}]", clientId, socket.remoteAddress(), ex);
|
||||
cleanupConnection(socket);
|
||||
});
|
||||
|
||||
socket.closeHandler(v -> {
|
||||
log.debug("[handle][连接关闭，客户端 ID: {}，地址: {}]", clientId, socket.remoteAddress());
|
||||
cleanupConnection(socket);
|
||||
});
|
||||
|
||||
// 设置消息处理器
|
||||
socket.handler(buffer -> {
|
||||
try {
|
||||
processMessage(clientId, buffer, socket);
|
||||
} catch (Exception e) {
|
||||
log.error("[handle][消息处理失败，断开连接，客户端 ID: {}，地址: {}，错误: {}]",
|
||||
clientId, socket.remoteAddress(), e.getMessage(), e);
|
||||
cleanupConnection(socket);
|
||||
socket.close();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* 处理消息
|
||||
* <p>
|
||||
* 流程：
|
||||
* 1. 检测消息格式类型（JSON/Binary/JT808）
|
||||
* 2. 解码消息
|
||||
* 3. 查找协议处理器
|
||||
* 4. 判断是否为认证消息
|
||||
* 5. 路由到协议处理器处理
|
||||
*
|
||||
* @param clientId 客户端 ID
|
||||
* @param buffer 消息
|
||||
* @param socket 网络连接
|
||||
* @throws Exception 消息解码失败时抛出异常
|
||||
*/
|
||||
private void processMessage(String clientId, Buffer buffer, NetSocket socket) throws Exception {
|
||||
// 1. 基础检查
|
||||
if (buffer == null || buffer.length() == 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
// 2. 获取消息格式类型
|
||||
String codecType = getMessageCodecType(buffer, socket);
|
||||
if (codecType == null) {
|
||||
log.warn("[processMessage][未知消息格式，断开连接，clientId: {}，数据开头: {}]",
|
||||
clientId, buffer.length() > 20 ? buffer.getString(0, 20) : buffer.toString());
|
||||
socket.close();
|
||||
return;
|
||||
}
|
||||
|
||||
// 3. 解码消息
|
||||
IotDeviceMessage message;
|
||||
try {
|
||||
message = deviceMessageService.decodeDeviceMessage(buffer.getBytes(), codecType);
|
||||
if (message == null) {
|
||||
throw new Exception("解码后消息为空");
|
||||
}
|
||||
} catch (Exception e) {
|
||||
// 消息格式错误时抛出异常，由上层处理连接断开
|
||||
throw new Exception("消息解码失败: " + e.getMessage(), e);
|
||||
}
|
||||
|
||||
// 4. 查找协议处理器
|
||||
ProtocolHandler handler = findProtocolHandler(message, codecType);
|
||||
if (handler == null) {
|
||||
log.warn("[processMessage][未找到协议处理器，codecType: {}, method: {}]",
|
||||
codecType, message.getMethod());
|
||||
return;
|
||||
}
|
||||
|
||||
// 5. 判断是否为认证消息
|
||||
if (isAuthenticationMessage(message)) {
|
||||
handleAuthenticationWithProtocol(clientId, message, codecType, socket, handler);
|
||||
} else {
|
||||
handleBusinessWithProtocol(clientId, message, codecType, socket, handler);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 使用协议处理器处理认证消息
|
||||
* <p>
|
||||
* 认证结果处理：
|
||||
* - SUCCESS：注册连接，发送上线消息
|
||||
* - PENDING：等待后续认证步骤（如 JT808 注册后等待鉴权）
|
||||
* - FAILURE：认证失败，不做处理（协议处理器已发送失败响应）
|
||||
*
|
||||
* @param clientId 客户端 ID
|
||||
* @param message 认证消息
|
||||
* @param codecType 消息编解码类型
|
||||
* @param socket 网络连接
|
||||
* @param handler 协议处理器
|
||||
*/
|
||||
private void handleAuthenticationWithProtocol(String clientId, IotDeviceMessage message,
|
||||
String codecType, NetSocket socket,
|
||||
ProtocolHandler handler) {
|
||||
try {
|
||||
// 委托给协议处理器
|
||||
AuthResult result = handler.handleAuthentication(clientId, message, codecType, socket);
|
||||
|
||||
// 根据认证结果处理
|
||||
if (result.isSuccess()) {
|
||||
// 认证成功：注册连接并发送上线消息
|
||||
registerConnection(socket, result.getDevice(), clientId, codecType);
|
||||
sendOnlineMessage(result.getDevice());
|
||||
|
||||
log.info("[handleAuthentication][认证成功，设备: {}, 协议: {}]",
|
||||
result.getDevice().getDeviceName(), handler.getProtocolType());
|
||||
|
||||
} else if (result.isPending()) {
|
||||
// 认证待定：等待后续认证步骤（如 JT808 注册后等待鉴权）
|
||||
log.info("[handleAuthentication][认证待定，设备: {}, 协议: {}, 消息: {}]",
|
||||
result.getDevice() != null ? result.getDevice().getDeviceName() : "unknown",
|
||||
handler.getProtocolType(), result.getMessage());
|
||||
|
||||
} else {
|
||||
// 认证失败：协议处理器已发送失败响应，这里只记录日志
|
||||
log.warn("[handleAuthentication][认证失败，clientId: {}, 协议: {}, 原因: {}]",
|
||||
clientId, handler.getProtocolType(), result.getMessage());
|
||||
}
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("[handleAuthentication][认证异常，clientId: {}, 协议: {}]",
|
||||
clientId, handler.getProtocolType(), e);
|
||||
handler.sendResponse(socket, message, false, "认证异常", codecType);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 使用协议处理器处理业务消息
|
||||
* <p>
|
||||
* 前置条件：设备已认证
|
||||
*
|
||||
* @param clientId 客户端 ID
|
||||
* @param message 业务消息
|
||||
* @param codecType 消息编解码类型
|
||||
* @param socket 网络连接
|
||||
* @param handler 协议处理器
|
||||
*/
|
||||
private void handleBusinessWithProtocol(String clientId, IotDeviceMessage message,
|
||||
String codecType, NetSocket socket,
|
||||
ProtocolHandler handler) {
|
||||
try {
|
||||
// 1. 检查认证状态
|
||||
if (connectionManager.isNotAuthenticated(socket)) {
|
||||
log.warn("[handleBusinessMessage][设备未认证，clientId: {}]", clientId);
|
||||
handler.sendResponse(socket, message, false, "请先进行认证", codecType);
|
||||
return;
|
||||
}
|
||||
|
||||
// 2. 获取连接信息
|
||||
IotTcpConnectionManager.ConnectionInfo connectionInfo = connectionManager.getConnectionInfo(socket);
|
||||
if (connectionInfo == null) {
|
||||
log.error("[handleBusinessMessage][连接信息不存在，clientId: {}]", clientId);
|
||||
return;
|
||||
}
|
||||
|
||||
// 3. 委托给协议处理器处理业务消息
|
||||
handler.handleBusinessMessage(
|
||||
clientId,
|
||||
message,
|
||||
codecType,
|
||||
socket,
|
||||
connectionInfo.getProductKey(),
|
||||
connectionInfo.getDeviceName(),
|
||||
serverId);
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("[handleBusinessMessage][业务消息处理异常，clientId: {}]", clientId, e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取消息编解码类型
|
||||
* <p>
|
||||
* 检测优先级：
|
||||
* 1. 如果已认证，使用缓存的编解码类型
|
||||
* 2. 未认证时，通过消息格式自动检测：
|
||||
* - JT808：首尾标识符 0x7e
|
||||
* - Binary：魔术字 0x7E
|
||||
* - JSON：默认
|
||||
*
|
||||
* @param buffer 消息
|
||||
* @param socket 网络连接
|
||||
* @return 消息编解码类型
|
||||
*/
|
||||
private String getMessageCodecType(Buffer buffer, NetSocket socket) {
|
||||
// 1. 如果已认证，优先使用缓存的编解码类型
|
||||
IotTcpConnectionManager.ConnectionInfo connectionInfo = connectionManager.getConnectionInfo(socket);
|
||||
if (connectionInfo != null && connectionInfo.isAuthenticated() &&
|
||||
StrUtil.isNotBlank(connectionInfo.getCodecType())) {
|
||||
return connectionInfo.getCodecType();
|
||||
}
|
||||
|
||||
// 2. 未认证时检测消息格式类型
|
||||
byte[] data = buffer.getBytes();
|
||||
|
||||
// 2.1 检测是否为 JT808 格式（首尾标识符 0x7e）
|
||||
if (IotJt808DeviceMessageCodec.isJt808Format(data)) {
|
||||
return CODEC_TYPE_JT808;
|
||||
}
|
||||
|
||||
// 2.2 检测是否为自定义二进制格式（魔术字 0x7E）
|
||||
if (IotTcpBinaryDeviceMessageCodec.isBinaryFormatQuick(data)) {
|
||||
return CODEC_TYPE_BINARY;
|
||||
}
|
||||
|
||||
// 2.3 检测是否为 JSON 格式（以 { 或 [ 开头）
|
||||
String jsonStr = StrUtil.utf8Str(data).trim();
|
||||
if (StrUtil.startWithAny(jsonStr, "{", "[")) {
|
||||
return CODEC_TYPE_JSON;
|
||||
}
|
||||
|
||||
// 2.4 未知格式
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* 查找协议处理器
|
||||
* <p>
|
||||
* 遍历所有协议处理器，返回第一个能处理该消息的处理器
|
||||
*
|
||||
* @param message 设备消息
|
||||
* @param codecType 消息编解码类型
|
||||
* @return 协议处理器，未找到返回 null
|
||||
*/
|
||||
private ProtocolHandler findProtocolHandler(IotDeviceMessage message, String codecType) {
|
||||
return protocolHandlers.stream()
|
||||
.filter(handler -> handler.canHandle(message, codecType))
|
||||
.findFirst()
|
||||
.orElse(null);
|
||||
}
|
||||
|
||||
/**
|
||||
* 判断是否为认证消息
|
||||
* <p>
|
||||
* 认证消息包括：
|
||||
* - 标准认证：auth
|
||||
* - JT808 注册：jt808.terminal.register
|
||||
* - JT808 鉴权：jt808.terminal.auth
|
||||
*
|
||||
* @param message 设备消息
|
||||
* @return true-是认证消息，false-不是
|
||||
*/
|
||||
private boolean isAuthenticationMessage(IotDeviceMessage message) {
|
||||
String method = message.getMethod();
|
||||
return "auth".equals(method)
|
||||
|| "jt808.terminal.register".equals(method)
|
||||
|| "jt808.terminal.auth".equals(method);
|
||||
}
|
||||
|
||||
/**
|
||||
* 注册连接信息
|
||||
* <p>
|
||||
* 将设备连接信息注册到连接管理器，包括：
|
||||
* - 设备 ID
|
||||
* - 产品 Key
|
||||
* - 设备名称
|
||||
* - 客户端 ID
|
||||
* - 编解码类型
|
||||
* - 认证状态
|
||||
*
|
||||
* @param socket 网络连接
|
||||
* @param device 设备信息
|
||||
* @param clientId 客户端 ID
|
||||
* @param codecType 消息编解码类型
|
||||
*/
|
||||
private void registerConnection(NetSocket socket, IotDeviceRespDTO device,
|
||||
String clientId, String codecType) {
|
||||
IotTcpConnectionManager.ConnectionInfo connectionInfo = new IotTcpConnectionManager.ConnectionInfo()
|
||||
.setDeviceId(device.getId())
|
||||
.setProductKey(device.getProductKey())
|
||||
.setDeviceName(device.getDeviceName())
|
||||
.setClientId(clientId)
|
||||
.setCodecType(codecType)
|
||||
.setAuthenticated(true);
|
||||
|
||||
// 注册连接（如果设备已有其他连接，会自动断开旧连接）
|
||||
connectionManager.registerConnection(socket, device.getId(), connectionInfo);
|
||||
}
|
||||
|
||||
/**
|
||||
* 发送设备上线消息
|
||||
* <p>
|
||||
* 设备认证成功后，发送上线消息到消息总线，通知业务层设备已上线
|
||||
*
|
||||
* @param device 设备信息
|
||||
*/
|
||||
private void sendOnlineMessage(IotDeviceRespDTO device) {
|
||||
try {
|
||||
IotDeviceMessage onlineMessage = IotDeviceMessage.buildStateUpdateOnline();
|
||||
deviceMessageService.sendDeviceMessage(onlineMessage, device.getProductKey(),
|
||||
device.getDeviceName(), serverId);
|
||||
|
||||
log.debug("[sendOnlineMessage][发送上线消息成功，设备: {}]", device.getDeviceName());
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("[sendOnlineMessage][发送上线消息失败，设备: {}]", device.getDeviceName(), e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 清理连接
|
||||
* <p>
|
||||
* 连接关闭或异常时，清理连接信息并发送离线消息
|
||||
*
|
||||
* @param socket 网络连接
|
||||
*/
|
||||
private void cleanupConnection(NetSocket socket) {
|
||||
try {
|
||||
// 1. 发送离线消息（如果已认证）
|
||||
IotTcpConnectionManager.ConnectionInfo connectionInfo = connectionManager.getConnectionInfo(socket);
|
||||
if (connectionInfo != null && connectionInfo.isAuthenticated()) {
|
||||
IotDeviceMessage offlineMessage = IotDeviceMessage.buildStateOffline();
|
||||
deviceMessageService.sendDeviceMessage(offlineMessage, connectionInfo.getProductKey(),
|
||||
connectionInfo.getDeviceName(), serverId);
|
||||
|
||||
log.debug("[cleanupConnection][发送离线消息成功，设备: {}]", connectionInfo.getDeviceName());
|
||||
}
|
||||
|
||||
// 2. 注销连接
|
||||
connectionManager.unregisterConnection(socket);
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("[cleanupConnection][清理连接失败]", e);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -1,75 +1,75 @@
|
||||
package com.viewsh.module.iot.api.device;
|
||||
|
||||
import com.viewsh.framework.common.enums.RpcConstants;
|
||||
import com.viewsh.framework.common.pojo.CommonResult;
|
||||
import com.viewsh.framework.common.util.object.BeanUtils;
|
||||
import com.viewsh.module.iot.core.biz.IotDeviceCommonApi;
|
||||
import com.viewsh.module.iot.core.biz.dto.IotDeviceAuthReqDTO;
|
||||
import com.viewsh.module.iot.core.biz.dto.IotDeviceGetReqDTO;
|
||||
import com.viewsh.module.iot.core.biz.dto.IotDeviceRespDTO;
|
||||
import com.viewsh.module.iot.dal.dataobject.device.IotDeviceDO;
|
||||
import com.viewsh.module.iot.dal.dataobject.product.IotProductDO;
|
||||
import com.viewsh.module.iot.service.device.IotDeviceService;
|
||||
import com.viewsh.module.iot.service.product.IotProductService;
|
||||
import jakarta.annotation.Resource;
|
||||
import jakarta.annotation.security.PermitAll;
|
||||
import org.springframework.context.annotation.Primary;
|
||||
import org.springframework.validation.annotation.Validated;
|
||||
import org.springframework.web.bind.annotation.PostMapping;
|
||||
import org.springframework.web.bind.annotation.RequestBody;
|
||||
import org.springframework.web.bind.annotation.RestController;
|
||||
|
||||
import static com.viewsh.framework.common.pojo.CommonResult.success;
|
||||
|
||||
/**
|
||||
* IoT 设备 API 实现类
|
||||
*
|
||||
* @author haohao
|
||||
*/
|
||||
@RestController
|
||||
@Validated
|
||||
@Primary // 保证优先匹配，因为 viewsh-iot-gateway 也有 IotDeviceCommonApi 的实现，并且也可能会被 biz 引入
|
||||
public class IoTDeviceApiImpl implements IotDeviceCommonApi {
|
||||
|
||||
@Resource
|
||||
private IotDeviceService deviceService;
|
||||
@Resource
|
||||
private IotProductService productService;
|
||||
|
||||
@Override
|
||||
@PostMapping(RpcConstants.RPC_API_PREFIX + "/iot/device/auth")
|
||||
@PermitAll
|
||||
public CommonResult<Boolean> authDevice(@RequestBody IotDeviceAuthReqDTO authReqDTO) {
|
||||
return success(deviceService.authDevice(authReqDTO));
|
||||
}
|
||||
|
||||
@Override
|
||||
@PostMapping(RpcConstants.RPC_API_PREFIX + "/iot/device/get") // 特殊：方便调用，暂时使用 POST，实际更推荐 GET
|
||||
@PermitAll
|
||||
public CommonResult<IotDeviceRespDTO> getDevice(@RequestBody IotDeviceGetReqDTO getReqDTO) {
|
||||
IotDeviceDO device;
|
||||
|
||||
// 查询优先级：id > (productKey + deviceName) > deviceName
|
||||
if (getReqDTO.getId() != null) {
|
||||
// 通过设备 ID 查询
|
||||
device = deviceService.getDeviceFromCache(getReqDTO.getId());
|
||||
} else if (getReqDTO.getProductKey() != null && getReqDTO.getDeviceName() != null) {
|
||||
// 通过 productKey + deviceName 查询
|
||||
device = deviceService.getDeviceFromCache(getReqDTO.getProductKey(), getReqDTO.getDeviceName());
|
||||
} else if (getReqDTO.getDeviceName() != null) {
|
||||
// 仅通过 deviceName 查询（用于 JT808 等协议，终端手机号应该是全局唯一的）
|
||||
device = deviceService.getDeviceFromCacheByDeviceName(getReqDTO.getDeviceName());
|
||||
} else {
|
||||
device = null;
|
||||
}
|
||||
|
||||
return success(BeanUtils.toBean(device, IotDeviceRespDTO.class, deviceDTO -> {
|
||||
IotProductDO product = productService.getProductFromCache(deviceDTO.getProductId());
|
||||
if (product != null) {
|
||||
deviceDTO.setCodecType(product.getCodecType());
|
||||
deviceDTO.setProductAuthType(product.getAuthType());
|
||||
}
|
||||
}));
|
||||
}
|
||||
|
||||
package com.viewsh.module.iot.api.device;
|
||||
|
||||
import com.viewsh.framework.common.enums.RpcConstants;
|
||||
import com.viewsh.framework.common.pojo.CommonResult;
|
||||
import com.viewsh.framework.common.util.object.BeanUtils;
|
||||
import com.viewsh.module.iot.core.biz.IotDeviceCommonApi;
|
||||
import com.viewsh.module.iot.core.biz.dto.IotDeviceAuthReqDTO;
|
||||
import com.viewsh.module.iot.core.biz.dto.IotDeviceGetReqDTO;
|
||||
import com.viewsh.module.iot.core.biz.dto.IotDeviceRespDTO;
|
||||
import com.viewsh.module.iot.dal.dataobject.device.IotDeviceDO;
|
||||
import com.viewsh.module.iot.dal.dataobject.product.IotProductDO;
|
||||
import com.viewsh.module.iot.service.device.IotDeviceService;
|
||||
import com.viewsh.module.iot.service.product.IotProductService;
|
||||
import jakarta.annotation.Resource;
|
||||
import jakarta.annotation.security.PermitAll;
|
||||
import org.springframework.context.annotation.Primary;
|
||||
import org.springframework.validation.annotation.Validated;
|
||||
import org.springframework.web.bind.annotation.PostMapping;
|
||||
import org.springframework.web.bind.annotation.RequestBody;
|
||||
import org.springframework.web.bind.annotation.RestController;
|
||||
|
||||
import static com.viewsh.framework.common.pojo.CommonResult.success;
|
||||
|
||||
/**
|
||||
* IoT 设备 API 实现类（供IOT GATEWAY使用）
|
||||
*
|
||||
* @author haohao
|
||||
*/
|
||||
@RestController
|
||||
@Validated
|
||||
@Primary // 保证优先匹配，因为 viewsh-iot-gateway 也有 IotDeviceCommonApi 的实现，并且也可能会被 biz 引入
|
||||
public class IoTDeviceApiImpl implements IotDeviceCommonApi {
|
||||
|
||||
@Resource
|
||||
private IotDeviceService deviceService;
|
||||
@Resource
|
||||
private IotProductService productService;
|
||||
|
||||
@Override
|
||||
@PostMapping(RpcConstants.RPC_API_PREFIX + "/iot/device/auth")
|
||||
@PermitAll
|
||||
public CommonResult<Boolean> authDevice(@RequestBody IotDeviceAuthReqDTO authReqDTO) {
|
||||
return success(deviceService.authDevice(authReqDTO));
|
||||
}
|
||||
|
||||
@Override
|
||||
@PostMapping(RpcConstants.RPC_API_PREFIX + "/iot/device/get") // 特殊：方便调用，暂时使用 POST，实际更推荐 GET
|
||||
@PermitAll
|
||||
public CommonResult<IotDeviceRespDTO> getDevice(@RequestBody IotDeviceGetReqDTO getReqDTO) {
|
||||
IotDeviceDO device;
|
||||
|
||||
// 查询优先级：id > (productKey + deviceName) > deviceName
|
||||
if (getReqDTO.getId() != null) {
|
||||
// 通过设备 ID 查询
|
||||
device = deviceService.getDeviceFromCache(getReqDTO.getId());
|
||||
} else if (getReqDTO.getProductKey() != null && getReqDTO.getDeviceName() != null) {
|
||||
// 通过 productKey + deviceName 查询
|
||||
device = deviceService.getDeviceFromCache(getReqDTO.getProductKey(), getReqDTO.getDeviceName());
|
||||
} else if (getReqDTO.getDeviceName() != null) {
|
||||
// 仅通过 deviceName 查询（用于 JT808 等协议，终端手机号应该是全局唯一的）
|
||||
device = deviceService.getDeviceFromCacheByDeviceName(getReqDTO.getDeviceName());
|
||||
} else {
|
||||
device = null;
|
||||
}
|
||||
|
||||
return success(BeanUtils.toBean(device, IotDeviceRespDTO.class, deviceDTO -> {
|
||||
IotProductDO product = productService.getProductFromCache(deviceDTO.getProductId());
|
||||
if (product != null) {
|
||||
deviceDTO.setCodecType(product.getCodecType());
|
||||
deviceDTO.setProductAuthType(product.getAuthType());
|
||||
}
|
||||
}));
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,123 @@
|
||||
package com.viewsh.module.iot.api.device;
|
||||
|
||||
import cn.hutool.core.map.MapUtil;
|
||||
import com.viewsh.framework.common.pojo.CommonResult;
|
||||
import com.viewsh.module.iot.api.device.dto.IotDeviceServiceInvokeReqDTO;
|
||||
import com.viewsh.module.iot.api.device.dto.IotDeviceServiceInvokeRespDTO;
|
||||
import com.viewsh.module.iot.core.enums.IotDeviceMessageMethodEnum;
|
||||
import com.viewsh.module.iot.core.enums.IotDeviceStateEnum;
|
||||
import com.viewsh.module.iot.core.mq.message.IotDeviceMessage;
|
||||
import com.viewsh.module.iot.dal.dataobject.device.IotDeviceDO;
|
||||
import com.viewsh.module.iot.service.device.IotDeviceService;
|
||||
import com.viewsh.module.iot.service.device.message.IotDeviceMessageService;
|
||||
import io.swagger.v3.oas.annotations.Operation;
|
||||
import jakarta.annotation.Resource;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.context.annotation.Primary;
|
||||
import org.springframework.validation.annotation.Validated;
|
||||
import org.springframework.web.bind.annotation.PostMapping;
|
||||
import org.springframework.web.bind.annotation.RequestBody;
|
||||
import org.springframework.web.bind.annotation.RestController;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
import static com.viewsh.framework.common.pojo.CommonResult.success;
|
||||
|
||||
/**
|
||||
* IoT 设备控制 API 实现类
|
||||
* <p>
|
||||
* 提供 RPC 接口供其他模块调用 IoT 设备服务
|
||||
*
|
||||
* @author lzh
|
||||
*/
|
||||
@RestController
|
||||
@Validated
|
||||
@Primary
|
||||
@Slf4j
|
||||
public class IotDeviceControlApiImpl implements IotDeviceControlApi {
|
||||
|
||||
@Resource
|
||||
private IotDeviceService deviceService;
|
||||
|
||||
@Resource
|
||||
private IotDeviceMessageService deviceMessageService;
|
||||
|
||||
@Override
|
||||
@PostMapping(PREFIX + "/invoke-service")
|
||||
@Operation(summary = "调用设备服务")
|
||||
public CommonResult<IotDeviceServiceInvokeRespDTO> invokeService(@RequestBody IotDeviceServiceInvokeReqDTO reqDTO) {
|
||||
try {
|
||||
// 1. 获取设备信息
|
||||
IotDeviceDO device = deviceService.getDeviceFromCache(reqDTO.getDeviceId());
|
||||
if (device == null) {
|
||||
return success(IotDeviceServiceInvokeRespDTO.builder()
|
||||
.success(false)
|
||||
.code(404)
|
||||
.errorMsg("设备不存在")
|
||||
.responseTime(LocalDateTime.now())
|
||||
.build());
|
||||
}
|
||||
|
||||
// 2. 检查设备是否在线
|
||||
if (!IotDeviceStateEnum.ONLINE.getState().equals(device.getState())) {
|
||||
return success(IotDeviceServiceInvokeRespDTO.builder()
|
||||
.success(false)
|
||||
.code(400)
|
||||
.errorMsg("设备不在线，当前状态: " + device.getState())
|
||||
.responseTime(LocalDateTime.now())
|
||||
.build());
|
||||
}
|
||||
|
||||
// 3. 构建服务调用消息
|
||||
@SuppressWarnings("unchecked")
|
||||
Map<String, Object> params = (Map<String, Object>) (Map<?, ?>) MapUtil.builder()
|
||||
.put("identifier", reqDTO.getIdentifier())
|
||||
.put("params", reqDTO.getParams() != null ? reqDTO.getParams() : new HashMap<>())
|
||||
.build();
|
||||
|
||||
IotDeviceMessage message = IotDeviceMessage.requestOf(
|
||||
IotDeviceMessageMethodEnum.SERVICE_INVOKE.getMethod(), params);
|
||||
|
||||
// 4. 发送消息
|
||||
IotDeviceMessage result = deviceMessageService.sendDeviceMessage(message, device);
|
||||
|
||||
// 5. 返回结果（异步模式，立即返回消息ID）
|
||||
return success(IotDeviceServiceInvokeRespDTO.builder()
|
||||
.messageId(result.getId())
|
||||
.success(true)
|
||||
.data(null)
|
||||
.responseTime(LocalDateTime.now())
|
||||
.build());
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("[invokeService] 设备服务调用失败: deviceId={}, identifier={}",
|
||||
reqDTO.getDeviceId(), reqDTO.getIdentifier(), e);
|
||||
return success(IotDeviceServiceInvokeRespDTO.builder()
|
||||
.success(false)
|
||||
.code(500)
|
||||
.errorMsg(e.getMessage())
|
||||
.responseTime(LocalDateTime.now())
|
||||
.build());
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
@PostMapping(PREFIX + "/invoke-service-batch")
|
||||
@Operation(summary = "批量调用设备服务")
|
||||
public CommonResult<List<IotDeviceServiceInvokeRespDTO>> invokeServiceBatch(
|
||||
@RequestBody List<IotDeviceServiceInvokeReqDTO> reqDTOList) {
|
||||
|
||||
List<IotDeviceServiceInvokeRespDTO> results = new java.util.ArrayList<>();
|
||||
for (IotDeviceServiceInvokeReqDTO reqDTO : reqDTOList) {
|
||||
CommonResult<IotDeviceServiceInvokeRespDTO> result = invokeService(reqDTO);
|
||||
results.add(result.getData());
|
||||
}
|
||||
|
||||
return success(results);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -0,0 +1,175 @@
|
||||
package com.viewsh.module.iot.api.device;
|
||||
|
||||
import com.viewsh.framework.common.pojo.CommonResult;
|
||||
import com.viewsh.module.iot.api.device.dto.property.DevicePropertyBatchQueryReqDTO;
|
||||
import com.viewsh.module.iot.api.device.dto.property.DevicePropertyHistoryQueryReqDTO;
|
||||
import com.viewsh.module.iot.api.device.dto.property.DevicePropertyHistoryRespDTO;
|
||||
import com.viewsh.module.iot.api.device.dto.property.DevicePropertyRespDTO;
|
||||
import com.viewsh.module.iot.controller.admin.device.vo.property.IotDevicePropertyHistoryListReqVO;
|
||||
import com.viewsh.module.iot.controller.admin.device.vo.property.IotDevicePropertyRespVO;
|
||||
import com.viewsh.module.iot.dal.dataobject.device.IotDeviceDO;
|
||||
import com.viewsh.module.iot.dal.dataobject.device.IotDevicePropertyDO;
|
||||
import com.viewsh.module.iot.service.device.IotDeviceService;
|
||||
import com.viewsh.module.iot.service.device.property.IotDevicePropertyService;
|
||||
import io.swagger.v3.oas.annotations.Operation;
|
||||
import jakarta.annotation.Resource;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.context.annotation.Primary;
|
||||
import org.springframework.validation.annotation.Validated;
|
||||
import org.springframework.web.bind.annotation.GetMapping;
|
||||
import org.springframework.web.bind.annotation.PostMapping;
|
||||
import org.springframework.web.bind.annotation.RequestBody;
|
||||
import org.springframework.web.bind.annotation.RequestParam;
|
||||
import org.springframework.web.bind.annotation.RestController;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
import static com.viewsh.framework.common.pojo.CommonResult.success;
|
||||
|
||||
/**
|
||||
* IoT 设备属性查询 API 实现类
|
||||
* <p>
|
||||
* 提供 RPC 接口供其他模块（如 Ops 模块）查询设备属性
|
||||
*
|
||||
* @author lzh
|
||||
*/
|
||||
@RestController
|
||||
@Validated
|
||||
@Primary
|
||||
@Slf4j
|
||||
public class IotDevicePropertyQueryApiImpl implements IotDevicePropertyQueryApi {
|
||||
|
||||
@Resource
|
||||
private IotDeviceService deviceService;
|
||||
|
||||
@Resource
|
||||
private IotDevicePropertyService devicePropertyService;
|
||||
|
||||
@Override
|
||||
@GetMapping(PREFIX + "/get")
|
||||
@Operation(summary = "获取设备单个属性")
|
||||
public CommonResult<DevicePropertyRespDTO> getProperty(@RequestParam("deviceId") Long deviceId,
|
||||
@RequestParam("identifier") String identifier) {
|
||||
try {
|
||||
Map<String, IotDevicePropertyDO> properties = devicePropertyService.getLatestDeviceProperties(deviceId);
|
||||
IotDevicePropertyDO property = properties.get(identifier);
|
||||
|
||||
if (property == null) {
|
||||
return success(DevicePropertyRespDTO.builder()
|
||||
.identifier(identifier)
|
||||
.value(null)
|
||||
.updateTime(null)
|
||||
.build());
|
||||
}
|
||||
|
||||
return success(DevicePropertyRespDTO.builder()
|
||||
.identifier(identifier)
|
||||
.value(property.getValue())
|
||||
.updateTime(property.getUpdateTime() != null ?
|
||||
property.getUpdateTime().atZone(java.time.ZoneId.systemDefault()).toInstant().toEpochMilli() : null)
|
||||
.build());
|
||||
} catch (Exception e) {
|
||||
log.error("[getProperty] 获取设备属性失败: deviceId={}, identifier={}", deviceId, identifier, e);
|
||||
return success(DevicePropertyRespDTO.builder()
|
||||
.identifier(identifier)
|
||||
.value(null)
|
||||
.updateTime(null)
|
||||
.build());
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
@GetMapping(PREFIX + "/get-latest")
|
||||
@Operation(summary = "获取设备最新属性")
|
||||
public CommonResult<Map<String, Object>> getLatestProperties(@RequestParam("deviceId") Long deviceId) {
|
||||
try {
|
||||
// 验证设备是否存在
|
||||
IotDeviceDO device = deviceService.getDeviceFromCache(deviceId);
|
||||
if (device == null) {
|
||||
log.warn("[getLatestProperties] 设备不存在: deviceId={}", deviceId);
|
||||
return success(Map.of());
|
||||
}
|
||||
|
||||
Map<String, IotDevicePropertyDO> properties = devicePropertyService.getLatestDeviceProperties(deviceId);
|
||||
Map<String, Object> result = properties.entrySet().stream()
|
||||
.collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().getValue()));
|
||||
|
||||
return success(result);
|
||||
} catch (Exception e) {
|
||||
log.error("[getLatestProperties] 获取设备最新属性失败: deviceId={}", deviceId, e);
|
||||
return success(Map.of());
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
@PostMapping(PREFIX + "/batch-get")
|
||||
@Operation(summary = "批量获取多个设备的指定属性")
|
||||
public CommonResult<Map<Long, Map<String, Object>>> batchGetProperties(
|
||||
@RequestBody DevicePropertyBatchQueryReqDTO reqDTO) {
|
||||
try {
|
||||
Map<Long, Map<String, Object>> result = reqDTO.getDeviceIds().stream()
|
||||
.collect(Collectors.toMap(
|
||||
deviceId -> deviceId,
|
||||
deviceId -> {
|
||||
Map<String, IotDevicePropertyDO> properties = devicePropertyService.getLatestDeviceProperties(deviceId);
|
||||
|
||||
if (reqDTO.getIdentifiers() == null || reqDTO.getIdentifiers().isEmpty()) {
|
||||
// 返回所有属性
|
||||
return properties.entrySet().stream()
|
||||
.collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().getValue()));
|
||||
} else {
|
||||
// 只返回指定的属性
|
||||
return reqDTO.getIdentifiers().stream()
|
||||
.filter(properties::containsKey)
|
||||
.collect(Collectors.toMap(
|
||||
identifier -> identifier,
|
||||
identifier -> properties.get(identifier).getValue()));
|
||||
}
|
||||
}
|
||||
));
|
||||
|
||||
return success(result);
|
||||
} catch (Exception e) {
|
||||
log.error("[batchGetProperties] 批量获取设备属性失败: deviceIds={}", reqDTO.getDeviceIds(), e);
|
||||
return success(Map.of());
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
@PostMapping(PREFIX + "/history")
|
||||
@Operation(summary = "查询设备属性历史数据")
|
||||
public CommonResult<List<DevicePropertyHistoryRespDTO>> getPropertyHistory(
|
||||
@RequestBody DevicePropertyHistoryQueryReqDTO reqDTO) {
|
||||
try {
|
||||
// 构建查询请求
|
||||
IotDevicePropertyHistoryListReqVO listReqVO = new IotDevicePropertyHistoryListReqVO();
|
||||
listReqVO.setDeviceId(reqDTO.getDeviceId());
|
||||
listReqVO.setIdentifier(reqDTO.getIdentifier());
|
||||
if (reqDTO.getStartTime() != null && reqDTO.getEndTime() != null) {
|
||||
listReqVO.setTimes(new LocalDateTime[]{reqDTO.getStartTime(), reqDTO.getEndTime()});
|
||||
}
|
||||
// listReqVO.setPageSize(reqDTO.getLimit()); // 暂不支持分页
|
||||
|
||||
List<IotDevicePropertyRespVO> historyList = devicePropertyService.getHistoryDevicePropertyList(listReqVO);
|
||||
|
||||
List<DevicePropertyHistoryRespDTO> result = historyList.stream()
|
||||
.map(vo -> DevicePropertyHistoryRespDTO.builder()
|
||||
.identifier(vo.getIdentifier())
|
||||
.value(vo.getValue())
|
||||
.timestamp(vo.getUpdateTime())
|
||||
.build())
|
||||
.collect(Collectors.toList());
|
||||
|
||||
return success(result);
|
||||
} catch (Exception e) {
|
||||
log.error("[getPropertyHistory] 查询设备属性历史失败: deviceId={}, identifier={}",
|
||||
reqDTO.getDeviceId(), reqDTO.getIdentifier(), e);
|
||||
return success(List.of());
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,66 @@
|
||||
package com.viewsh.module.iot.api.device;
|
||||
|
||||
import com.viewsh.framework.common.pojo.CommonResult;
|
||||
import com.viewsh.module.iot.api.device.dto.status.DeviceStatusRespDTO;
|
||||
import com.viewsh.module.iot.core.enums.IotDeviceStateEnum;
|
||||
import com.viewsh.module.iot.dal.dataobject.device.IotDeviceDO;
|
||||
import com.viewsh.module.iot.service.device.IotDeviceService;
|
||||
import io.swagger.v3.oas.annotations.Operation;
|
||||
import jakarta.annotation.Resource;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.context.annotation.Primary;
|
||||
import org.springframework.validation.annotation.Validated;
|
||||
import org.springframework.web.bind.annotation.GetMapping;
|
||||
import org.springframework.web.bind.annotation.RequestParam;
|
||||
import org.springframework.web.bind.annotation.RestController;
|
||||
|
||||
import static com.viewsh.framework.common.pojo.CommonResult.success;
|
||||
|
||||
/**
|
||||
* IoT 设备状态查询 API 实现类
|
||||
* <p>
|
||||
* 提供 RPC 接口供其他模块查询设备状态
|
||||
*
|
||||
* @author lzh
|
||||
*/
|
||||
@RestController
|
||||
@Validated
|
||||
@Primary
|
||||
@Slf4j
|
||||
public class IotDeviceStatusQueryApiImpl implements IotDeviceStatusQueryApi {
|
||||
|
||||
@Resource
|
||||
private IotDeviceService deviceService;
|
||||
|
||||
@Override
|
||||
@GetMapping(PREFIX + "/is-online")
|
||||
@Operation(summary = "检查设备是否在线")
|
||||
public CommonResult<Boolean> isDeviceOnline(@RequestParam("deviceId") Long deviceId) {
|
||||
IotDeviceDO device = deviceService.getDeviceFromCache(deviceId);
|
||||
if (device == null) {
|
||||
return success(false);
|
||||
}
|
||||
return success(IotDeviceStateEnum.ONLINE.getState().equals(device.getState()));
|
||||
}
|
||||
|
||||
@Override
|
||||
@GetMapping(PREFIX + "/get-status")
|
||||
@Operation(summary = "获取设备状态")
|
||||
public CommonResult<DeviceStatusRespDTO> getDeviceStatus(@RequestParam("deviceId") Long deviceId) {
|
||||
IotDeviceDO device = deviceService.getDeviceFromCache(deviceId);
|
||||
if (device == null) {
|
||||
return success(DeviceStatusRespDTO.builder()
|
||||
.deviceId(deviceId)
|
||||
.status(IotDeviceStateEnum.OFFLINE.getState())
|
||||
.build());
|
||||
}
|
||||
return success(DeviceStatusRespDTO.builder()
|
||||
.deviceId(device.getId())
|
||||
.deviceCode(device.getSerialNumber())
|
||||
.status(device.getState())
|
||||
.statusChangeTime(IotDeviceStateEnum.ONLINE.getState().equals(device.getState()) ?
|
||||
device.getOnlineTime() : device.getOfflineTime())
|
||||
.build());
|
||||
}
|
||||
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1,209 +1,274 @@
|
||||
package com.viewsh.module.iot.service.device.property;
|
||||
|
||||
import cn.hutool.core.collection.CollUtil;
|
||||
import cn.hutool.core.date.LocalDateTimeUtil;
|
||||
import cn.hutool.core.map.MapUtil;
|
||||
import cn.hutool.core.util.StrUtil;
|
||||
import com.viewsh.framework.common.util.json.JsonUtils;
|
||||
import com.viewsh.framework.common.util.object.ObjectUtils;
|
||||
import com.viewsh.module.iot.controller.admin.device.vo.property.IotDevicePropertyHistoryListReqVO;
|
||||
import com.viewsh.module.iot.controller.admin.device.vo.property.IotDevicePropertyRespVO;
|
||||
import com.viewsh.module.iot.core.mq.message.IotDeviceMessage;
|
||||
import com.viewsh.module.iot.dal.dataobject.device.IotDeviceDO;
|
||||
import com.viewsh.module.iot.dal.dataobject.device.IotDevicePropertyDO;
|
||||
import com.viewsh.module.iot.dal.dataobject.product.IotProductDO;
|
||||
import com.viewsh.module.iot.dal.dataobject.thingmodel.IotThingModelDO;
|
||||
import com.viewsh.module.iot.dal.dataobject.thingmodel.model.dataType.ThingModelDateOrTextDataSpecs;
|
||||
import com.viewsh.module.iot.dal.redis.device.DevicePropertyRedisDAO;
|
||||
import com.viewsh.module.iot.dal.redis.device.DeviceReportTimeRedisDAO;
|
||||
import com.viewsh.module.iot.dal.redis.device.DeviceServerIdRedisDAO;
|
||||
import com.viewsh.module.iot.dal.tdengine.IotDevicePropertyMapper;
|
||||
import com.viewsh.module.iot.enums.thingmodel.IotDataSpecsDataTypeEnum;
|
||||
import com.viewsh.module.iot.enums.thingmodel.IotThingModelTypeEnum;
|
||||
import com.viewsh.module.iot.framework.tdengine.core.TDengineTableField;
|
||||
import com.viewsh.module.iot.service.product.IotProductService;
|
||||
import com.viewsh.module.iot.service.thingmodel.IotThingModelService;
|
||||
import jakarta.annotation.Resource;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.context.annotation.Lazy;
|
||||
import org.springframework.scheduling.annotation.Async;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
import java.util.*;
|
||||
|
||||
import static com.viewsh.framework.common.util.collection.CollectionUtils.*;
|
||||
|
||||
/**
|
||||
* IoT 设备【属性】数据 Service 实现类
|
||||
*
|
||||
* @author 芋道源码
|
||||
*/
|
||||
@Service
|
||||
@Slf4j
|
||||
public class IotDevicePropertyServiceImpl implements IotDevicePropertyService {
|
||||
|
||||
/**
|
||||
* 物模型的数据类型，与 TDengine 数据类型的映射关系
|
||||
*
|
||||
* @see <a href="https://docs.taosdata.com/reference/taos-sql/data-type/">TDEngine 数据类型</a>
|
||||
*/
|
||||
private static final Map<String, String> TYPE_MAPPING = MapUtil.<String, String>builder()
|
||||
.put(IotDataSpecsDataTypeEnum.INT.getDataType(), TDengineTableField.TYPE_INT)
|
||||
.put(IotDataSpecsDataTypeEnum.FLOAT.getDataType(), TDengineTableField.TYPE_FLOAT)
|
||||
.put(IotDataSpecsDataTypeEnum.DOUBLE.getDataType(), TDengineTableField.TYPE_DOUBLE)
|
||||
.put(IotDataSpecsDataTypeEnum.ENUM.getDataType(), TDengineTableField.TYPE_TINYINT)
|
||||
.put(IotDataSpecsDataTypeEnum.BOOL.getDataType(), TDengineTableField.TYPE_TINYINT)
|
||||
.put(IotDataSpecsDataTypeEnum.TEXT.getDataType(), TDengineTableField.TYPE_VARCHAR)
|
||||
.put(IotDataSpecsDataTypeEnum.DATE.getDataType(), TDengineTableField.TYPE_TIMESTAMP)
|
||||
.put(IotDataSpecsDataTypeEnum.STRUCT.getDataType(), TDengineTableField.TYPE_VARCHAR)
|
||||
.put(IotDataSpecsDataTypeEnum.ARRAY.getDataType(), TDengineTableField.TYPE_VARCHAR)
|
||||
.build();
|
||||
|
||||
@Resource
|
||||
private IotThingModelService thingModelService;
|
||||
@Resource
|
||||
@Lazy // 延迟加载，解决循环依赖
|
||||
private IotProductService productService;
|
||||
|
||||
@Resource
|
||||
private DevicePropertyRedisDAO deviceDataRedisDAO;
|
||||
@Resource
|
||||
private DeviceReportTimeRedisDAO deviceReportTimeRedisDAO;
|
||||
@Resource
|
||||
private DeviceServerIdRedisDAO deviceServerIdRedisDAO;
|
||||
|
||||
@Resource
|
||||
private IotDevicePropertyMapper devicePropertyMapper;
|
||||
|
||||
// ========== 设备属性相关操作 ==========
|
||||
|
||||
@Override
|
||||
public void defineDevicePropertyData(Long productId) {
|
||||
// 1.1 查询产品和物模型
|
||||
IotProductDO product = productService.validateProductExists(productId);
|
||||
List<IotThingModelDO> thingModels = filterList(thingModelService.getThingModelListByProductId(productId),
|
||||
thingModel -> IotThingModelTypeEnum.PROPERTY.getType().equals(thingModel.getType()));
|
||||
// 1.2 解析 DB 里的字段
|
||||
List<TDengineTableField> oldFields = new ArrayList<>();
|
||||
try {
|
||||
oldFields.addAll(devicePropertyMapper.getProductPropertySTableFieldList(product.getId()));
|
||||
} catch (Exception e) {
|
||||
if (!e.getMessage().contains("Table does not exist")) {
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
|
||||
// 2.1 情况一：如果是新增的时候，需要创建表
|
||||
List<TDengineTableField> newFields = buildTableFieldList(thingModels);
|
||||
if (CollUtil.isEmpty(oldFields)) {
|
||||
if (CollUtil.isEmpty(newFields)) {
|
||||
log.info("[defineDevicePropertyData][productId({}) 没有需要定义的属性]", productId);
|
||||
return;
|
||||
}
|
||||
devicePropertyMapper.createProductPropertySTable(product.getId(), newFields);
|
||||
return;
|
||||
}
|
||||
// 2.2 情况二：如果是修改的时候，需要更新表
|
||||
devicePropertyMapper.alterProductPropertySTable(product.getId(), oldFields, newFields);
|
||||
}
|
||||
|
||||
private List<TDengineTableField> buildTableFieldList(List<IotThingModelDO> thingModels) {
|
||||
return convertList(thingModels, thingModel -> {
|
||||
TDengineTableField field = new TDengineTableField(
|
||||
StrUtil.toUnderlineCase(thingModel.getIdentifier()), // TDengine 字段默认都是小写
|
||||
TYPE_MAPPING.get(thingModel.getProperty().getDataType()));
|
||||
String dataType = thingModel.getProperty().getDataType();
|
||||
if (Objects.equals(dataType, IotDataSpecsDataTypeEnum.TEXT.getDataType())) {
|
||||
field.setLength(((ThingModelDateOrTextDataSpecs) thingModel.getProperty().getDataSpecs()).getLength());
|
||||
} else if (ObjectUtils.equalsAny(dataType, IotDataSpecsDataTypeEnum.STRUCT.getDataType(),
|
||||
IotDataSpecsDataTypeEnum.ARRAY.getDataType())) {
|
||||
field.setLength(TDengineTableField.LENGTH_VARCHAR);
|
||||
}
|
||||
return field;
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public void saveDeviceProperty(IotDeviceDO device, IotDeviceMessage message) {
|
||||
if (!(message.getParams() instanceof Map)) {
|
||||
log.error("[saveDeviceProperty][消息内容({}) 的 data 类型不正确]", message);
|
||||
return;
|
||||
}
|
||||
|
||||
// 1. 根据物模型，拼接合法的属性
|
||||
// TODO @芋艿：【待定 004】赋能后，属性到底以 thingModel 为准（ik），还是 db 的表结构为准（tl）？
|
||||
List<IotThingModelDO> thingModels = thingModelService.getThingModelListByProductIdFromCache(device.getProductId());
|
||||
Map<String, Object> properties = new HashMap<>();
|
||||
((Map<?, ?>) message.getParams()).forEach((key, value) -> {
|
||||
IotThingModelDO thingModel = CollUtil.findOne(thingModels, o -> o.getIdentifier().equals(key));
|
||||
if (thingModel == null || thingModel.getProperty() == null) {
|
||||
log.error("[saveDeviceProperty][消息({}) 的属性({}) 不存在]", message, key);
|
||||
return;
|
||||
}
|
||||
if (ObjectUtils.equalsAny(thingModel.getProperty().getDataType(),
|
||||
IotDataSpecsDataTypeEnum.STRUCT.getDataType(), IotDataSpecsDataTypeEnum.ARRAY.getDataType())) {
|
||||
// 特殊：STRUCT 和 ARRAY 类型，在 TDengine 里，有没对应数据类型，只能通过 JSON 来存储
|
||||
properties.put((String) key, JsonUtils.toJsonString(value));
|
||||
} else {
|
||||
properties.put((String) key, value);
|
||||
}
|
||||
});
|
||||
if (CollUtil.isEmpty(properties)) {
|
||||
log.error("[saveDeviceProperty][消息({}) 没有合法的属性]", message);
|
||||
return;
|
||||
}
|
||||
|
||||
// 2.1 保存设备属性【数据】
|
||||
devicePropertyMapper.insert(device, properties, LocalDateTimeUtil.toEpochMilli(message.getReportTime()));
|
||||
|
||||
// 2.2 保存设备属性【日志】
|
||||
Map<String, IotDevicePropertyDO> properties2 = convertMap(properties.entrySet(), Map.Entry::getKey, entry ->
|
||||
IotDevicePropertyDO.builder().value(entry.getValue()).updateTime(message.getReportTime()).build());
|
||||
deviceDataRedisDAO.putAll(device.getId(), properties2);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Map<String, IotDevicePropertyDO> getLatestDeviceProperties(Long deviceId) {
|
||||
return deviceDataRedisDAO.get(deviceId);
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<IotDevicePropertyRespVO> getHistoryDevicePropertyList(IotDevicePropertyHistoryListReqVO listReqVO) {
|
||||
try {
|
||||
return devicePropertyMapper.selectListByHistory(listReqVO);
|
||||
} catch (Exception exception) {
|
||||
if (exception.getMessage().contains("Table does not exist")) {
|
||||
return Collections.emptyList();
|
||||
}
|
||||
throw exception;
|
||||
}
|
||||
}
|
||||
|
||||
// ========== 设备时间相关操作 ==========
|
||||
|
||||
@Override
|
||||
public Set<Long> getDeviceIdListByReportTime(LocalDateTime maxReportTime) {
|
||||
return deviceReportTimeRedisDAO.range(maxReportTime);
|
||||
}
|
||||
|
||||
@Override
|
||||
@Async
|
||||
public void updateDeviceReportTimeAsync(Long id, LocalDateTime reportTime) {
|
||||
deviceReportTimeRedisDAO.update(id, reportTime);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void updateDeviceServerIdAsync(Long id, String serverId) {
|
||||
if (StrUtil.isEmpty(serverId)) {
|
||||
return;
|
||||
}
|
||||
deviceServerIdRedisDAO.update(id, serverId);
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getDeviceServerId(Long id) {
|
||||
return deviceServerIdRedisDAO.get(id);
|
||||
}
|
||||
|
||||
package com.viewsh.module.iot.service.device.property;
|
||||
|
||||
import cn.hutool.core.collection.CollUtil;
|
||||
import cn.hutool.core.date.LocalDateTimeUtil;
|
||||
import cn.hutool.core.map.MapUtil;
|
||||
import cn.hutool.core.util.StrUtil;
|
||||
import com.viewsh.framework.common.util.json.JsonUtils;
|
||||
import com.viewsh.framework.common.util.object.ObjectUtils;
|
||||
import com.viewsh.framework.mq.redis.core.RedisMQTemplate;
|
||||
import com.viewsh.module.iot.core.integration.event.DevicePropertyChangedEvent;
|
||||
import com.viewsh.module.iot.controller.admin.device.vo.property.IotDevicePropertyHistoryListReqVO;
|
||||
import com.viewsh.module.iot.controller.admin.device.vo.property.IotDevicePropertyRespVO;
|
||||
import com.viewsh.module.iot.core.integration.publisher.IntegrationEventPublisher;
|
||||
import com.viewsh.module.iot.core.mq.message.IotDeviceMessage;
|
||||
import com.viewsh.module.iot.dal.dataobject.device.IotDeviceDO;
|
||||
import com.viewsh.module.iot.dal.dataobject.device.IotDevicePropertyDO;
|
||||
import com.viewsh.module.iot.dal.dataobject.product.IotProductDO;
|
||||
import com.viewsh.module.iot.dal.dataobject.thingmodel.IotThingModelDO;
|
||||
import com.viewsh.module.iot.dal.dataobject.thingmodel.model.dataType.ThingModelDateOrTextDataSpecs;
|
||||
import com.viewsh.module.iot.dal.redis.device.DevicePropertyRedisDAO;
|
||||
import com.viewsh.module.iot.dal.redis.device.DeviceReportTimeRedisDAO;
|
||||
import com.viewsh.module.iot.dal.redis.device.DeviceServerIdRedisDAO;
|
||||
import com.viewsh.module.iot.dal.tdengine.IotDevicePropertyMapper;
|
||||
import com.viewsh.module.iot.enums.thingmodel.IotDataSpecsDataTypeEnum;
|
||||
import com.viewsh.module.iot.enums.thingmodel.IotThingModelTypeEnum;
|
||||
import com.viewsh.module.iot.framework.tdengine.core.TDengineTableField;
|
||||
import com.viewsh.module.iot.service.product.IotProductService;
|
||||
import com.viewsh.module.iot.service.thingmodel.IotThingModelService;
|
||||
import jakarta.annotation.Resource;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.context.annotation.Lazy;
|
||||
import org.springframework.scheduling.annotation.Async;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
import java.util.*;
|
||||
|
||||
import static com.viewsh.framework.common.util.collection.CollectionUtils.*;
|
||||
|
||||
/**
|
||||
* IoT 设备【属性】数据 Service 实现类
|
||||
*
|
||||
* @author 芋道源码
|
||||
*/
|
||||
@Service
|
||||
@Slf4j
|
||||
public class IotDevicePropertyServiceImpl implements IotDevicePropertyService {
|
||||
|
||||
/**
|
||||
* 物模型的数据类型，与 TDengine 数据类型的映射关系
|
||||
*
|
||||
* @see <a href="https://docs.taosdata.com/reference/taos-sql/data-type/">TDEngine 数据类型</a>
|
||||
*/
|
||||
private static final Map<String, String> TYPE_MAPPING = MapUtil.<String, String>builder()
|
||||
.put(IotDataSpecsDataTypeEnum.INT.getDataType(), TDengineTableField.TYPE_INT)
|
||||
.put(IotDataSpecsDataTypeEnum.FLOAT.getDataType(), TDengineTableField.TYPE_FLOAT)
|
||||
.put(IotDataSpecsDataTypeEnum.DOUBLE.getDataType(), TDengineTableField.TYPE_DOUBLE)
|
||||
.put(IotDataSpecsDataTypeEnum.ENUM.getDataType(), TDengineTableField.TYPE_TINYINT)
|
||||
.put(IotDataSpecsDataTypeEnum.BOOL.getDataType(), TDengineTableField.TYPE_TINYINT)
|
||||
.put(IotDataSpecsDataTypeEnum.TEXT.getDataType(), TDengineTableField.TYPE_VARCHAR)
|
||||
.put(IotDataSpecsDataTypeEnum.DATE.getDataType(), TDengineTableField.TYPE_TIMESTAMP)
|
||||
.put(IotDataSpecsDataTypeEnum.STRUCT.getDataType(), TDengineTableField.TYPE_VARCHAR)
|
||||
.put(IotDataSpecsDataTypeEnum.ARRAY.getDataType(), TDengineTableField.TYPE_VARCHAR)
|
||||
.build();
|
||||
|
||||
@Resource
|
||||
private IotThingModelService thingModelService;
|
||||
@Resource
|
||||
@Lazy // 延迟加载，解决循环依赖
|
||||
private IotProductService productService;
|
||||
|
||||
@Resource
|
||||
private DevicePropertyRedisDAO deviceDataRedisDAO;
|
||||
@Resource
|
||||
private DeviceReportTimeRedisDAO deviceReportTimeRedisDAO;
|
||||
@Resource
|
||||
private DeviceServerIdRedisDAO deviceServerIdRedisDAO;
|
||||
|
||||
@Resource
|
||||
private IotDevicePropertyMapper devicePropertyMapper;
|
||||
|
||||
@Resource
|
||||
private RedisMQTemplate redisMQTemplate;
|
||||
|
||||
@Resource
|
||||
private IntegrationEventPublisher integrationEventPublisher;
|
||||
|
||||
// ========== 设备属性相关操作 ==========
|
||||
|
||||
@Override
|
||||
public void defineDevicePropertyData(Long productId) {
|
||||
// 1.1 查询产品和物模型
|
||||
IotProductDO product = productService.validateProductExists(productId);
|
||||
List<IotThingModelDO> thingModels = filterList(thingModelService.getThingModelListByProductId(productId),
|
||||
thingModel -> IotThingModelTypeEnum.PROPERTY.getType().equals(thingModel.getType()));
|
||||
// 1.2 解析 DB 里的字段
|
||||
List<TDengineTableField> oldFields = new ArrayList<>();
|
||||
try {
|
||||
oldFields.addAll(devicePropertyMapper.getProductPropertySTableFieldList(product.getId()));
|
||||
} catch (Exception e) {
|
||||
if (!e.getMessage().contains("Table does not exist")) {
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
|
||||
// 2.1 情况一：如果是新增的时候，需要创建表
|
||||
List<TDengineTableField> newFields = buildTableFieldList(thingModels);
|
||||
if (CollUtil.isEmpty(oldFields)) {
|
||||
if (CollUtil.isEmpty(newFields)) {
|
||||
log.info("[defineDevicePropertyData][productId({}) 没有需要定义的属性]", productId);
|
||||
return;
|
||||
}
|
||||
devicePropertyMapper.createProductPropertySTable(product.getId(), newFields);
|
||||
return;
|
||||
}
|
||||
// 2.2 情况二：如果是修改的时候，需要更新表
|
||||
devicePropertyMapper.alterProductPropertySTable(product.getId(), oldFields, newFields);
|
||||
}
|
||||
|
||||
private List<TDengineTableField> buildTableFieldList(List<IotThingModelDO> thingModels) {
|
||||
return convertList(thingModels, thingModel -> {
|
||||
TDengineTableField field = new TDengineTableField(
|
||||
StrUtil.toUnderlineCase(thingModel.getIdentifier()), // TDengine 字段默认都是小写
|
||||
TYPE_MAPPING.get(thingModel.getProperty().getDataType()));
|
||||
String dataType = thingModel.getProperty().getDataType();
|
||||
if (Objects.equals(dataType, IotDataSpecsDataTypeEnum.TEXT.getDataType())) {
|
||||
field.setLength(((ThingModelDateOrTextDataSpecs) thingModel.getProperty().getDataSpecs()).getLength());
|
||||
} else if (ObjectUtils.equalsAny(dataType, IotDataSpecsDataTypeEnum.STRUCT.getDataType(),
|
||||
IotDataSpecsDataTypeEnum.ARRAY.getDataType())) {
|
||||
field.setLength(TDengineTableField.LENGTH_VARCHAR);
|
||||
}
|
||||
return field;
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public void saveDeviceProperty(IotDeviceDO device, IotDeviceMessage message) {
|
||||
if (!(message.getParams() instanceof Map)) {
|
||||
log.error("[saveDeviceProperty][消息内容({}) 的 data 类型不正确]", message);
|
||||
return;
|
||||
}
|
||||
|
||||
// 1. 根据物模型，拼接合法的属性
|
||||
// TODO @芋艿：【待定 004】赋能后，属性到底以 thingModel 为准（ik），还是 db 的表结构为准（tl）？
|
||||
List<IotThingModelDO> thingModels = thingModelService.getThingModelListByProductIdFromCache(device.getProductId());
|
||||
Map<String, Object> properties = new HashMap<>();
|
||||
((Map<?, ?>) message.getParams()).forEach((key, value) -> {
|
||||
IotThingModelDO thingModel = CollUtil.findOne(thingModels, o -> o.getIdentifier().equals(key));
|
||||
if (thingModel == null || thingModel.getProperty() == null) {
|
||||
log.error("[saveDeviceProperty][消息({}) 的属性({}) 不存在]", message, key);
|
||||
return;
|
||||
}
|
||||
if (ObjectUtils.equalsAny(thingModel.getProperty().getDataType(),
|
||||
IotDataSpecsDataTypeEnum.STRUCT.getDataType(), IotDataSpecsDataTypeEnum.ARRAY.getDataType())) {
|
||||
// 特殊：STRUCT 和 ARRAY 类型，在 TDengine 里，有没对应数据类型，只能通过 JSON 来存储
|
||||
properties.put((String) key, JsonUtils.toJsonString(value));
|
||||
} else {
|
||||
properties.put((String) key, value);
|
||||
}
|
||||
});
|
||||
if (CollUtil.isEmpty(properties)) {
|
||||
log.error("[saveDeviceProperty][消息({}) 没有合法的属性]", message);
|
||||
return;
|
||||
}
|
||||
|
||||
// 2.1 保存设备属性【数据】
|
||||
devicePropertyMapper.insert(device, properties, LocalDateTimeUtil.toEpochMilli(message.getReportTime()));
|
||||
|
||||
// 2.2 保存设备属性【日志】
|
||||
Map<String, IotDevicePropertyDO> properties2 = convertMap(properties.entrySet(), Map.Entry::getKey, entry ->
|
||||
IotDevicePropertyDO.builder().value(entry.getValue()).updateTime(message.getReportTime()).build());
|
||||
deviceDataRedisDAO.putAll(device.getId(), properties2);
|
||||
|
||||
// 2.3 发布属性消息到 Redis Stream（供其他模块如 Ops 订阅）
|
||||
publishPropertyMessage(device, properties, message.getReportTime());
|
||||
}
|
||||
|
||||
/**
|
||||
* 发布设备属性消息
|
||||
* <p>
|
||||
* 同时发布到 Redis Stream（旧方式，保持兼容）和 RocketMQ IntegrationEventBus（新方式）
|
||||
*
|
||||
* @param device 设备信息
|
||||
* @param properties 属性数据
|
||||
* @param reportTime 上报时间
|
||||
*/
|
||||
private void publishPropertyMessage(IotDeviceDO device, Map<String, Object> properties, LocalDateTime reportTime) {
|
||||
try {
|
||||
// 获取产品信息
|
||||
IotProductDO product = productService.getProductFromCache(device.getProductId());
|
||||
String productKey = product != null ? product.getProductKey() : "unknown";
|
||||
|
||||
// 发布到 IntegrationEventBus（RocketMQ）
|
||||
publishToIntegrationEventBus(device, productKey, properties, reportTime);
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("[publishPropertyMessage] 属性消息发布失败: deviceId={}", device.getId(), e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 发布跨模块属性变更事件到 RocketMQ
|
||||
*
|
||||
* @param device 设备信息
|
||||
* @param productKey 产品标识符（用作 RocketMQ Tag）
|
||||
* @param properties 属性数据
|
||||
* @param reportTime 上报时间
|
||||
*/
|
||||
private void publishToIntegrationEventBus(IotDeviceDO device, String productKey,
|
||||
Map<String, Object> properties, LocalDateTime reportTime) {
|
||||
try {
|
||||
DevicePropertyChangedEvent event = DevicePropertyChangedEvent.builder()
|
||||
.deviceId(device.getId())
|
||||
.deviceName(device.getDeviceName())
|
||||
.productId(device.getProductId())
|
||||
.productKey(productKey)
|
||||
.tenantId(device.getTenantId())
|
||||
.properties(properties)
|
||||
.changedIdentifiers(properties.keySet())
|
||||
.eventTime(reportTime)
|
||||
.build();
|
||||
|
||||
integrationEventPublisher.publishPropertyChanged(event);
|
||||
log.debug("[publishToIntegrationEventBus] 跨模块属性变更事件已发布: eventId={}, deviceId={}, productKey={}, properties={}",
|
||||
event.getEventId(), device.getId(), productKey, properties.keySet());
|
||||
} catch (Exception e) {
|
||||
log.error("[publishToIntegrationEventBus] 跨模块属性变更事件发布失败: deviceId={}", device.getId(), e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public Map<String, IotDevicePropertyDO> getLatestDeviceProperties(Long deviceId) {
|
||||
return deviceDataRedisDAO.get(deviceId);
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<IotDevicePropertyRespVO> getHistoryDevicePropertyList(IotDevicePropertyHistoryListReqVO listReqVO) {
|
||||
try {
|
||||
return devicePropertyMapper.selectListByHistory(listReqVO);
|
||||
} catch (Exception exception) {
|
||||
if (exception.getMessage().contains("Table does not exist")) {
|
||||
return Collections.emptyList();
|
||||
}
|
||||
throw exception;
|
||||
}
|
||||
}
|
||||
|
||||
// ========== 设备时间相关操作 ==========
|
||||
|
||||
@Override
|
||||
public Set<Long> getDeviceIdListByReportTime(LocalDateTime maxReportTime) {
|
||||
return deviceReportTimeRedisDAO.range(maxReportTime);
|
||||
}
|
||||
|
||||
@Override
|
||||
@Async
|
||||
public void updateDeviceReportTimeAsync(Long id, LocalDateTime reportTime) {
|
||||
deviceReportTimeRedisDAO.update(id, reportTime);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void updateDeviceServerIdAsync(Long id, String serverId) {
|
||||
if (StrUtil.isEmpty(serverId)) {
|
||||
return;
|
||||
}
|
||||
deviceServerIdRedisDAO.update(id, serverId);
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getDeviceServerId(Long id) {
|
||||
return deviceServerIdRedisDAO.get(id);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,157 +1,163 @@
|
||||
spring:
|
||||
application:
|
||||
name: iot-server
|
||||
|
||||
profiles:
|
||||
active: local
|
||||
|
||||
main:
|
||||
allow-circular-references: true # 允许循环依赖，因为项目是三层架构，无法避免这个情况。
|
||||
allow-bean-definition-overriding: true # 允许 Bean 覆盖，例如说 Feign 等会存在重复定义的服务
|
||||
|
||||
config:
|
||||
import:
|
||||
- optional:classpath:application-${spring.profiles.active}.yaml # 加载【本地】配置
|
||||
- optional:nacos:${spring.application.name}-${spring.profiles.active}.yaml # 加载【Nacos】的配置
|
||||
|
||||
# Servlet 配置
|
||||
servlet:
|
||||
# 文件上传相关配置项
|
||||
multipart:
|
||||
max-file-size: 16MB # 单个文件大小
|
||||
max-request-size: 32MB # 设置总上传的文件大小
|
||||
|
||||
# Jackson 配置项
|
||||
jackson:
|
||||
serialization:
|
||||
write-dates-as-timestamps: true # 设置 LocalDateTime 的格式，使用时间戳
|
||||
write-date-timestamps-as-nanoseconds: false # 设置不使用 nanoseconds 的格式。例如说 1611460870.401，而是直接 1611460870401
|
||||
write-durations-as-timestamps: true # 设置 Duration 的格式，使用时间戳
|
||||
fail-on-empty-beans: false # 允许序列化无属性的 Bean
|
||||
|
||||
# Cache 配置项
|
||||
cache:
|
||||
type: REDIS
|
||||
redis:
|
||||
time-to-live: 1h # 设置过期时间为 1 小时
|
||||
|
||||
server:
|
||||
port: 48091
|
||||
|
||||
logging:
|
||||
file:
|
||||
name: ${user.home}/logs/${spring.application.name}.log # 日志文件名，全路径
|
||||
|
||||
--- #################### 接口文档配置 ####################
|
||||
|
||||
springdoc:
|
||||
api-docs:
|
||||
enabled: true # 1. 是否开启 Swagger 接文档的元数据
|
||||
path: /v3/api-docs
|
||||
swagger-ui:
|
||||
enabled: true # 2.1 是否开启 Swagger 文档的官方 UI 界面
|
||||
path: /swagger-ui
|
||||
default-flat-param-object: true # 参见 https://doc.xiaominfo.com/docs/faq/v4/knife4j-parameterobject-flat-param 文档
|
||||
|
||||
knife4j:
|
||||
enable: true
|
||||
setting:
|
||||
language: zh_cn
|
||||
|
||||
# MyBatis Plus 的配置项
|
||||
mybatis-plus:
|
||||
configuration:
|
||||
map-underscore-to-camel-case: true # 虽然默认为 true ，但是还是显示去指定下。
|
||||
global-config:
|
||||
db-config:
|
||||
id-type: NONE # “智能”模式，基于 IdTypeEnvironmentPostProcessor + 数据源的类型，自动适配成 AUTO、INPUT 模式。
|
||||
# id-type: AUTO # 自增 ID，适合 MySQL 等直接自增的数据库
|
||||
# id-type: INPUT # 用户输入 ID，适合 Oracle、PostgreSQL、Kingbase、DB2、H2 数据库
|
||||
# id-type: ASSIGN_ID # 分配 ID，默认使用雪花算法。注意，Oracle、PostgreSQL、Kingbase、DB2、H2 数据库时，需要去除实体类上的 @KeySequence 注解
|
||||
logic-delete-value: 1 # 逻辑已删除值(默认为 1)
|
||||
logic-not-delete-value: 0 # 逻辑未删除值(默认为 0)
|
||||
banner: false # 关闭控制台的 Banner 打印
|
||||
type-aliases-package: ${viewsh.info.base-package}.dal.dataobject
|
||||
encryptor:
|
||||
password: XDV71a+xqStEA3WH # 加解密的秘钥，可使用 https://www.imaegoo.com/2020/aes-key-generator/ 网站生成
|
||||
|
||||
mybatis-plus-join:
|
||||
banner: false # 关闭控制台的 Banner 打印
|
||||
|
||||
# Spring Data Redis 配置
|
||||
spring:
|
||||
data:
|
||||
redis:
|
||||
repositories:
|
||||
enabled: false # 项目未使用到 Spring Data Redis 的 Repository，所以直接禁用，保证启动速度
|
||||
|
||||
# VO 转换（数据翻译）相关
|
||||
easy-trans:
|
||||
is-enable-global: false # 【默认禁用，对性能确认压力大】启用全局翻译（拦截所有 SpringMVC ResponseBody 进行自动翻译 )。如果对于性能要求很高可关闭此配置，或通过 @IgnoreTrans 忽略某个接口
|
||||
|
||||
--- #################### RPC 远程调用相关配置 ####################
|
||||
|
||||
--- #################### 消息队列相关 ####################
|
||||
|
||||
# rocketmq 配置项，对应 RocketMQProperties 配置类
|
||||
rocketmq:
|
||||
# Producer 配置项
|
||||
producer:
|
||||
group: ${spring.application.name}_PRODUCER # 生产者分组
|
||||
|
||||
spring:
|
||||
# Kafka 配置项，对应 KafkaProperties 配置类
|
||||
kafka:
|
||||
# Kafka Producer 配置项
|
||||
producer:
|
||||
acks: 1 # 0-不应答。1-leader 应答。all-所有 leader 和 follower 应答。
|
||||
retries: 3 # 发送失败时，重试发送的次数
|
||||
value-serializer: org.springframework.kafka.support.serializer.JsonSerializer # 消息的 value 的序列化
|
||||
# Kafka Consumer 配置项
|
||||
consumer:
|
||||
auto-offset-reset: earliest # 设置消费者分组最初的消费进度为 earliest 。可参考博客 https://blog.csdn.net/lishuangzhe7047/article/details/74530417 理解
|
||||
value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
|
||||
properties:
|
||||
spring.json.trusted.packages: '*'
|
||||
# Kafka Consumer Listener 监听器配置
|
||||
listener:
|
||||
missing-topics-fatal: false # 消费监听接口监听的主题不存在时，默认会报错。所以通过设置为 false ，解决报错
|
||||
|
||||
--- #################### 定时任务相关配置 ####################
|
||||
|
||||
xxl:
|
||||
job:
|
||||
executor:
|
||||
appname: ${spring.application.name} # 执行器 AppName
|
||||
logpath: ${user.home}/logs/xxl-job/${spring.application.name} # 执行器运行日志文件存储磁盘路径
|
||||
accessToken: default_token # 执行器通讯TOKEN
|
||||
|
||||
--- #################### 芋道相关配置 ####################
|
||||
|
||||
viewsh:
|
||||
info:
|
||||
version: 1.0.0
|
||||
base-package: com.viewsh.module.iot
|
||||
web:
|
||||
admin-ui:
|
||||
url: http://dashboard.viewsh.iocoder.cn # Admin 管理后台 UI 的地址
|
||||
xss:
|
||||
enable: false
|
||||
exclude-urls: # 如下 url，仅仅是为了演示，去掉配置也没关系
|
||||
- ${management.endpoints.web.base-path}/** # 不处理 Actuator 的请求
|
||||
swagger:
|
||||
title: 管理后台
|
||||
description: 提供管理员管理的所有功能
|
||||
version: ${viewsh.info.version}
|
||||
tenant: # 多租户相关配置项
|
||||
enable: true
|
||||
ignore-urls:
|
||||
ignore-tables:
|
||||
ignore-caches:
|
||||
- iot:device
|
||||
- iot:thing_model_list
|
||||
iot:
|
||||
message-bus:
|
||||
type: redis # 消息总线的类型
|
||||
|
||||
spring:
|
||||
application:
|
||||
name: iot-server
|
||||
|
||||
profiles:
|
||||
active: local
|
||||
|
||||
main:
|
||||
allow-circular-references: true # 允许循环依赖，因为项目是三层架构，无法避免这个情况。
|
||||
allow-bean-definition-overriding: true # 允许 Bean 覆盖，例如说 Feign 等会存在重复定义的服务
|
||||
|
||||
config:
|
||||
import:
|
||||
- optional:classpath:application-${spring.profiles.active}.yaml # 加载【本地】配置
|
||||
- optional:nacos:${spring.application.name}-${spring.profiles.active}.yaml # 加载【Nacos】的配置
|
||||
|
||||
# Servlet 配置
|
||||
servlet:
|
||||
# 文件上传相关配置项
|
||||
multipart:
|
||||
max-file-size: 16MB # 单个文件大小
|
||||
max-request-size: 32MB # 设置总上传的文件大小
|
||||
|
||||
# Jackson 配置项
|
||||
jackson:
|
||||
serialization:
|
||||
write-dates-as-timestamps: true # 设置 LocalDateTime 的格式，使用时间戳
|
||||
write-date-timestamps-as-nanoseconds: false # 设置不使用 nanoseconds 的格式。例如说 1611460870.401，而是直接 1611460870401
|
||||
write-durations-as-timestamps: true # 设置 Duration 的格式，使用时间戳
|
||||
fail-on-empty-beans: false # 允许序列化无属性的 Bean
|
||||
|
||||
# Cache 配置项
|
||||
cache:
|
||||
type: REDIS
|
||||
redis:
|
||||
time-to-live: 1h # 设置过期时间为 1 小时
|
||||
|
||||
server:
|
||||
port: 48091
|
||||
|
||||
logging:
|
||||
file:
|
||||
name: ${user.home}/logs/${spring.application.name}.log # 日志文件名，全路径
|
||||
|
||||
--- #################### 接口文档配置 ####################
|
||||
|
||||
springdoc:
|
||||
api-docs:
|
||||
enabled: true # 1. 是否开启 Swagger 接文档的元数据
|
||||
path: /v3/api-docs
|
||||
swagger-ui:
|
||||
enabled: true # 2.1 是否开启 Swagger 文档的官方 UI 界面
|
||||
path: /swagger-ui
|
||||
default-flat-param-object: true # 参见 https://doc.xiaominfo.com/docs/faq/v4/knife4j-parameterobject-flat-param 文档
|
||||
|
||||
knife4j:
|
||||
enable: true
|
||||
setting:
|
||||
language: zh_cn
|
||||
|
||||
# MyBatis Plus 的配置项
|
||||
mybatis-plus:
|
||||
configuration:
|
||||
map-underscore-to-camel-case: true # 虽然默认为 true ，但是还是显示去指定下。
|
||||
global-config:
|
||||
db-config:
|
||||
id-type: NONE # “智能”模式，基于 IdTypeEnvironmentPostProcessor + 数据源的类型，自动适配成 AUTO、INPUT 模式。
|
||||
# id-type: AUTO # 自增 ID，适合 MySQL 等直接自增的数据库
|
||||
# id-type: INPUT # 用户输入 ID，适合 Oracle、PostgreSQL、Kingbase、DB2、H2 数据库
|
||||
# id-type: ASSIGN_ID # 分配 ID，默认使用雪花算法。注意，Oracle、PostgreSQL、Kingbase、DB2、H2 数据库时，需要去除实体类上的 @KeySequence 注解
|
||||
logic-delete-value: 1 # 逻辑已删除值(默认为 1)
|
||||
logic-not-delete-value: 0 # 逻辑未删除值(默认为 0)
|
||||
banner: false # 关闭控制台的 Banner 打印
|
||||
type-aliases-package: ${viewsh.info.base-package}.dal.dataobject
|
||||
encryptor:
|
||||
password: XDV71a+xqStEA3WH # 加解密的秘钥，可使用 https://www.imaegoo.com/2020/aes-key-generator/ 网站生成
|
||||
|
||||
mybatis-plus-join:
|
||||
banner: false # 关闭控制台的 Banner 打印
|
||||
|
||||
# Spring Data Redis 配置
|
||||
spring:
|
||||
data:
|
||||
redis:
|
||||
repositories:
|
||||
enabled: false # 项目未使用到 Spring Data Redis 的 Repository，所以直接禁用，保证启动速度
|
||||
|
||||
# VO 转换（数据翻译）相关
|
||||
easy-trans:
|
||||
is-enable-global: false # 【默认禁用，对性能确认压力大】启用全局翻译（拦截所有 SpringMVC ResponseBody 进行自动翻译 )。如果对于性能要求很高可关闭此配置，或通过 @IgnoreTrans 忽略某个接口
|
||||
|
||||
--- #################### RPC 远程调用相关配置 ####################
|
||||
|
||||
--- #################### 消息队列相关 ####################
|
||||
|
||||
# rocketmq 配置项，对应 RocketMQProperties 配置类
|
||||
rocketmq:
|
||||
# Producer 配置项
|
||||
producer:
|
||||
group: ${spring.application.name}_PRODUCER # 生产者分组
|
||||
|
||||
spring:
|
||||
# Kafka 配置项，对应 KafkaProperties 配置类
|
||||
kafka:
|
||||
# Kafka Producer 配置项
|
||||
producer:
|
||||
acks: 1 # 0-不应答。1-leader 应答。all-所有 leader 和 follower 应答。
|
||||
retries: 3 # 发送失败时，重试发送的次数
|
||||
value-serializer: org.springframework.kafka.support.serializer.JsonSerializer # 消息的 value 的序列化
|
||||
# Kafka Consumer 配置项
|
||||
consumer:
|
||||
auto-offset-reset: earliest # 设置消费者分组最初的消费进度为 earliest 。可参考博客 https://blog.csdn.net/lishuangzhe7047/article/details/74530417 理解
|
||||
value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
|
||||
properties:
|
||||
spring.json.trusted.packages: '*'
|
||||
# Kafka Consumer Listener 监听器配置
|
||||
listener:
|
||||
missing-topics-fatal: false # 消费监听接口监听的主题不存在时，默认会报错。所以通过设置为 false ，解决报错
|
||||
|
||||
--- #################### 定时任务相关配置 ####################
|
||||
|
||||
xxl:
|
||||
job:
|
||||
executor:
|
||||
appname: ${spring.application.name} # 执行器 AppName
|
||||
logpath: ${user.home}/logs/xxl-job/${spring.application.name} # 执行器运行日志文件存储磁盘路径
|
||||
accessToken: default_token # 执行器通讯TOKEN
|
||||
|
||||
--- #################### 芋道相关配置 ####################
|
||||
|
||||
viewsh:
|
||||
info:
|
||||
version: 1.0.0
|
||||
base-package: com.viewsh.module.iot
|
||||
web:
|
||||
admin-ui:
|
||||
url: http://dashboard.viewsh.iocoder.cn # Admin 管理后台 UI 的地址
|
||||
xss:
|
||||
enable: false
|
||||
exclude-urls: # 如下 url，仅仅是为了演示，去掉配置也没关系
|
||||
- ${management.endpoints.web.base-path}/** # 不处理 Actuator 的请求
|
||||
swagger:
|
||||
title: 管理后台
|
||||
description: 提供管理员管理的所有功能
|
||||
version: ${viewsh.info.version}
|
||||
tenant: # 多租户相关配置项
|
||||
enable: true
|
||||
ignore-urls:
|
||||
ignore-tables:
|
||||
ignore-caches:
|
||||
- iot:device
|
||||
- iot:thing_model_list
|
||||
iot:
|
||||
message-bus:
|
||||
type: redis # 消息总线的类型
|
||||
# 跨模块事件总线配置（IntegrationEventBus）
|
||||
integration:
|
||||
mq:
|
||||
enabled: true # 是否启用跨模块事件发布
|
||||
producer-group: integration-event-producer # 生产者组名
|
||||
send-timeout-ms: 3000 # 发送超时时间（毫秒）
|
||||
|
||||
debug: false
|
||||
Reference in New Issue
Block a user