commandPublisherProvider;
private final DeviceIdentityResolver deviceIdentityResolver;
+ private final MqttDeviceStatusService deviceStatusService;
@Override
public Pattern topicPattern() {
@@ -60,8 +61,9 @@ DeviceRegisterHandler implements MqttTopicHandler {
}
AppDeviceBo device = buildRegisterDevice(deviceNo, normalizedDeviceMac, dto);
appDeviceService.registerByMqtt(device);
+ deviceStatusService.markOnline(deviceNo);
sendDeviceNoToDevice(deviceNo, normalizedDeviceMac);
- log.info("[MQTT] 设备注册 时间={} MAC={} 设备编号={} 消息体={}",
+ log.info("[MQTT] 设备注册并上线 时间={} MAC={} 设备编号={} 消息体={}",
HandlerLogTime.now(), normalizedDeviceMac, deviceNo, payload);
} catch (Exception e) {
log.error("[MQTT] 设备注册失败 时间={} 设备标识={} 消息体={}", HandlerLogTime.now(), deviceIdentity, payload, e);
diff --git a/water-modules/water-app/src/main/java/org/dromara/app/handler/DeviceStatusHandler.java b/water-modules/water-app/src/main/java/org/dromara/app/handler/DeviceStatusHandler.java
index 67cf924..5975e13 100644
--- a/water-modules/water-app/src/main/java/org/dromara/app/handler/DeviceStatusHandler.java
+++ b/water-modules/water-app/src/main/java/org/dromara/app/handler/DeviceStatusHandler.java
@@ -13,10 +13,10 @@ import java.util.Map;
import java.util.regex.Pattern;
/**
- * 设备在线/离线状态处理器，匹配 /{deviceIdentity}/publish/status。
+ * 设备遗嘱离线状态处理器，匹配 /{deviceIdentity}/publish/status。
*
* deviceIdentity 支持设备编号；设备遗嘱消息允许使用 MAC 地址，处理前统一解析为设备编号。
- * 设备通过 status=online 标记上线，通过 LWT status=offline 标记异常离线。
+ * 在线状态由电量上报刷新；本处理器仅通过 LWT status=offline 标记异常离线。
*/
@Slf4j
@Component
@@ -42,17 +42,18 @@ public class DeviceStatusHandler implements MqttTopicHandler {
@Override
public void handle(String deviceIdentity, String payload, boolean retained) {
- String deviceNo = deviceIdentityResolver.resolveDeviceNo(deviceIdentity);
+ Map body = parsePayload(payload);
+ String deviceNo = resolveDeviceNo(deviceIdentity, body);
if (deviceNo == null) {
log.warn("[MQTT] 设备状态上报未找到设备 时间={} 设备标识={} 消息体={}",
HandlerLogTime.now(), deviceIdentity, payload);
return;
}
- String status = parseStatus(payload);
+ String status = parseStatus(payload, body);
if ("online".equals(status) || "1".equals(status)) {
- deviceStatusService.markOnline(deviceNo);
- log.info("[MQTT] 设备状态在线 时间={} 设备编号={} 消息体={}", HandlerLogTime.now(), deviceNo, payload);
+ log.debug("[MQTT] 忽略设备主动在线状态，在线状态由电量心跳维护 时间={} 设备编号={} 消息体={}",
+ HandlerLogTime.now(), deviceNo, payload);
return;
}
if ("offline".equals(status) || "0".equals(status)) {
@@ -69,22 +70,52 @@ public class DeviceStatusHandler implements MqttTopicHandler {
HandlerLogTime.now(), deviceNo, payload);
}
- private String parseStatus(String payload) {
+ private String resolveDeviceNo(String deviceIdentity, Map body) {
+ String deviceNo = deviceIdentityResolver.resolveDeviceNo(deviceIdentity);
+ if (deviceNo != null || body == null) {
+ return deviceNo;
+ }
+ Object deviceMac = body.get("deviceMac");
+ if (deviceMac == null || StringUtils.isBlank(String.valueOf(deviceMac))) {
+ return null;
+ }
+ return deviceIdentityResolver.resolveDeviceNo(String.valueOf(deviceMac));
+ }
+
+ private Map parsePayload(String payload) {
if (StringUtils.isBlank(payload)) {
return null;
}
String trimmed = payload.trim();
if (!trimmed.startsWith("{")) {
- return trimmed.toLowerCase(Locale.ROOT);
+ return null;
}
try {
- Map body = objectMapper.readValue(trimmed, new TypeReference>() {
+ return objectMapper.readValue(trimmed, new TypeReference>() {
});
- Object status = body.get("status");
- return status == null ? null : String.valueOf(status).trim().toLowerCase(Locale.ROOT);
} catch (Exception e) {
log.warn("[MQTT] 设备状态上报 JSON 格式错误 时间={} 消息体={}", HandlerLogTime.now(), payload, e);
return null;
}
}
+
+ private String parseStatus(String payload, Map body) {
+ if (body == null) {
+ if (StringUtils.isBlank(payload)) {
+ return null;
+ }
+ String trimmed = payload.trim();
+ return trimmed.startsWith("{") ? null : trimmed.toLowerCase(Locale.ROOT);
+ }
+
+ Object status = body.get("status");
+ if (status != null) {
+ return String.valueOf(status).trim().toLowerCase(Locale.ROOT);
+ }
+ Object offline = body.get("offline");
+ if (offline != null && "true".equalsIgnoreCase(String.valueOf(offline).trim())) {
+ return "offline";
+ }
+ return null;
+ }
}
diff --git a/water-modules/water-app/src/main/java/org/dromara/app/handler/MqttDeviceStatusService.java b/water-modules/water-app/src/main/java/org/dromara/app/handler/MqttDeviceStatusService.java
index 4c5b6e0..4b4eb4d 100644
--- a/water-modules/water-app/src/main/java/org/dromara/app/handler/MqttDeviceStatusService.java
+++ b/water-modules/water-app/src/main/java/org/dromara/app/handler/MqttDeviceStatusService.java
@@ -32,7 +32,7 @@ public class MqttDeviceStatusService {
@Value("${mqtt.command-ack.device-status-cache-prefix}")
private String deviceStatusCachePrefix;
- @Value("${mqtt.command-ack.device-status-cache-ttl-seconds:900}")
+ @Value("${mqtt.command-ack.device-status-cache-ttl-seconds:600}")
private long deviceStatusCacheTtlSeconds;
public void markOnline(String deviceNo) {
diff --git a/water-modules/water-app/src/main/java/org/dromara/app/service/IAppDeviceService.java b/water-modules/water-app/src/main/java/org/dromara/app/service/IAppDeviceService.java
index 5a87735..606539f 100644
--- a/water-modules/water-app/src/main/java/org/dromara/app/service/IAppDeviceService.java
+++ b/water-modules/water-app/src/main/java/org/dromara/app/service/IAppDeviceService.java
@@ -29,6 +29,14 @@ public interface IAppDeviceService {
*/
AppDeviceVo queryById(String deviceNo);
+ /**
+ * 按设备编号批量查询设备。
+ *
+ * @param deviceNos 设备编号集合
+ * @return 设备列表
+ */
+ List queryByDeviceNos(Collection deviceNos);
+
/**
* 分页查询设备信息
列表
diff --git a/water-modules/water-app/src/main/java/org/dromara/app/service/IAppSchedulingDeviceService.java b/water-modules/water-app/src/main/java/org/dromara/app/service/IAppSchedulingDeviceService.java
index 279537c..e7786e2 100644
--- a/water-modules/water-app/src/main/java/org/dromara/app/service/IAppSchedulingDeviceService.java
+++ b/water-modules/water-app/src/main/java/org/dromara/app/service/IAppSchedulingDeviceService.java
@@ -8,6 +8,7 @@ import org.dromara.common.mybatis.core.page.TableDataInfo;
import java.util.Collection;
import java.util.List;
+import java.util.Set;
public interface IAppSchedulingDeviceService {
@@ -25,6 +26,14 @@ public interface IAppSchedulingDeviceService {
List findByDeviceNo(String deviceNo);
+ /**
+ * 批量查询已绑定排程的设备编号。
+ *
+ * @param deviceNos 待检查的设备编号
+ * @return 已绑定排程的设备编号
+ */
+ Set findBoundDeviceNos(Collection deviceNos);
+
Boolean deleteWithValidByScheduleIdAndDeviceNo(@NotEmpty(message = "主键不能为空") Long scheduleId, String deviceNo);
AppSchedulingDeviceVo queryByScheduleIdAndDeviceNo(Long scheduleId, String deviceNo);
diff --git a/water-modules/water-app/src/main/java/org/dromara/app/service/impl/AppDeviceServiceImpl.java b/water-modules/water-app/src/main/java/org/dromara/app/service/impl/AppDeviceServiceImpl.java
index 49aadce..0ac4694 100644
--- a/water-modules/water-app/src/main/java/org/dromara/app/service/impl/AppDeviceServiceImpl.java
+++ b/water-modules/water-app/src/main/java/org/dromara/app/service/impl/AppDeviceServiceImpl.java
@@ -73,6 +73,14 @@ public class AppDeviceServiceImpl implements IAppDeviceService {
return baseMapper.selectVoById(deviceNo);
}
+ @Override
+ public List queryByDeviceNos(Collection deviceNos) {
+ if (deviceNos == null || deviceNos.isEmpty()) {
+ return List.of();
+ }
+ return baseMapper.selectVoByIds(deviceNos);
+ }
+
@Override
public TableDataInfo queryPageList(AppDeviceBo bo, PageQuery pageQuery) {
Page page = pageQuery.build();
@@ -307,7 +315,7 @@ public class AppDeviceServiceImpl implements IAppDeviceService {
return params;
}
params.put("bindDevice", exists);
- if (exists.getUserId() == null && exists.getWorkStatus().equals("2") && exists.getStatus().equals("0") ) {
+ if (exists.getUserId() == null && "2".equals(exists.getWorkStatus()) && "0".equals(exists.getStatus())) {
params.put("bindDeviceStatus", 305);
params.put("bindDeviceStatusName", "设备未配网,请先去配置网络");
return params;
diff --git a/water-modules/water-app/src/main/java/org/dromara/app/service/impl/AppScheduleServiceImpl.java b/water-modules/water-app/src/main/java/org/dromara/app/service/impl/AppScheduleServiceImpl.java
index d6ab07e..487e266 100644
--- a/water-modules/water-app/src/main/java/org/dromara/app/service/impl/AppScheduleServiceImpl.java
+++ b/water-modules/water-app/src/main/java/org/dromara/app/service/impl/AppScheduleServiceImpl.java
@@ -165,10 +165,10 @@ public class AppScheduleServiceImpl implements IAppScheduleService {
return conflicts;
}
- private List saveDetails(Long scheduleId, List details) {
- List list = new ArrayList<>();
+ private List> saveDetails(Long scheduleId, List details) {
+ List> detailList = new ArrayList<>();
if (details == null) {
- return list;
+ return detailList;
}
for (AppScheduleBo.AppScheduleDetail item : details) {
AppScheduleDetail detail = new AppScheduleDetail();
@@ -181,13 +181,13 @@ public class AppScheduleServiceImpl implements IAppScheduleService {
detail.setTriggerType(item.getTriggerType());
detail.setStatus(item.getStatus());
scheduleDetailMapper.insert(detail);
- list.add(toDetailMap(detail, item.getTimeData()));
+ detailList.add(toDetailMap(detail, item.getTimeData()));
}
- return list;
+ return detailList;
}
- private List updateDetails(List details) {
- List detailList = new ArrayList<>();
+ private List> updateDetails(List details) {
+ List> detailList = new ArrayList<>();
if (details == null) {
return detailList;
}
diff --git a/water-modules/water-app/src/main/java/org/dromara/app/service/impl/AppSchedulingDeviceServiceImpl.java b/water-modules/water-app/src/main/java/org/dromara/app/service/impl/AppSchedulingDeviceServiceImpl.java
index 71c3361..0cc5727 100644
--- a/water-modules/water-app/src/main/java/org/dromara/app/service/impl/AppSchedulingDeviceServiceImpl.java
+++ b/water-modules/water-app/src/main/java/org/dromara/app/service/impl/AppSchedulingDeviceServiceImpl.java
@@ -21,6 +21,8 @@ import org.springframework.stereotype.Service;
import java.util.Collection;
import java.util.List;
import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
@Slf4j
@RequiredArgsConstructor
@@ -106,4 +108,19 @@ public class AppSchedulingDeviceServiceImpl implements IAppSchedulingDeviceServi
public List findByDeviceNo(String deviceNo) {
return baseMapper.selectVoList(new QueryWrapper().eq("device_no", deviceNo));
}
+
+ @Override
+ public Set findBoundDeviceNos(Collection deviceNos) {
+ if (deviceNos == null || deviceNos.isEmpty()) {
+ return Set.of();
+ }
+ return baseMapper.selectList(
+ new QueryWrapper()
+ .select("device_no")
+ .in("device_no", deviceNos))
+ .stream()
+ .map(AppSchedulingDevice::getDeviceNo)
+ .filter(StringUtils::isNotBlank)
+ .collect(Collectors.toSet());
+ }
}
diff --git a/water-modules/water-app/src/main/java/org/dromara/app/task/DeviceOfflineCheckTask.java b/water-modules/water-app/src/main/java/org/dromara/app/task/DeviceOfflineCheckTask.java
index 607a488..142d41b 100644
--- a/water-modules/water-app/src/main/java/org/dromara/app/task/DeviceOfflineCheckTask.java
+++ b/water-modules/water-app/src/main/java/org/dromara/app/task/DeviceOfflineCheckTask.java
@@ -18,8 +18,8 @@ import java.util.List;
/**
* 设备离线检测定时任务
*
- * 逻辑：默认通过设备 LWT 离线消息更新状态。
- * 兼容旧设备时，可打开 ttl-compat-enabled，继续使用 Redis 在线 Key 过期兜底离线。
+ * 逻辑：电量上报刷新 Redis 在线 Key，Key 超时后判定设备离线。
+ * 设备 LWT 离线消息仍可立即更新离线状态，本任务用于处理未收到遗嘱的异常断线。
* 本任务定期扫描数据库中非离线状态的设备，如果对应 Redis Key 已不存在，
* 则将数据库 status 更新为 0（离线）。
*
@@ -37,19 +37,19 @@ public class DeviceOfflineCheckTask {
@Value("${mqtt.command-ack.device-status-cache-prefix}")
private String deviceStatusCachePrefix;
- @Value("${mqtt.command-ack.device-status-cache-ttl-seconds:900}")
+ @Value("${mqtt.command-ack.device-status-cache-ttl-seconds:600}")
private long deviceStatusCacheTtlSeconds;
@Value("${mqtt.offline-check.enabled:true}")
private boolean enabled;
- @Value("${mqtt.offline-check.ttl-compat-enabled:false}")
+ @Value("${mqtt.offline-check.ttl-compat-enabled:true}")
private boolean ttlCompatEnabled;
/**
- * 每 90 秒检查一次。默认不开启 TTL 兼容离线检测，避免低功耗长连接设备被误判离线。
+ * 每 30 秒检查一次电量心跳是否超时。
* 可通过 mqtt.offline-check.interval-ms 覆盖（单位 ms）
*/
- @Scheduled(fixedDelayString = "${mqtt.offline-check.interval-ms:90000}")
+ @Scheduled(fixedDelayString = "${mqtt.offline-check.interval-ms:30000}")
public void checkOfflineDevices() {
if (!enabled || !ttlCompatEnabled) {
return;
@@ -96,7 +96,7 @@ public class DeviceOfflineCheckTask {
List updatedDeviceNos = new ArrayList<>();
for (String deviceNo : offlineDeviceNos) {
- if (deviceStatusService.markOfflineIfCacheMissing(deviceNo, "设备掉线")) {
+ if (deviceStatusService.markOfflineIfCacheMissing(deviceNo, "设备电量心跳超时")) {
updatedDeviceNos.add(deviceNo);
}
}
diff --git a/water-modules/water-app/src/test/java/org/dromara/app/controller/AppControllerConcurrencyTest.java b/water-modules/water-app/src/test/java/org/dromara/app/controller/AppControllerConcurrencyTest.java
new file mode 100644
index 0000000..cb5c914
--- /dev/null
+++ b/water-modules/water-app/src/test/java/org/dromara/app/controller/AppControllerConcurrencyTest.java
@@ -0,0 +1,306 @@
+package org.dromara.app.controller;
+
+import cn.hutool.extra.spring.SpringUtil;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.dromara.app.domain.bo.AppScheduleBo;
+import org.dromara.app.domain.bo.AppSchedulingDeviceBo;
+import org.dromara.app.domain.vo.AppScheduleDetailVo;
+import org.dromara.app.domain.vo.AppScheduleVo;
+import org.dromara.app.domain.vo.AppSchedulingDeviceVo;
+import org.dromara.app.service.*;
+import org.dromara.common.core.domain.R;
+import org.dromara.common.satoken.utils.LoginHelper;
+import org.dromara.system.domain.vo.SysOssUploadVo;
+import org.dromara.system.domain.vo.SysOssVo;
+import org.dromara.system.service.ISysOssService;
+import org.dromara.system.service.ISysUserService;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.MockedStatic;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.context.support.GenericApplicationContext;
+import org.springframework.mock.web.MockMultipartFile;
+import org.springframework.web.multipart.MultipartFile;
+
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
+import java.time.format.ResolverStyle;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Supplier;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.Mockito.*;
+
+@ExtendWith(MockitoExtension.class)
+@Tag("dev")
+class AppControllerConcurrencyTest {
+
+ private static final int WORKER_COUNT = 32;
+ private static final int GROUP_TIMEOUT_SECONDS = 30;
+ private static final Long USER_ID = 99L;
+ private static final DateTimeFormatter STRICT_SECOND_FORMATTER =
+ DateTimeFormatter.ofPattern("uuuu-MM-dd HH:mm:ss").withResolverStyle(ResolverStyle.STRICT);
+
+ @Mock private IAppDeviceService appDeviceService;
+ @Mock private IAppScheduleService appScheduleService;
+ @Mock private IAppScheduleDetailService appScheduleDetailService;
+ @Mock private IAppSchedulingDeviceService appSchedulingDeviceService;
+ @Mock private ISysUserService userService;
+ @Mock private IAppWateringLogService appWateringLogService;
+ @Mock private IDeviceCommandService deviceCommandService;
+ @Mock private ISysOssService ossService;
+ @Mock private IAppVersionService appVersionService;
+
+ private GenericApplicationContext applicationContext;
+
+ @BeforeEach
+ void setUp() {
+ applicationContext = new GenericApplicationContext();
+ Supplier objectMapperSupplier = ObjectMapper::new;
+ applicationContext.registerBean(ObjectMapper.class, objectMapperSupplier);
+ applicationContext.refresh();
+ new SpringUtil().setApplicationContext(applicationContext);
+ }
+
+ @AfterEach
+ void tearDown() {
+ applicationContext.close();
+ }
+
+ @Test
+ void switchDevice_formatsAllStartTimesCorrectlyUnderConcurrency() throws Exception {
+ AppController controller = newController();
+ int iterationsPerWorker = 100;
+ int expectedCalls = WORKER_COUNT * iterationsPerWorker;
+ ConcurrentLinkedQueue startTimes = new ConcurrentLinkedQueue<>();
+ AtomicInteger successfulResponses = new AtomicInteger();
+ when(appDeviceService.switchDevice(eq("D01"), eq("1"), anyString(), eq(10)))
+ .thenAnswer(invocation -> {
+ startTimes.add(invocation.getArgument(2, String.class));
+ return true;
+ });
+
+ runConcurrently(iterationsPerWorker, (worker, iteration) -> {
+ R result = controller.switchDevice(
+ "{\"deviceNo\":\"D01\",\"workStatus\":\"1\",\"durationMin\":\"10\"}");
+ assertThat(result.getCode()).isEqualTo(R.SUCCESS);
+ successfulResponses.incrementAndGet();
+ });
+
+ assertThat(successfulResponses.get()).isEqualTo(expectedCalls);
+ assertThat(startTimes).hasSize(expectedCalls).allSatisfy(startTime ->
+ assertThatCode(() -> LocalDateTime.parse(startTime, STRICT_SECOND_FORMATTER))
+ .doesNotThrowAnyException());
+ verify(appDeviceService, times(expectedCalls))
+ .switchDevice(eq("D01"), eq("1"), anyString(), eq(10));
+ }
+
+ @Test
+ void uploadImage_validatesAllSupportedFormatsUnderConcurrency() throws Exception {
+ AppController controller = newController();
+ int iterationsPerWorker = 50;
+ int expectedCalls = WORKER_COUNT * iterationsPerWorker;
+ List fixtures = supportedImages();
+ SysOssVo oss = new SysOssVo();
+ oss.setOssId(123L);
+ oss.setOriginalName("plant-image");
+ oss.setUrl("https://bucket.oss-cn-hangzhou.aliyuncs.com/plant-image");
+ when(ossService.upload(any(MultipartFile.class))).thenReturn(oss);
+
+ runConcurrently(iterationsPerWorker, (worker, iteration) -> {
+ ImageFixture fixture = fixtures.get((worker + iteration) % fixtures.size());
+ MockMultipartFile file = new MockMultipartFile(
+ "file", fixture.fileName(), fixture.contentType(), fixture.content());
+
+ R result = controller.uploadImage(file);
+
+ assertThat(result.getCode()).isEqualTo(R.SUCCESS);
+ assertThat(result.getData().getOssId()).isEqualTo("123");
+ assertThat(result.getData().getUrl()).isEqualTo(oss.getUrl());
+ });
+
+ verify(ossService, times(expectedCalls)).upload(any(MultipartFile.class));
+ }
+
+ @Test
+ void editScheduleStatus_keepsMqttPayloadsIsolatedUnderConcurrency() throws Exception {
+ AppController controller = newController();
+ int iterationsPerWorker = 20;
+ int expectedCalls = WORKER_COUNT * iterationsPerWorker;
+ ConcurrentHashMap> payloadsByDevice = new ConcurrentHashMap<>();
+ when(appScheduleService.queryById(anyLong())).thenAnswer(invocation -> {
+ Long scheduleId = invocation.getArgument(0, Long.class);
+ AppScheduleVo schedule = new AppScheduleVo();
+ schedule.setId(scheduleId);
+ schedule.setUserId(USER_ID);
+ schedule.setStatus("1");
+ return schedule;
+ });
+ when(appScheduleService.updateStatusByBo(any(AppScheduleBo.class))).thenReturn(true);
+ when(appSchedulingDeviceService.queryList(any(AppSchedulingDeviceBo.class))).thenAnswer(invocation -> {
+ AppSchedulingDeviceBo bo = invocation.getArgument(0, AppSchedulingDeviceBo.class);
+ AppSchedulingDeviceVo binding = new AppSchedulingDeviceVo();
+ binding.setScheduleId(bo.getScheduleId());
+ binding.setDeviceNo(deviceNo(bo.getScheduleId()));
+ return List.of(binding);
+ });
+ when(appScheduleDetailService.queryByScheduleIdByStatus(anyLong())).thenAnswer(invocation -> {
+ Long scheduleId = invocation.getArgument(0, Long.class);
+ AppScheduleDetailVo detail = new AppScheduleDetailVo();
+ detail.setScheduleId(scheduleId);
+ detail.setWeekday((int) ((scheduleId - 1) % 7) + 1);
+ detail.setTimeData("[{\"startTime\":\"08:00\",\"durationMin\":15}]");
+ detail.setTriggerType(detailMarker(scheduleId));
+ detail.setStatus("1");
+ return List.of(detail);
+ });
+ when(deviceCommandService.sendScheduleBindCommand(anyString(), any())).thenAnswer(invocation -> {
+ String deviceNo = invocation.getArgument(0, String.class);
+ Map