feat(device): 添加MQTT设备在线状态管理与Redis容错机制
- 集成Spring Integration MQTT与Eclipse Paho客户端 - 添加设备在线状态填充逻辑，支持批量查询心跳数据 - 优化Redis配置，使用Lazy初始化与通用序列化器 - 增加Redis异常处理与防重复提交Fail-Open机制 - 新增Redis设备模块常量与在线状态VO - 完善MQTT与Redis集群连接配置
This commit is contained in:
TongTongStudio
24
pom.xml
24
pom.xml
@@ -64,6 +64,12 @@
|
||||
|
||||
<weixin-java-miniapp.version>4.8.1.B</weixin-java-miniapp.version>
|
||||
|
||||
<!-- Spring Integration MQTT 版本 (Spring Boot 4 / Spring Framework 7 配套版本) -->
|
||||
<spring-integration.version>7.0.4</spring-integration.version>
|
||||
|
||||
<!-- Eclipse Paho MQTT 客户端版本 -->
|
||||
<paho-client.version>1.2.5</paho-client.version>
|
||||
|
||||
</properties>
|
||||
|
||||
<dependencies>
|
||||
@@ -328,6 +334,24 @@
|
||||
<version>3.1.1451</version>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-integration</artifactId>
|
||||
</dependency>
|
||||
<!-- Spring Integration MQTT (与 Spring Boot 4 / Spring Framework 7 配套的 7.x 版本) -->
|
||||
<dependency>
|
||||
<groupId>org.springframework.integration</groupId>
|
||||
<artifactId>spring-integration-mqtt</artifactId>
|
||||
<version>${spring-integration.version}</version>
|
||||
</dependency>
|
||||
|
||||
<!-- Eclipse Paho MQTT 客户端 (org.eclipse.paho.client.mqttv3) -->
|
||||
<dependency>
|
||||
<groupId>org.eclipse.paho</groupId>
|
||||
<artifactId>org.eclipse.paho.client.mqttv3</artifactId>
|
||||
<version>${paho-client.version}</version>
|
||||
</dependency>
|
||||
|
||||
</dependencies>
|
||||
|
||||
<build>
|
||||
|
||||
@@ -11,6 +11,7 @@ import com.youlai.boot.common.annotation.RepeatSubmit;
|
||||
import com.youlai.boot.common.util.IPUtils;
|
||||
import jakarta.servlet.http.HttpServletRequest;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.aspectj.lang.ProceedingJoinPoint;
|
||||
import org.aspectj.lang.annotation.Around;
|
||||
import org.aspectj.lang.annotation.Aspect;
|
||||
@@ -33,6 +34,7 @@ import java.util.concurrent.TimeUnit;
|
||||
@Aspect
|
||||
@Component
|
||||
@RequiredArgsConstructor
|
||||
@Slf4j
|
||||
public class RepeatSubmitAspect {
|
||||
|
||||
private final RedissonClient redissonClient;
|
||||
@@ -46,11 +48,18 @@ public class RepeatSubmitAspect {
|
||||
String lockKey = buildLockKey(pjp);
|
||||
|
||||
int expire = repeatSubmit.expire();
|
||||
RLock lock = redissonClient.getLock(lockKey);
|
||||
|
||||
boolean locked = lock.tryLock(0, expire, TimeUnit.SECONDS);
|
||||
if (!locked) {
|
||||
throw new BusinessException(ResultCode.DUPLICATE_SUBMISSION);
|
||||
// Redis 不可用时 Fail-Open：跳过防重复提交检查，放行请求
|
||||
try {
|
||||
RLock lock = redissonClient.getLock(lockKey);
|
||||
boolean locked = lock.tryLock(0, expire, TimeUnit.SECONDS);
|
||||
if (!locked) {
|
||||
throw new BusinessException(ResultCode.DUPLICATE_SUBMISSION);
|
||||
}
|
||||
} catch (BusinessException e) {
|
||||
throw e;
|
||||
} catch (Exception e) {
|
||||
log.warn("防重复提交 Redis 异常，Fail-Open 放行 lockKey={}", lockKey, e);
|
||||
}
|
||||
return pjp.proceed();
|
||||
}
|
||||
|
||||
@@ -66,4 +66,12 @@ public interface RedisConstants {
|
||||
String ROLE_PERMS = "system:role:perms"; // 系统角色和权限映射
|
||||
}
|
||||
|
||||
/**
|
||||
* 设备模块
|
||||
*/
|
||||
interface Device {
|
||||
/** 设备在线状态（示例：device:online:{sn}，值为 JSON，key 过期即视为离线） */
|
||||
String ONLINE = "device:online:{}";
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -135,7 +135,12 @@ public enum ResultCode implements IResultCode, Serializable {
|
||||
DATABASE_EXECUTION_ERROR("C0310", "数据库执行异常"),
|
||||
DATABASE_EXECUTION_SYNTAX_ERROR("C0313", "数据库执行语法错误"),
|
||||
INTEGRITY_CONSTRAINT_VIOLATION("C0342", "违反了完整性约束"),
|
||||
DATABASE_ACCESS_DENIED("C0351", "演示环境已禁用数据库写入功能，请本地部署修改数据库链接或开启Mock模式进行体验");
|
||||
DATABASE_ACCESS_DENIED("C0351", "演示环境已禁用数据库写入功能，请本地部署修改数据库链接或开启Mock模式进行体验"),
|
||||
|
||||
/** C04xx：Redis 服务错误 */
|
||||
REDIS_SERVICE_ERROR("C0400", "Redis 服务出错"),
|
||||
REDIS_CONNECTION_FAILURE("C0401", "Redis 连接失败"),
|
||||
REDIS_OPERATION_ERROR("C0402", "Redis 操作异常");
|
||||
|
||||
private final String code;
|
||||
|
||||
|
||||
313
src/main/java/com/youlai/boot/common/util/RedisSafeOps.java
Normal file
313
src/main/java/com/youlai/boot/common/util/RedisSafeOps.java
Normal file
@@ -0,0 +1,313 @@
|
||||
package com.youlai.boot.common.util;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.data.redis.core.RedisTemplate;
|
||||
import org.springframework.data.redis.core.StringRedisTemplate;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.*;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
/**
|
||||
* Redis 安全操作工具类
|
||||
* <p>
|
||||
* 对所有 Redis 操作提供 Fail-Open 保护：
|
||||
* Redis 不可用时不会抛异常导致业务中断，而是返回默认值/空值。
|
||||
* <p>
|
||||
* 使用方式：
|
||||
* <pre>
|
||||
* // 原有写法（Redis 不可用会抛异常）
|
||||
* String value = redisTemplate.opsForValue().get(key);
|
||||
*
|
||||
* // 安全写法（Redis 不可用返回 null，不中断业务）
|
||||
* String value = RedisSafeOps.get(() -> redisTemplate.opsForValue().get(key));
|
||||
* </pre>
|
||||
*
|
||||
* @since 4.6.0
|
||||
*/
|
||||
@Slf4j
|
||||
@Component
|
||||
public class RedisSafeOps {
|
||||
|
||||
private static RedisTemplate<String, Object> redisTemplate;
|
||||
private static StringRedisTemplate stringRedisTemplate;
|
||||
|
||||
@Autowired
|
||||
public void setRedisTemplate(RedisTemplate<String, Object> redisTemplate) {
|
||||
RedisSafeOps.redisTemplate = redisTemplate;
|
||||
}
|
||||
|
||||
@Autowired(required = false)
|
||||
public void setStringRedisTemplate(StringRedisTemplate stringRedisTemplate) {
|
||||
RedisSafeOps.stringRedisTemplate = stringRedisTemplate;
|
||||
}
|
||||
|
||||
// ==================== 通用安全执行 ====================
|
||||
|
||||
/**
|
||||
* 安全执行 Redis 操作，异常时返回 null
|
||||
*/
|
||||
public static <T> T get(Supplier<T> supplier) {
|
||||
return get(supplier, null);
|
||||
}
|
||||
|
||||
/**
|
||||
* 安全执行 Redis 操作，异常时返回默认值
|
||||
*/
|
||||
public static <T> T get(Supplier<T> supplier, T defaultValue) {
|
||||
try {
|
||||
return supplier.get();
|
||||
} catch (Exception e) {
|
||||
log.warn("Redis 操作失败，返回默认值: {}", defaultValue, e);
|
||||
return defaultValue;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 安全执行 Redis 写操作，返回是否成功
|
||||
*/
|
||||
public static boolean execute(Runnable runnable) {
|
||||
try {
|
||||
runnable.run();
|
||||
return true;
|
||||
} catch (Exception e) {
|
||||
log.warn("Redis 写操作失败", e);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
// ==================== String 类型 ====================
|
||||
|
||||
/**
|
||||
* 获取 String 值
|
||||
*/
|
||||
public static String stringGet(String key) {
|
||||
return get(() -> stringRedisTemplate != null
|
||||
? stringRedisTemplate.opsForValue().get(key)
|
||||
: (String) redisTemplate.opsForValue().get(key));
|
||||
}
|
||||
|
||||
/**
|
||||
* 设置 String 值
|
||||
*/
|
||||
public static boolean stringSet(String key, String value) {
|
||||
return execute(() -> {
|
||||
if (stringRedisTemplate != null) {
|
||||
stringRedisTemplate.opsForValue().set(key, value);
|
||||
} else {
|
||||
redisTemplate.opsForValue().set(key, value);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* 设置 String 值（带过期时间）
|
||||
*/
|
||||
public static boolean stringSet(String key, String value, Duration timeout) {
|
||||
return execute(() -> {
|
||||
if (stringRedisTemplate != null) {
|
||||
stringRedisTemplate.opsForValue().set(key, value, timeout);
|
||||
} else {
|
||||
redisTemplate.opsForValue().set(key, value, timeout);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* 设置 String 值（带过期时间，单位秒）
|
||||
*/
|
||||
public static boolean stringSet(String key, String value, long timeoutSeconds) {
|
||||
return execute(() -> {
|
||||
if (stringRedisTemplate != null) {
|
||||
stringRedisTemplate.opsForValue().set(key, value, timeoutSeconds, TimeUnit.SECONDS);
|
||||
} else {
|
||||
redisTemplate.opsForValue().set(key, value, timeoutSeconds, TimeUnit.SECONDS);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取并设置新值
|
||||
*/
|
||||
public static String stringGetAndSet(String key, String value) {
|
||||
return get(() -> stringRedisTemplate != null
|
||||
? stringRedisTemplate.opsForValue().getAndSet(key, value)
|
||||
: (String) redisTemplate.opsForValue().getAndSet(key, value));
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取并删除
|
||||
*/
|
||||
public static String stringGetAndDelete(String key) {
|
||||
return get(() -> stringRedisTemplate != null
|
||||
? stringRedisTemplate.opsForValue().getAndDelete(key)
|
||||
: (String) redisTemplate.opsForValue().getAndDelete(key));
|
||||
}
|
||||
|
||||
/**
|
||||
* 删除 key
|
||||
*/
|
||||
public static boolean delete(String key) {
|
||||
return execute(() -> redisTemplate.delete(key));
|
||||
}
|
||||
|
||||
/**
|
||||
* 批量删除 key
|
||||
*/
|
||||
public static boolean delete(Collection<String> keys) {
|
||||
return execute(() -> redisTemplate.delete(keys));
|
||||
}
|
||||
|
||||
/**
|
||||
* 设置过期时间
|
||||
*/
|
||||
public static boolean expire(String key, Duration timeout) {
|
||||
return Boolean.TRUE.equals(get(() -> redisTemplate.expire(key, timeout), false));
|
||||
}
|
||||
|
||||
/**
|
||||
* 判断 key 是否存在
|
||||
*/
|
||||
public static boolean hasKey(String key) {
|
||||
return Boolean.TRUE.equals(get(() -> redisTemplate.hasKey(key), false));
|
||||
}
|
||||
|
||||
// ==================== Hash 类型 ====================
|
||||
|
||||
/**
|
||||
* Hash 获取
|
||||
*/
|
||||
@SuppressWarnings("unchecked")
|
||||
public static <T> T hashGet(String key, String hashKey) {
|
||||
return get(() -> (T) redisTemplate.opsForHash().get(key, hashKey));
|
||||
}
|
||||
|
||||
/**
|
||||
* Hash 设置
|
||||
*/
|
||||
public static boolean hashPut(String key, String hashKey, Object value) {
|
||||
return execute(() -> redisTemplate.opsForHash().put(key, hashKey, value));
|
||||
}
|
||||
|
||||
/**
|
||||
* Hash 批量设置
|
||||
*/
|
||||
public static boolean hashPutAll(String key, Map<String, Object> map) {
|
||||
return execute(() -> redisTemplate.opsForHash().putAll(key, map));
|
||||
}
|
||||
|
||||
/**
|
||||
* Hash 获取所有
|
||||
*/
|
||||
@SuppressWarnings("unchecked")
|
||||
public static <K, V> Map<K, V> hashEntries(String key) {
|
||||
return get(() -> {
|
||||
Map<Object, Object> entries = redisTemplate.opsForHash().entries(key);
|
||||
Map<K, V> result = new LinkedHashMap<>();
|
||||
entries.forEach((k, v) -> result.put((K) k, (V) v));
|
||||
return result;
|
||||
}, Collections.emptyMap());
|
||||
}
|
||||
|
||||
/**
|
||||
* Hash 删除
|
||||
*/
|
||||
public static boolean hashDelete(String key, Object... hashKeys) {
|
||||
return execute(() -> redisTemplate.opsForHash().delete(key, hashKeys));
|
||||
}
|
||||
|
||||
// ==================== Set 类型 ====================
|
||||
|
||||
/**
|
||||
* Set 添加
|
||||
*/
|
||||
public static boolean setAdd(String key, Object... values) {
|
||||
return execute(() -> redisTemplate.opsForSet().add(key, values));
|
||||
}
|
||||
|
||||
/**
|
||||
* Set 判断是否成员
|
||||
*/
|
||||
public static boolean setIsMember(String key, Object value) {
|
||||
return Boolean.TRUE.equals(get(() -> redisTemplate.opsForSet().isMember(key, value), false));
|
||||
}
|
||||
|
||||
/**
|
||||
* Set 获取所有成员
|
||||
*/
|
||||
@SuppressWarnings("unchecked")
|
||||
public static <T> Set<T> setMembers(String key) {
|
||||
return get(() -> {
|
||||
Set<Object> members = redisTemplate.opsForSet().members(key);
|
||||
if (members == null) return Collections.emptySet();
|
||||
Set<T> result = new LinkedHashSet<>();
|
||||
members.forEach(m -> result.add((T) m));
|
||||
return result;
|
||||
}, Collections.emptySet());
|
||||
}
|
||||
|
||||
/**
|
||||
* Set 删除成员
|
||||
*/
|
||||
public static boolean setRemove(String key, Object... values) {
|
||||
return execute(() -> redisTemplate.opsForSet().remove(key, values));
|
||||
}
|
||||
|
||||
// ==================== List 类型 ====================
|
||||
|
||||
/**
|
||||
* List 右侧推入
|
||||
*/
|
||||
public static boolean listRightPush(String key, Object value) {
|
||||
return execute(() -> redisTemplate.opsForList().rightPush(key, value));
|
||||
}
|
||||
|
||||
/**
|
||||
* List 左侧弹出
|
||||
*/
|
||||
@SuppressWarnings("unchecked")
|
||||
public static <T> T listLeftPop(String key) {
|
||||
return get(() -> (T) redisTemplate.opsForList().leftPop(key));
|
||||
}
|
||||
|
||||
/**
|
||||
* List 范围获取
|
||||
*/
|
||||
@SuppressWarnings("unchecked")
|
||||
public static <T> List<T> listRange(String key, long start, long end) {
|
||||
return get(() -> (List<T>) redisTemplate.opsForList().range(key, start, end), Collections.emptyList());
|
||||
}
|
||||
|
||||
// ==================== Lua 脚本 ====================
|
||||
|
||||
/**
|
||||
* 执行 Lua 脚本（返回 Long）
|
||||
*/
|
||||
public static Long executeScript(String script, List<String> keys, Object... args) {
|
||||
return get(() -> {
|
||||
var result = redisTemplate.execute(
|
||||
new org.springframework.data.redis.core.script.DefaultRedisScript<>(script, Long.class),
|
||||
keys, args);
|
||||
return result;
|
||||
}, 0L);
|
||||
}
|
||||
|
||||
// ==================== 原始 RedisTemplate 访问 ====================
|
||||
|
||||
/**
|
||||
* 获取原始 RedisTemplate（仅在确定 Redis 可用时使用）
|
||||
*/
|
||||
public static RedisTemplate<String, Object> template() {
|
||||
return redisTemplate;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取 StringRedisTemplate（仅在确定 Redis 可用时使用）
|
||||
*/
|
||||
public static StringRedisTemplate stringTemplate() {
|
||||
return stringRedisTemplate;
|
||||
}
|
||||
}
|
||||
117
src/main/java/com/youlai/boot/config/MqttConfig.java
Normal file
117
src/main/java/com/youlai/boot/config/MqttConfig.java
Normal file
@@ -0,0 +1,117 @@
|
||||
package com.youlai.boot.config;
|
||||
|
||||
import com.youlai.boot.config.property.MqttProperties;
|
||||
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.integration.annotation.ServiceActivator;
|
||||
import org.springframework.integration.channel.DirectChannel;
|
||||
import org.springframework.integration.core.MessageProducer;
|
||||
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
|
||||
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
|
||||
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
import org.springframework.messaging.MessageHandler;
|
||||
|
||||
/**
|
||||
* MQTT 配置
|
||||
* <p>
|
||||
* 基于 spring-integration-mqtt (Eclipse Paho) 实现消息的发布与订阅。
|
||||
* 通过 {@code mqtt.enabled=false} 可整体关闭 MQTT 功能。
|
||||
*
|
||||
* @author TongTongStudio
|
||||
* @since 2026/8/8
|
||||
*/
|
||||
@Configuration
|
||||
@ConditionalOnProperty(prefix = "mqtt", name = "enabled", havingValue = "true")
|
||||
public class MqttConfig {
|
||||
|
||||
/**
|
||||
* MQTT 连接工厂
|
||||
*/
|
||||
@Bean
|
||||
public DefaultMqttPahoClientFactory mqttClientFactory(MqttProperties properties) {
|
||||
MqttConnectOptions options = new MqttConnectOptions();
|
||||
options.setServerURIs(new String[]{normalizeServerUri(properties.getUrl())});
|
||||
if (properties.getUsername() != null && !properties.getUsername().isEmpty()) {
|
||||
options.setUserName(properties.getUsername());
|
||||
}
|
||||
if (properties.getPassword() != null && !properties.getPassword().isEmpty()) {
|
||||
options.setPassword(properties.getPassword().toCharArray());
|
||||
}
|
||||
options.setConnectionTimeout(properties.getConnectionTimeout());
|
||||
options.setKeepAliveInterval(properties.getKeepAliveInterval());
|
||||
options.setAutomaticReconnect(properties.isAutomaticReconnect());
|
||||
options.setCleanSession(true);
|
||||
|
||||
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
|
||||
factory.setConnectionOptions(options);
|
||||
return factory;
|
||||
}
|
||||
|
||||
/**
|
||||
* 将 Broker 地址统一规范为 Paho 支持的 scheme。
|
||||
* <p>Eclipse Paho 仅内置 {@code tcp://} 与 {@code ssl://} 两种 scheme，
|
||||
* 配置中常见的 {@code mqtt://} 需转换为 {@code tcp://}（明文）或 {@code ssl://}（加密）。</p>
|
||||
*/
|
||||
private String normalizeServerUri(String uri) {
|
||||
if (uri == null || uri.isEmpty()) {
|
||||
return uri;
|
||||
}
|
||||
if (uri.startsWith("mqtt://")) {
|
||||
return "tcp://" + uri.substring("mqtt://".length());
|
||||
}
|
||||
if (uri.startsWith("mqtts://")) {
|
||||
return "ssl://" + uri.substring("mqtts://".length());
|
||||
}
|
||||
return uri;
|
||||
}
|
||||
|
||||
/* ============================ 发布 (Publish) ============================ */
|
||||
|
||||
/**
|
||||
* 发布消息的出站通道
|
||||
*/
|
||||
@Bean
|
||||
public MessageChannel mqttOutboundChannel() {
|
||||
return new DirectChannel();
|
||||
}
|
||||
|
||||
/**
|
||||
* 发布消息处理器，通过 {@link org.springframework.messaging.support.MessageBuilder} 发送到出站通道即可发布
|
||||
*/
|
||||
@Bean
|
||||
@ServiceActivator(inputChannel = "mqttOutboundChannel")
|
||||
public MessageHandler mqttOutbound(DefaultMqttPahoClientFactory factory, MqttProperties properties) {
|
||||
MqttPahoMessageHandler handler = new MqttPahoMessageHandler(properties.getClientId() + "-pub", factory);
|
||||
handler.setAsync(false);
|
||||
handler.setDefaultQos(properties.getQos());
|
||||
return handler;
|
||||
}
|
||||
|
||||
/* ============================ 订阅 (Subscribe) ============================ */
|
||||
|
||||
/**
|
||||
* 订阅消息的入站通道
|
||||
*/
|
||||
@Bean
|
||||
public MessageChannel mqttInboundChannel() {
|
||||
return new DirectChannel();
|
||||
}
|
||||
|
||||
/**
|
||||
* 入站订阅适配器：订阅 {@link MqttProperties#getTopics()} 中配置的主题
|
||||
*/
|
||||
@Bean
|
||||
public MessageProducer mqttInbound(DefaultMqttPahoClientFactory factory, MqttProperties properties) {
|
||||
String[] topics = properties.getTopics().toArray(new String[0]);
|
||||
MqttPahoMessageDrivenChannelAdapter adapter =
|
||||
new MqttPahoMessageDrivenChannelAdapter(
|
||||
properties.getClientId() + "-sub", factory, topics);
|
||||
adapter.setCompletionTimeout(5000L);
|
||||
adapter.setQos(properties.getQos());
|
||||
adapter.setOutputChannel(mqttInboundChannel());
|
||||
return adapter;
|
||||
}
|
||||
}
|
||||
@@ -1,33 +1,42 @@
|
||||
package com.youlai.boot.config;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.context.annotation.Lazy;
|
||||
import org.springframework.data.redis.connection.RedisConnectionFactory;
|
||||
import org.springframework.data.redis.core.RedisTemplate;
|
||||
import org.springframework.data.redis.serializer.JacksonJsonRedisSerializer;
|
||||
import org.springframework.data.redis.serializer.GenericJacksonJsonRedisSerializer;
|
||||
import org.springframework.data.redis.serializer.RedisSerializer;
|
||||
import tools.jackson.databind.cfg.DateTimeFeature;
|
||||
import tools.jackson.databind.json.JsonMapper;
|
||||
import tools.jackson.databind.ObjectMapper;
|
||||
|
||||
/**
|
||||
* Redis 配置
|
||||
* <p>
|
||||
* 使用 Lazy 初始化，避免 Redis 不可用时阻塞应用启动。
|
||||
*
|
||||
* @author Ray.Hao
|
||||
* @since 2023/5/15
|
||||
*/
|
||||
@Slf4j
|
||||
@Configuration
|
||||
@ConditionalOnClass(RedisConnectionFactory.class)
|
||||
public class RedisConfig {
|
||||
|
||||
/**
|
||||
* 自定义 RedisTemplate
|
||||
* <p>
|
||||
* 修改 Redis 序列化方式，默认 JdkSerializationRedisSerializer
|
||||
* 修改 Redis 序列化方式，默认 JdkSerializationRedisSerializer。
|
||||
* 延迟初始化：即使 Redis 不可用，应用也能正常启动。
|
||||
*
|
||||
* @param redisConnectionFactory {@link RedisConnectionFactory}
|
||||
* @return {@link RedisTemplate}
|
||||
*/
|
||||
@Bean
|
||||
@Lazy
|
||||
public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
|
||||
log.info("初始化 RedisTemplate...");
|
||||
|
||||
RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
|
||||
redisTemplate.setConnectionFactory(redisConnectionFactory);
|
||||
@@ -36,15 +45,14 @@ public class RedisConfig {
|
||||
redisTemplate.setKeySerializer(RedisSerializer.string());
|
||||
redisTemplate.setHashKeySerializer(RedisSerializer.string());
|
||||
|
||||
JsonMapper jsonMapper = JsonMapper.builder()
|
||||
.disable(DateTimeFeature.WRITE_DATES_AS_TIMESTAMPS)
|
||||
.build();
|
||||
JacksonJsonRedisSerializer<Object> jsonSerializer = new JacksonJsonRedisSerializer<>(jsonMapper, Object.class);
|
||||
GenericJacksonJsonRedisSerializer jsonSerializer = new GenericJacksonJsonRedisSerializer(new ObjectMapper());
|
||||
|
||||
redisTemplate.setValueSerializer(jsonSerializer);
|
||||
redisTemplate.setHashValueSerializer(jsonSerializer);
|
||||
|
||||
redisTemplate.afterPropertiesSet();
|
||||
|
||||
log.info("RedisTemplate 初始化完成");
|
||||
return redisTemplate;
|
||||
}
|
||||
|
||||
|
||||
64
src/main/java/com/youlai/boot/config/RedisHealthConfig.java
Normal file
64
src/main/java/com/youlai/boot/config/RedisHealthConfig.java
Normal file
@@ -0,0 +1,64 @@
|
||||
package com.youlai.boot.config;
|
||||
|
||||
import io.lettuce.core.ClientOptions;
|
||||
import io.lettuce.core.SocketOptions;
|
||||
import io.lettuce.core.resource.ClientResources;
|
||||
import io.lettuce.core.resource.DefaultClientResources;
|
||||
import io.lettuce.core.resource.Delay;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.context.annotation.Primary;
|
||||
import org.springframework.data.redis.connection.RedisConnectionFactory;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
/**
|
||||
* Redis 健康检查与启动容错配置
|
||||
* <p>
|
||||
* 确保 Redis 不可用时程序仍能正常启动，并在 Redis 恢复后自动重连。
|
||||
*
|
||||
* @since 4.6.0
|
||||
*/
|
||||
@Slf4j
|
||||
@Configuration
|
||||
@ConditionalOnClass(RedisConnectionFactory.class)
|
||||
public class RedisHealthConfig {
|
||||
|
||||
/**
|
||||
* 自定义 Lettuce ClientResources，增加优雅关闭和重连支持
|
||||
*/
|
||||
@Bean(destroyMethod = "shutdown")
|
||||
@Primary
|
||||
public ClientResources lettuceClientResources() {
|
||||
DefaultClientResources resources = DefaultClientResources.builder()
|
||||
.reconnectDelay(Delay.exponential(Duration.ofSeconds(2), Duration.ofSeconds(30), 2, TimeUnit.SECONDS))
|
||||
.build();
|
||||
log.info("Lettuce ClientResources 初始化完成，重连延迟: 2s");
|
||||
return resources;
|
||||
}
|
||||
|
||||
/**
|
||||
* 自定义 ClientOptions，配置断开重连策略
|
||||
*/
|
||||
@Bean
|
||||
@Primary
|
||||
public ClientOptions lettuceClientOptions() {
|
||||
ClientOptions options = ClientOptions.builder()
|
||||
.socketOptions(SocketOptions.builder()
|
||||
.connectTimeout(Duration.ofSeconds(5)) // 连接超时 5 秒
|
||||
.keepAlive(true) // 启用 TCP KeepAlive
|
||||
.build())
|
||||
// 断开后自动重连
|
||||
.autoReconnect(true)
|
||||
// 取消正在执行的命令超时（避免连接断开时命令永久挂起）
|
||||
.cancelCommandsOnReconnectFailure(true)
|
||||
// 连接断开时挂起的请求超时
|
||||
.suspendReconnectOnProtocolFailure(true)
|
||||
.build();
|
||||
log.info("Lettuce ClientOptions 初始化完成，autoReconnect=true");
|
||||
return options;
|
||||
}
|
||||
}
|
||||
88
src/main/java/com/youlai/boot/config/RedissonConfig.java
Normal file
88
src/main/java/com/youlai/boot/config/RedissonConfig.java
Normal file
@@ -0,0 +1,88 @@
|
||||
package com.youlai.boot.config;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.redisson.Redisson;
|
||||
import org.redisson.api.RedissonClient;
|
||||
import org.redisson.config.Config;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.context.annotation.Lazy;
|
||||
import org.springframework.context.annotation.Primary;
|
||||
|
||||
/**
|
||||
* Redisson 分布式锁配置
|
||||
* <p>
|
||||
* 延迟初始化 + 连接容错：Redis 不可用时不影响应用启动，
|
||||
* 首次实际使用时才建立连接。
|
||||
*
|
||||
* @since 4.6.0
|
||||
*/
|
||||
@Slf4j
|
||||
@Configuration
|
||||
@ConditionalOnClass(Redisson.class)
|
||||
@ConditionalOnProperty(prefix = "spring.data.redis", name = "host")
|
||||
public class RedissonConfig {
|
||||
|
||||
@Value("${spring.data.redis.host:127.0.0.1}")
|
||||
private String host;
|
||||
|
||||
@Value("${spring.data.redis.port:6379}")
|
||||
private int port;
|
||||
|
||||
@Value("${spring.data.redis.password:}")
|
||||
private String password;
|
||||
|
||||
@Value("${spring.data.redis.database:0}")
|
||||
private int database;
|
||||
|
||||
@Value("${spring.data.redis.timeout:10s}")
|
||||
private String timeout;
|
||||
|
||||
@Bean(destroyMethod = "shutdown")
|
||||
@Primary
|
||||
@Lazy
|
||||
public RedissonClient redissonClient() {
|
||||
log.info("初始化 RedissonClient (延迟初始化)...");
|
||||
|
||||
String address = "redis://" + host + ":" + port;
|
||||
Config config = new Config();
|
||||
config.useSingleServer()
|
||||
.setAddress(address)
|
||||
.setDatabase(database)
|
||||
.setPassword(password.isBlank() ? null : password)
|
||||
// 连接超时
|
||||
.setConnectTimeout((int) parseDurationMillis(timeout))
|
||||
// 命令等待超时
|
||||
.setTimeout(3000)
|
||||
// 连接池大小
|
||||
.setConnectionPoolSize(8)
|
||||
.setConnectionMinimumIdleSize(2)
|
||||
// 重连间隔
|
||||
.setRetryInterval(2000)
|
||||
// 重试次数（3 次后放弃本次操作，不会一直阻塞）
|
||||
.setRetryAttempts(3);
|
||||
|
||||
RedissonClient client = Redisson.create(config);
|
||||
log.info("RedissonClient 初始化完成，address={}", address);
|
||||
return client;
|
||||
}
|
||||
|
||||
/**
|
||||
* 简单解析 duration 字符串为毫秒数
|
||||
*/
|
||||
private long parseDurationMillis(String duration) {
|
||||
if (duration == null || duration.isBlank()) return 10_000;
|
||||
String s = duration.trim().toLowerCase();
|
||||
try {
|
||||
if (s.endsWith("ms")) return Long.parseLong(s.replace("ms", ""));
|
||||
if (s.endsWith("s")) return Long.parseLong(s.replace("s", "")) * 1000;
|
||||
if (s.endsWith("m")) return Long.parseLong(s.replace("m", "")) * 60_000;
|
||||
return Long.parseLong(s);
|
||||
} catch (NumberFormatException e) {
|
||||
return 10_000;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,64 @@
|
||||
package com.youlai.boot.config.property;
|
||||
|
||||
import lombok.Data;
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* MQTT 配置属性
|
||||
* <p>
|
||||
* 始终注册为 Bean，供设备在线状态服务等使用（不受 {@code mqtt.enabled} 开关影响）。
|
||||
*
|
||||
* @author TongTongStudio
|
||||
* @since 2026/8/8
|
||||
*/
|
||||
@Component
|
||||
@ConfigurationProperties(prefix = "mqtt")
|
||||
@Data
|
||||
public class MqttProperties {
|
||||
|
||||
/** 是否启用 MQTT */
|
||||
private boolean enabled = false;
|
||||
|
||||
/** Broker 地址，例如 tcp://192.168.5.224:1883（Paho 仅支持 tcp:// 与 ssl://） */
|
||||
private String url = "tcp://192.168.5.224:1883";
|
||||
|
||||
/** 用户名 */
|
||||
private String username;
|
||||
|
||||
/** 密码 */
|
||||
private String password;
|
||||
|
||||
/** 客户端 ID */
|
||||
private String clientId = "youlai-boot-server";
|
||||
|
||||
/** 默认 QoS (0-2) */
|
||||
private int qos = 1;
|
||||
|
||||
/** 连接超时时间(秒) */
|
||||
private int connectionTimeout = 10;
|
||||
|
||||
/** 保持连接时间(秒) */
|
||||
private int keepAliveInterval = 60;
|
||||
|
||||
/** 是否自动重连 */
|
||||
private boolean automaticReconnect = true;
|
||||
|
||||
/** 订阅的主题列表 */
|
||||
private List<String> topics = new ArrayList<>();
|
||||
|
||||
/** 设备心跳主题（用于接收设备状态/心跳上报） */
|
||||
private String statusTopic = "ttstd/device/status";
|
||||
|
||||
/**
|
||||
* 设备在线判定超时时间(秒)
|
||||
* <p>
|
||||
* 设备周期性上报心跳，每收到一次心跳刷新该 key 的 TTL；超过该时间未上报即自动判定为离线。
|
||||
* 建议设置为心跳间隔的 3 倍左右，默认 90 秒（对应心跳间隔 30 秒）。
|
||||
*/
|
||||
private long onlineTimeout = 90;
|
||||
|
||||
}
|
||||
@@ -0,0 +1,52 @@
|
||||
package com.youlai.boot.device.controller;
|
||||
|
||||
import com.youlai.boot.common.result.Result;
|
||||
import com.youlai.boot.device.model.vo.DeviceOnlineVO;
|
||||
import com.youlai.boot.device.service.DeviceOnlineService;
|
||||
import io.swagger.v3.oas.annotations.Operation;
|
||||
import io.swagger.v3.oas.annotations.Parameter;
|
||||
import io.swagger.v3.oas.annotations.tags.Tag;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||
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.RequestMapping;
|
||||
import org.springframework.web.bind.annotation.RequestParam;
|
||||
import org.springframework.web.bind.annotation.RestController;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* 设备在线状态控制层
|
||||
* <p>
|
||||
* 基于 Redis 记录的设备在线/离线状态查询接口。
|
||||
*
|
||||
* @author TongTongStudio
|
||||
* @since 2026/8/8
|
||||
*/
|
||||
@Tag(name = "设备在线状态")
|
||||
@RestController
|
||||
@RequestMapping("/api/v1/device/online")
|
||||
@RequiredArgsConstructor
|
||||
@ConditionalOnProperty(prefix = "mqtt", name = "enabled", havingValue = "true")
|
||||
public class DeviceOnlineController {
|
||||
|
||||
private final DeviceOnlineService deviceOnlineService;
|
||||
|
||||
@Operation(summary = "查询单个设备在线状态")
|
||||
@GetMapping("/status")
|
||||
public Result<DeviceOnlineVO> getOnlineStatus(
|
||||
@Parameter(description = "设备序列号") @RequestParam String sn) {
|
||||
return Result.success(deviceOnlineService.getOnlineInfo(sn));
|
||||
}
|
||||
|
||||
@Operation(summary = "批量查询设备在线状态")
|
||||
@PostMapping("/status/batch")
|
||||
public Result<Map<String, DeviceOnlineVO>> getOnlineStatusBatch(
|
||||
@Parameter(description = "设备序列号列表") @RequestBody List<String> sns) {
|
||||
return Result.success(deviceOnlineService.getOnlineStatusBySns(sns));
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,33 @@
|
||||
package com.youlai.boot.device.model.vo;
|
||||
|
||||
import io.swagger.v3.oas.annotations.media.Schema;
|
||||
import lombok.Data;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
|
||||
/**
|
||||
* 设备在线状态视图对象
|
||||
*
|
||||
* @author TongTongStudio
|
||||
* @since 2026/8/8
|
||||
*/
|
||||
@Schema(description = "设备在线状态")
|
||||
@Data
|
||||
public class DeviceOnlineVO {
|
||||
|
||||
@Schema(description = "设备序列号")
|
||||
private String serialno;
|
||||
|
||||
@Schema(description = "在线状态(1:在线;0:离线)")
|
||||
private Integer online;
|
||||
|
||||
@Schema(description = "最后心跳时间")
|
||||
private LocalDateTime lastHeartbeatTime;
|
||||
|
||||
@Schema(description = "设备 IP(可选)")
|
||||
private String ip;
|
||||
|
||||
@Schema(description = "屏幕状态(1:亮屏;0:熄屏;null:未知)，由设备上报")
|
||||
private Integer screenState;
|
||||
|
||||
}
|
||||
@@ -40,6 +40,13 @@ public class DevicePageVO {
|
||||
@Schema(description="是否删除(1:删除;0:未删除)")
|
||||
private Integer isDelete;
|
||||
|
||||
@Schema(description="在线状态(1:在线;0:离线)")
|
||||
private Integer online;
|
||||
|
||||
@Schema(description="最后心跳时间")
|
||||
@JsonFormat(pattern = "yyyy/MM/dd HH:mm:ss")
|
||||
private LocalDateTime lastHeartbeatTime;
|
||||
|
||||
@Schema(description="激活时间")
|
||||
@JsonFormat(pattern = "yyyy/MM/dd HH:mm")
|
||||
private LocalDateTime activateTime;
|
||||
|
||||
@@ -0,0 +1,71 @@
|
||||
package com.youlai.boot.device.service;
|
||||
|
||||
import com.youlai.boot.device.model.vo.DeviceOnlineVO;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* 设备在线状态服务
|
||||
* <p>
|
||||
* 基于 Redis 记录设备在线/离线状态：
|
||||
* <ul>
|
||||
* <li>设备周期性上报心跳（经 MQTT 到达 {@code MqttMessageHandler}），每上报一次刷新 Redis key 的 TTL；</li>
|
||||
* <li>Redis key 使用带过期时间的 {@code SET key value EX timeout}，超时未上报则 key 自动过期，即判定设备离线；</li>
|
||||
* <li>查询时 key 存在 = 在线，不存在 = 离线。</li>
|
||||
* </ul>
|
||||
*
|
||||
* @author TongTongStudio
|
||||
* @since 2026/8/8
|
||||
*/
|
||||
public interface DeviceOnlineService {
|
||||
|
||||
/**
|
||||
* 设备上报心跳，记录在线状态（刷新 Redis TTL）
|
||||
*
|
||||
* @param sn 设备序列号
|
||||
* @param ip 设备 IP（可为空）
|
||||
*/
|
||||
void reportHeartbeat(String sn, String ip);
|
||||
|
||||
/**
|
||||
* 设备上报心跳，记录在线状态（刷新 Redis TTL）
|
||||
*
|
||||
* @param sn 设备序列号
|
||||
* @param ip 设备 IP（可为空）
|
||||
* @param screenState 屏幕状态(1:亮屏;0:熄屏;null:未知)，可为空
|
||||
*/
|
||||
void reportHeartbeat(String sn, String ip, Integer screenState);
|
||||
|
||||
/**
|
||||
* 设备主动离线（如 MQTT Last Will / 关机通知）
|
||||
*
|
||||
* @param sn 设备序列号
|
||||
*/
|
||||
void markOffline(String sn);
|
||||
|
||||
/**
|
||||
* 查询单个设备是否在线
|
||||
*
|
||||
* @param sn 设备序列号
|
||||
* @return true=在线, false=离线
|
||||
*/
|
||||
boolean isOnline(String sn);
|
||||
|
||||
/**
|
||||
* 获取单个设备在线状态信息
|
||||
*
|
||||
* @param sn 设备序列号
|
||||
* @return 在线状态信息（离线时 online=0）
|
||||
*/
|
||||
DeviceOnlineVO getOnlineInfo(String sn);
|
||||
|
||||
/**
|
||||
* 批量查询设备在线状态
|
||||
*
|
||||
* @param sns 设备序列号集合
|
||||
* @return key 为 SN，value 为在线状态信息
|
||||
*/
|
||||
Map<String, DeviceOnlineVO> getOnlineStatusBySns(List<String> sns);
|
||||
|
||||
}
|
||||
@@ -0,0 +1,147 @@
|
||||
package com.youlai.boot.device.service.impl;
|
||||
|
||||
import cn.hutool.core.util.StrUtil;
|
||||
import com.youlai.boot.common.constant.RedisConstants;
|
||||
import com.youlai.boot.config.property.MqttProperties;
|
||||
import com.youlai.boot.device.model.vo.DeviceOnlineVO;
|
||||
import com.youlai.boot.device.service.DeviceOnlineService;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.data.redis.core.RedisTemplate;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
import java.util.ArrayList;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
/**
|
||||
* 设备在线状态服务实现
|
||||
* <p>
|
||||
* 该 Bean 始终注册（不依赖 MQTT 开关），保证分页列表等无条件组件可正常注入；
|
||||
* 设备在线状态通过 Redis key 的 TTL 自动判定，MQTT 心跳仅为在线状态的一个数据来源。
|
||||
*
|
||||
* @author TongTongStudio
|
||||
* @since 2026/8/8
|
||||
*/
|
||||
@Slf4j
|
||||
@Service
|
||||
@RequiredArgsConstructor
|
||||
public class DeviceOnlineServiceImpl implements DeviceOnlineService {
|
||||
|
||||
private final RedisTemplate<String, Object> redisTemplate;
|
||||
private final MqttProperties mqttProperties;
|
||||
|
||||
@Override
|
||||
public void reportHeartbeat(String sn, String ip) {
|
||||
reportHeartbeat(sn, ip, null);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void reportHeartbeat(String sn, String ip, Integer screenState) {
|
||||
if (StrUtil.isBlank(sn)) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
String key = RedisConstants.Device.ONLINE.replace("{}", sn);
|
||||
DeviceOnlineVO vo = getOnlineInfo(sn);
|
||||
if (vo == null) {
|
||||
vo = new DeviceOnlineVO();
|
||||
vo.setSerialno(sn);
|
||||
}
|
||||
vo.setOnline(1);
|
||||
vo.setLastHeartbeatTime(LocalDateTime.now());
|
||||
if (StrUtil.isNotBlank(ip)) {
|
||||
vo.setIp(ip);
|
||||
}
|
||||
// 仅在本次上报携带屏幕状态时更新，避免心跳覆盖屏幕状态
|
||||
if (screenState != null) {
|
||||
vo.setScreenState(screenState);
|
||||
}
|
||||
redisTemplate.opsForValue().set(key, vo, mqttProperties.getOnlineTimeout(), TimeUnit.SECONDS);
|
||||
log.debug("设备上报心跳, sn={}, ip={}, screenState={}, timeout={}s", sn, ip, screenState, mqttProperties.getOnlineTimeout());
|
||||
} catch (Exception e) {
|
||||
log.error("设备上报心跳写入 Redis 失败, sn={}", sn, e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void markOffline(String sn) {
|
||||
if (StrUtil.isBlank(sn)) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
String key = RedisConstants.Device.ONLINE.replace("{}", sn);
|
||||
redisTemplate.delete(key);
|
||||
log.info("设备标记离线, sn={}", sn);
|
||||
} catch (Exception e) {
|
||||
log.error("设备标记离线失败, sn={}", sn, e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isOnline(String sn) {
|
||||
if (StrUtil.isBlank(sn)) {
|
||||
return false;
|
||||
}
|
||||
String key = RedisConstants.Device.ONLINE.replace("{}", sn);
|
||||
Boolean exists = redisTemplate.hasKey(key);
|
||||
return Boolean.TRUE.equals(exists);
|
||||
}
|
||||
|
||||
@Override
|
||||
public DeviceOnlineVO getOnlineInfo(String sn) {
|
||||
if (StrUtil.isBlank(sn)) {
|
||||
return null;
|
||||
}
|
||||
String key = RedisConstants.Device.ONLINE.replace("{}", sn);
|
||||
Object value = redisTemplate.opsForValue().get(key);
|
||||
if (value instanceof DeviceOnlineVO vo) {
|
||||
return vo;
|
||||
}
|
||||
// 离线
|
||||
DeviceOnlineVO offline = new DeviceOnlineVO();
|
||||
offline.setSerialno(sn);
|
||||
offline.setOnline(0);
|
||||
return offline;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Map<String, DeviceOnlineVO> getOnlineStatusBySns(List<String> sns) {
|
||||
Map<String, DeviceOnlineVO> result = new LinkedHashMap<>();
|
||||
if (sns == null || sns.isEmpty()) {
|
||||
return result;
|
||||
}
|
||||
List<String> keys = new ArrayList<>(sns.size());
|
||||
for (String sn : sns) {
|
||||
if (StrUtil.isBlank(sn)) {
|
||||
continue;
|
||||
}
|
||||
String key = RedisConstants.Device.ONLINE.replace("{}", sn);
|
||||
keys.add(key);
|
||||
// 预置离线状态，供缺失 key 时填充
|
||||
DeviceOnlineVO offline = new DeviceOnlineVO();
|
||||
offline.setSerialno(sn);
|
||||
offline.setOnline(0);
|
||||
result.put(sn, offline);
|
||||
}
|
||||
try {
|
||||
List<Object> values = redisTemplate.opsForValue().multiGet(keys);
|
||||
if (values != null) {
|
||||
for (int i = 0; i < values.size(); i++) {
|
||||
Object value = values.get(i);
|
||||
String sn = sns.get(i);
|
||||
if (value instanceof DeviceOnlineVO vo) {
|
||||
result.put(sn, vo);
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch (Exception e) {
|
||||
log.error("批量查询设备在线状态失败", e);
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -24,8 +24,10 @@ import com.youlai.boot.device.model.vo.DeviceNetworkInfoVO;
|
||||
import com.youlai.boot.device.model.vo.DeviceOtherInfoVO;
|
||||
import com.youlai.boot.device.model.vo.DeviceSecurityInfoVO;
|
||||
import com.youlai.boot.device.model.vo.DeviceSystemInfoVO;
|
||||
import com.youlai.boot.device.model.vo.DeviceOnlineVO;
|
||||
import com.youlai.boot.device.model.vo.DevicePageVO;
|
||||
import com.youlai.boot.device.service.DeveloperService;
|
||||
import com.youlai.boot.device.service.DeviceOnlineService;
|
||||
import com.youlai.boot.device.service.DeviceService;
|
||||
import com.youlai.boot.device.service.HardwareInfoService;
|
||||
import com.youlai.boot.device.service.NetworkInfoService;
|
||||
@@ -37,8 +39,10 @@ import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.beans.BeanUtils;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.HashMap;
|
||||
import java.util.HashSet;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
|
||||
@@ -66,6 +70,8 @@ public class DeviceServiceImpl extends ServiceImpl<DeviceMapper, SnDeviceInfo> i
|
||||
|
||||
private final SystemInfoConverter systemInfoConverter;
|
||||
|
||||
private final DeviceOnlineService deviceOnlineService;
|
||||
|
||||
@Override
|
||||
public IPage<DevicePageVO> getSnPage(DeviceQuery queryParams) {
|
||||
// 参数构建
|
||||
@@ -77,7 +83,37 @@ public class DeviceServiceImpl extends ServiceImpl<DeviceMapper, SnDeviceInfo> i
|
||||
queryParams.setIsRoot(isRoot);
|
||||
|
||||
// 查询数据
|
||||
return this.baseMapper.getSnPage(page, queryParams);
|
||||
IPage<DevicePageVO> result = this.baseMapper.getSnPage(page, queryParams);
|
||||
|
||||
// 填充设备在线状态（Redis）
|
||||
fillOnlineStatus(result.getRecords());
|
||||
return result;
|
||||
}
|
||||
|
||||
/**
|
||||
* 批量填充设备在线状态
|
||||
*/
|
||||
private void fillOnlineStatus(List<DevicePageVO> records) {
|
||||
if (records == null || records.isEmpty()) {
|
||||
return;
|
||||
}
|
||||
List<String> sns = new ArrayList<>(records.size());
|
||||
for (DevicePageVO record : records) {
|
||||
if (record.getSerialno() != null) {
|
||||
sns.add(record.getSerialno());
|
||||
}
|
||||
}
|
||||
Map<String, DeviceOnlineVO> onlineMap =
|
||||
deviceOnlineService.getOnlineStatusBySns(sns);
|
||||
for (DevicePageVO record : records) {
|
||||
DeviceOnlineVO online = onlineMap.get(record.getSerialno());
|
||||
if (online != null) {
|
||||
record.setOnline(online.getOnline());
|
||||
record.setLastHeartbeatTime(online.getLastHeartbeatTime());
|
||||
} else {
|
||||
record.setOnline(0);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
@@ -6,6 +6,7 @@ import com.youlai.boot.common.result.Result;
|
||||
import com.youlai.boot.common.result.ResultCode;
|
||||
import com.youlai.boot.common.exception.TokenInvalidException;
|
||||
import com.youlai.boot.common.exception.RateLimitException;
|
||||
import io.lettuce.core.RedisException;
|
||||
import jakarta.servlet.ServletException;
|
||||
import jakarta.servlet.http.HttpServletRequest;
|
||||
import jakarta.validation.ConstraintViolation;
|
||||
@@ -16,6 +17,8 @@ import org.springframework.context.support.DefaultMessageSourceResolvable;
|
||||
import org.springframework.http.HttpStatus;
|
||||
import org.springframework.http.MediaType;
|
||||
import org.springframework.http.converter.HttpMessageNotReadableException;
|
||||
import org.springframework.data.redis.RedisConnectionFailureException;
|
||||
import org.springframework.data.redis.RedisSystemException;
|
||||
import org.springframework.jdbc.BadSqlGrammarException;
|
||||
import org.springframework.security.access.AccessDeniedException;
|
||||
import org.springframework.security.core.AuthenticationException;
|
||||
@@ -191,6 +194,36 @@ public class GlobalExceptionHandler {
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Redis 连接失败异常
|
||||
*/
|
||||
@ExceptionHandler(RedisConnectionFailureException.class)
|
||||
@ResponseStatus(HttpStatus.SERVICE_UNAVAILABLE)
|
||||
public <T> Result<T> handleRedisConnectionFailureException(RedisConnectionFailureException e) {
|
||||
log.error("Redis 连接失败", e);
|
||||
return Result.failed(ResultCode.REDIS_CONNECTION_FAILURE, "Redis 服务暂时不可用，请稍后重试");
|
||||
}
|
||||
|
||||
/**
|
||||
* Redis 系统异常
|
||||
*/
|
||||
@ExceptionHandler(RedisSystemException.class)
|
||||
@ResponseStatus(HttpStatus.INTERNAL_SERVER_ERROR)
|
||||
public <T> Result<T> handleRedisSystemException(RedisSystemException e) {
|
||||
log.error("Redis 系统异常", e);
|
||||
return Result.failed(ResultCode.REDIS_OPERATION_ERROR, "Redis 操作异常，请稍后重试");
|
||||
}
|
||||
|
||||
/**
|
||||
* Lettuce Redis 异常
|
||||
*/
|
||||
@ExceptionHandler(RedisException.class)
|
||||
@ResponseStatus(HttpStatus.INTERNAL_SERVER_ERROR)
|
||||
public <T> Result<T> handleRedisException(RedisException e) {
|
||||
log.error("Redis 异常", e);
|
||||
return Result.failed(ResultCode.REDIS_SERVICE_ERROR, "Redis 服务异常，请稍后重试");
|
||||
}
|
||||
|
||||
@ExceptionHandler(Exception.class)
|
||||
@ResponseStatus(HttpStatus.INTERNAL_SERVER_ERROR)
|
||||
public <T> Result<T> handleException(Exception e, HttpServletRequest request) throws Exception {
|
||||
|
||||
@@ -0,0 +1,41 @@
|
||||
package com.youlai.boot.message.mqtt;
|
||||
|
||||
import com.youlai.boot.common.result.Result;
|
||||
import io.swagger.v3.oas.annotations.Operation;
|
||||
import io.swagger.v3.oas.annotations.Parameter;
|
||||
import io.swagger.v3.oas.annotations.tags.Tag;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||
import org.springframework.web.bind.annotation.PostMapping;
|
||||
import org.springframework.web.bind.annotation.RequestMapping;
|
||||
import org.springframework.web.bind.annotation.RequestParam;
|
||||
import org.springframework.web.bind.annotation.RestController;
|
||||
|
||||
/**
|
||||
* MQTT 控制层
|
||||
* <p>
|
||||
* 提供发布消息的接口，方便联调测试。当 {@code mqtt.enabled=false} 时该 Bean 不创建。
|
||||
*
|
||||
* @author TongTongStudio
|
||||
* @since 2026/8/8
|
||||
*/
|
||||
@Tag(name = "MQTT 消息")
|
||||
@RestController
|
||||
@RequestMapping("/api/v1/mqtt")
|
||||
@RequiredArgsConstructor
|
||||
@ConditionalOnProperty(prefix = "mqtt", name = "enabled", havingValue = "true")
|
||||
public class MqttController {
|
||||
|
||||
private final MqttProducer mqttProducer;
|
||||
|
||||
@Operation(summary = "发布 MQTT 消息")
|
||||
@PostMapping("/publish")
|
||||
public Result<Boolean> publish(
|
||||
@Parameter(description = "主题") @RequestParam String topic,
|
||||
@Parameter(description = "消息内容") @RequestParam String payload
|
||||
) {
|
||||
mqttProducer.publish(topic, payload);
|
||||
return Result.success(true);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,103 @@
|
||||
package com.youlai.boot.message.mqtt;
|
||||
|
||||
import cn.hutool.core.util.StrUtil;
|
||||
import cn.hutool.json.JSONObject;
|
||||
import cn.hutool.json.JSONUtil;
|
||||
import com.youlai.boot.config.property.MqttProperties;
|
||||
import com.youlai.boot.device.service.DeviceOnlineService;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||
import org.springframework.integration.annotation.ServiceActivator;
|
||||
import org.springframework.integration.mqtt.support.MqttHeaders;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageHandler;
|
||||
import org.springframework.messaging.MessagingException;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.nio.charset.StandardCharsets;
|
||||
|
||||
/**
|
||||
* MQTT 订阅消息处理器
|
||||
* <p>
|
||||
* 消费 {@code mqttInboundChannel} 通道中的消息，根据主题分发处理。
|
||||
*
|
||||
* @author TongTongStudio
|
||||
* @since 2026/8/8
|
||||
*/
|
||||
@Slf4j
|
||||
@Component
|
||||
@ConditionalOnProperty(prefix = "mqtt", name = "enabled", havingValue = "true")
|
||||
public class MqttMessageHandler implements MessageHandler {
|
||||
|
||||
private final DeviceOnlineService deviceOnlineService;
|
||||
private final MqttProperties mqttProperties;
|
||||
|
||||
public MqttMessageHandler(DeviceOnlineService deviceOnlineService, MqttProperties mqttProperties) {
|
||||
this.deviceOnlineService = deviceOnlineService;
|
||||
this.mqttProperties = mqttProperties;
|
||||
}
|
||||
|
||||
/**
|
||||
* 处理从 MQTT Broker 订阅到的消息
|
||||
*
|
||||
* @param message MQTT 入站消息，可通过 {@link MqttHeaders#RECEIVED_TOPIC} 获取主题
|
||||
*/
|
||||
@ServiceActivator(inputChannel = "mqttInboundChannel")
|
||||
@Override
|
||||
public void handleMessage(Message<?> message) throws MessagingException {
|
||||
String topic = message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC, String.class);
|
||||
String payload = message.getPayload() instanceof byte[]
|
||||
? new String((byte[]) message.getPayload(), StandardCharsets.UTF_8)
|
||||
: String.valueOf(message.getPayload());
|
||||
|
||||
log.info("收到 MQTT 消息, topic={}, payload={}", topic, payload);
|
||||
|
||||
// 根据业务需要，按主题分发处理
|
||||
String statusTopic = mqttProperties.getStatusTopic();
|
||||
if (statusTopic != null && statusTopic.equals(topic)) {
|
||||
handleDeviceStatus(payload);
|
||||
} else {
|
||||
log.debug("未匹配到主题 {} 的处理逻辑", topic);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 处理设备状态上报（心跳/屏幕状态），并同步更新 Redis 中的在线状态
|
||||
* <p>
|
||||
* 消息体为 JSON，格式：{"sn":"xxx", "status":"online|offline", "ip":"1.2.3.4", "screenOn":true|false}
|
||||
* 其中 screenOn 为可选字段，表示设备亮屏(亮屏=屏幕点亮)状态。
|
||||
*/
|
||||
private void handleDeviceStatus(String payload) {
|
||||
log.info("处理设备状态上报: {}", payload);
|
||||
if (StrUtil.isBlank(payload)) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
JSONObject json = JSONUtil.parseObj(payload);
|
||||
String sn = json.getStr("sn");
|
||||
if (StrUtil.isBlank(sn)) {
|
||||
log.warn("设备状态上报缺少 sn 字段, payload={}", payload);
|
||||
return;
|
||||
}
|
||||
String status = json.getStr("status");
|
||||
String ip = json.getStr("ip");
|
||||
|
||||
// 解析屏幕状态（可选）：true=亮屏, false=熄屏, 缺省=null（未知）
|
||||
Integer screenState = null;
|
||||
if (json.containsKey("screenOn")) {
|
||||
screenState = json.getBool("screenOn") ? 1 : 0;
|
||||
}
|
||||
|
||||
if ("offline".equalsIgnoreCase(status)) {
|
||||
// 设备主动上报离线（如关机/退出），立即标记离线
|
||||
deviceOnlineService.markOffline(sn);
|
||||
} else {
|
||||
// 默认视为心跳/在线，刷新 Redis TTL，并同步屏幕状态
|
||||
deviceOnlineService.reportHeartbeat(sn, ip, screenState);
|
||||
}
|
||||
} catch (Exception e) {
|
||||
log.error("解析设备状态上报失败, payload={}", payload, e);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
62
src/main/java/com/youlai/boot/message/mqtt/MqttProducer.java
Normal file
62
src/main/java/com/youlai/boot/message/mqtt/MqttProducer.java
Normal file
@@ -0,0 +1,62 @@
|
||||
package com.youlai.boot.message.mqtt;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||
import org.springframework.integration.core.MessagingTemplate;
|
||||
import org.springframework.integration.mqtt.support.MqttHeaders;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
import org.springframework.messaging.support.MessageBuilder;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* MQTT 消息发布服务
|
||||
* <p>
|
||||
* 提供向指定主题发布消息的能力，消息经过 {@code mqttOutboundChannel} 通道发送到 MQTT Broker。
|
||||
* 当 {@code mqtt.enabled=false} 时该 Bean 不创建。
|
||||
*
|
||||
* @author TongTongStudio
|
||||
* @since 2026/8/8
|
||||
*/
|
||||
@Slf4j
|
||||
@Component
|
||||
@ConditionalOnProperty(prefix = "mqtt", name = "enabled", havingValue = "true")
|
||||
public class MqttProducer {
|
||||
|
||||
private final MessagingTemplate messagingTemplate;
|
||||
|
||||
@Autowired
|
||||
public MqttProducer(MessageChannel mqttOutboundChannel) {
|
||||
this.messagingTemplate = new MessagingTemplate(mqttOutboundChannel);
|
||||
}
|
||||
|
||||
/**
|
||||
* 发布消息
|
||||
*
|
||||
* @param topic 主题
|
||||
* @param payload 消息内容(字符串)
|
||||
*/
|
||||
public void publish(String topic, String payload) {
|
||||
publish(topic, payload, null);
|
||||
}
|
||||
|
||||
/**
|
||||
* 发布消息
|
||||
*
|
||||
* @param topic 主题
|
||||
* @param payload 消息内容(字符串)
|
||||
* @param qos QoS 级别，为 null 时使用默认 QoS
|
||||
*/
|
||||
public void publish(String topic, String payload, Integer qos) {
|
||||
MessageBuilder<String> builder = MessageBuilder.withPayload(payload)
|
||||
.setHeader(MqttHeaders.TOPIC, topic);
|
||||
if (qos != null) {
|
||||
builder.setHeader(MqttHeaders.QOS, qos);
|
||||
}
|
||||
Message<String> message = builder.build();
|
||||
messagingTemplate.send(message);
|
||||
log.info("发布 MQTT 消息, topic={}, payload={}", topic, payload);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -20,6 +20,8 @@ spring:
|
||||
port: 26379
|
||||
password: fanhuitong
|
||||
timeout: 10s
|
||||
# 连接超时时间
|
||||
connect-timeout: 5s
|
||||
lettuce:
|
||||
pool:
|
||||
# 连接池最大连接数 默认8 ，负数表示没有限制
|
||||
@@ -30,6 +32,10 @@ spring:
|
||||
max-idle: 8
|
||||
# 连接池中的最小空闲连接 默认0
|
||||
min-idle: 0
|
||||
cluster:
|
||||
refresh:
|
||||
adaptive: true
|
||||
period: 30s
|
||||
cache:
|
||||
enabled: false
|
||||
# 缓存类型 redis、none(不使用缓存)
|
||||
@@ -259,3 +265,26 @@ wx:
|
||||
miniapp:
|
||||
appid: Your_AppId
|
||||
secret: Your_AppSecret
|
||||
|
||||
# MQTT 配置
|
||||
mqtt:
|
||||
enabled: true
|
||||
# Broker 地址
|
||||
url: mqtt://175.178.213.60:1883
|
||||
# 用户名/密码(按需配置)
|
||||
username:
|
||||
password:
|
||||
# 客户端 ID
|
||||
client-id: youlai-boot-server
|
||||
# 默认 QoS
|
||||
qos: 1
|
||||
connection-timeout: 10
|
||||
keep-alive-interval: 60
|
||||
automatic-reconnect: true
|
||||
# 订阅的主题列表
|
||||
topics:
|
||||
- ttstd/device/status
|
||||
# 设备状态/心跳上报主题
|
||||
status-topic: ttstd/device/status
|
||||
# 设备在线判定超时时间(秒)，设备每上报一次心跳刷新该 TTL，超时未上报即判定离线
|
||||
online-timeout: 90
|
||||
|
||||
@@ -15,6 +15,8 @@ spring:
|
||||
port: 26379
|
||||
password: fanhuitong
|
||||
timeout: 10s
|
||||
# 连接超时时间
|
||||
connect-timeout: 5s
|
||||
lettuce:
|
||||
pool:
|
||||
# 连接池最大连接数 默认8 ，负数表示没有限制
|
||||
@@ -25,6 +27,10 @@ spring:
|
||||
max-idle: 8
|
||||
# 连接池中的最小空闲连接 默认0
|
||||
min-idle: 0
|
||||
cluster:
|
||||
refresh:
|
||||
adaptive: true
|
||||
period: 30s
|
||||
cache:
|
||||
enabled: false
|
||||
# 缓存类型 redis、none(不使用缓存)
|
||||
@@ -240,3 +246,26 @@ wx:
|
||||
miniapp:
|
||||
app-id: Your_AppId
|
||||
app-secret: Your_AppSecret
|
||||
|
||||
# MQTT 配置
|
||||
mqtt:
|
||||
enabled: true
|
||||
# Broker 地址
|
||||
url: mqtt://175.178.213.60:1883
|
||||
# 用户名/密码(按需配置)
|
||||
username:
|
||||
password:
|
||||
# 客户端 ID
|
||||
client-id: youlai-boot-server
|
||||
# 默认 QoS
|
||||
qos: 1
|
||||
connection-timeout: 10
|
||||
keep-alive-interval: 60
|
||||
automatic-reconnect: true
|
||||
# 订阅的主题列表
|
||||
topics:
|
||||
- ttstd/device/status
|
||||
# 设备状态/心跳上报主题
|
||||
status-topic: ttstd/device/status
|
||||
# 设备在线判定超时时间(秒)，设备每上报一次心跳刷新该 TTL，超时未上报即判定离线
|
||||
online-timeout: 90
|
||||
|
||||
@@ -11,3 +11,14 @@ spring:
|
||||
# 单文件大小上限，取自 file-storage.upload.max-file-size（单一来源）
|
||||
max-file-size: ${file-storage.upload.max-file-size:50MB}
|
||||
max-request-size: ${file-storage.upload.max-file-size:50MB}
|
||||
|
||||
# 延迟初始化：应用启动时不立即初始化 Bean，首次使用时才初始化
|
||||
# 配合 Redis Lazy 初始化，即使 Redis 不可用也能正常启动
|
||||
main:
|
||||
lazy-initialization: false # 保持 false（全局延迟初始化会导致很多问题）
|
||||
autoconfigure:
|
||||
# 不排除 Redis 自动配置，而是通过 Lazy Bean 来控制
|
||||
# 如果 Redis 完全不可用且希望跳过，可以取消下面注释：
|
||||
# exclude:
|
||||
# - org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration
|
||||
# - org.redisson.spring.starter.RedissonAutoConfiguration
|
||||
|
||||
Reference in New Issue
Block a user