yuhaiming/water
Template
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

设备bug修改 新增设备解绑功能

Browse Source
This commit is contained in:
yuhaiming
2026-08-25 16:18:34 +08:00
parent b789c1c07e
commit 509e760f3d
87 changed files with 4458 additions and 405 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/DeviceMqttCommandPublisher.java
View File

@@ -21,6 +21,7 @@ public class DeviceMqttCommandPublisher implements IDeviceCommandPublisher {
@Override
public String send(DeviceCommand command) {
ackService.awaitStartupCleanup();
long now = System.currentTimeMillis();
if (StringUtils.isBlank(command.getCommandId())) {
command.setCommandId(UUID.randomUUID().toString().replace("-", ""));

24
water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/MqttClientManager.java
View File

@@ -51,6 +51,18 @@ public class MqttClientManager implements DisposableBean {
return identity.hashCode();
}
static String subscriptionFilter(String topic, String sharedGroup) {
if (StringUtils.isBlank(topic) || StringUtils.isBlank(sharedGroup)
|| topic.startsWith("$share/") || topic.startsWith("$queue/")) {
return topic;
}
String group = sharedGroup.trim();
if (group.contains("/") || group.contains("+") || group.contains("#")) {
throw new ServiceException("MQTT 共享订阅组名称不能包含 /、+ 或 #");
}
return "$share/" + group + "/" + topic;
}
private void configureTls(MqttConnectOptions options) throws MqttException {
if (!isTlsEnabled()) {
return;
@@ -254,14 +266,18 @@ public class MqttClientManager implements DisposableBean {
if (client == null || !client.isConnected()) {
return;
}
List<String> topics = props.getTopics() == null ? List.of() : props.getTopics().getSubscribe();
MqttProperties.Topics topicProperties = props.getTopics();
List<String> topics = topicProperties == null ? List.of() : topicProperties.getSubscribe();
if (topics == null || topics.isEmpty()) {
log.warn("[MQTT] 未配置订阅主题");
return;
}
int[] qos = topics.stream().mapToInt(topic -> props.getQos()).toArray();
client.subscribe(topics.toArray(new String[0]), qos).waitForCompletion();
topics.forEach(topic -> log.info("[MQTT] 已订阅 主题={} 服务质量等级={}", topic, props.getQos()));
List<String> subscriptionFilters = topics.stream()
.map(topic -> subscriptionFilter(topic, topicProperties.getSharedGroup()))
.toList();
int[] qos = subscriptionFilters.stream().mapToInt(topic -> props.getQos()).toArray();
client.subscribe(subscriptionFilters.toArray(new String[0]), qos).waitForCompletion();
subscriptionFilters.forEach(topic -> log.info("[MQTT] 已订阅 主题={} 服务质量等级={}", topic, props.getQos()));
} catch (MqttException e) {
log.error("[MQTT] 订阅失败", e);
}

94
water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/MqttCommandAckService.java
View File

@@ -7,7 +7,10 @@ import org.dromara.app.domain.AppDevice;
import org.dromara.app.domain.mqtt.DeviceCommand;
import org.dromara.app.domain.mqtt.DeviceCommandAck;
import org.dromara.app.mapper.AppDeviceMapper;
import org.dromara.app.service.IAppDeviceBindingService;
import org.dromara.app.service.IDeviceCommandAckHandler;
import org.dromara.app.service.IDeviceCommandLifecycleListener;
import org.dromara.common.core.exception.ServiceException;
import org.dromara.common.core.utils.StringUtils;
import org.dromara.common.json.utils.JsonUtils;
import org.dromara.common.redis.utils.RedisUtils;
@@ -21,6 +24,7 @@ import org.springframework.stereotype.Service;
import java.time.Duration;
import java.time.Instant;
import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.function.BooleanSupplier;
@@ -37,7 +41,13 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
private MqttClientManager mqttClientManager;
private final MqttProperties mqttProperties;
private final AppDeviceMapper appDeviceMapper;
private final CountDownLatch startupCleanupLatch = new CountDownLatch(1);
@Autowired(required = false)
private List<IDeviceCommandLifecycleListener> lifecycleListeners = Collections.emptyList();
private volatile boolean startupCleanupReady;
@Lazy
@Autowired(required = false)
private IAppDeviceBindingService deviceBindingService;
private static final String DEVICE_STATUS_LOCK_PREFIX = "lock:mqtt:device:status:";
static boolean ackMatchesPendingCommand(String topicDeviceNo, DeviceCommand pendingCommand) {
@@ -109,6 +119,7 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
deviceNo, ack.getCommandId(), pending.getDeviceNo());
return false;
}
notifyCommandAcknowledged(pending, ack);
deletePending(ack.getCommandId());
RedisUtils.setCacheObject(ackKey(ack.getCommandId()), ack, Duration.ofSeconds(mqttProperties.getCommandAck().getAckTtlSeconds()));
return true;
@@ -137,19 +148,23 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
RSet<String> pendingIds = pendingIds();
Set<String> commandIds = pendingIds.readAll();
pendingIds.removeAll(commandIds);
startupCleanupReady = true;
RuntimeException cleanupFailure = null;
for (String commandId : commandIds) {
try {
RedisUtils.deleteObject(pendingKey(commandId));
} catch (RuntimeException e) {
if (cleanupFailure == null) {
cleanupFailure = new RuntimeException("Failed to delete MQTT pending command object: " + commandId, e);
} else {
cleanupFailure.addSuppressed(e);
try {
for (String commandId : commandIds) {
try {
RedisUtils.deleteObject(pendingKey(commandId));
} catch (RuntimeException e) {
if (cleanupFailure == null) {
cleanupFailure = new RuntimeException("Failed to delete MQTT pending command object: " + commandId, e);
} else {
cleanupFailure.addSuppressed(e);
}
}
}
} finally {
startupCleanupReady = true;
startupCleanupLatch.countDown();
}
if (cleanupFailure != null) {
throw cleanupFailure;
@@ -161,14 +176,51 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
return startupCleanupReady;
}
void awaitStartupCleanup() {
if (startupCleanupReady) {
return;
}
long waitMs = Math.max(1000L, mqttProperties.getCommandAck().getStartupCleanupWaitMs());
try {
if (!startupCleanupLatch.await(waitMs, TimeUnit.MILLISECONDS)) {
throw new ServiceException("MQTT 启动清理未完成，暂不下发命令");
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new ServiceException("等待 MQTT 启动清理被中断");
}
}
public void removePending(String commandId) {
if (StringUtils.isBlank(commandId)) {
return;
}
withCommandLock(commandId, () -> {
CommandLockResult result = withCommandLock(commandId, () -> {
deletePending(commandId);
return true;
});
if (result != CommandLockResult.SUCCESS) {
throw new IllegalStateException("清理 MQTT 待确认命令失败，命令编号=" + commandId + "，结果=" + result);
}
}
@Override
public void clearPendingCommands(Collection<String> deviceNos) {
if (deviceNos == null || deviceNos.isEmpty()) {
return;
}
Set<String> targetDeviceNos = deviceNos.stream()
.filter(StringUtils::isNotBlank)
.collect(java.util.stream.Collectors.toSet());
if (targetDeviceNos.isEmpty()) {
return;
}
for (String commandId : pendingIds().readAll()) {
DeviceCommand command = RedisUtils.getCacheObject(pendingKey(commandId));
if (command != null && targetDeviceNos.contains(command.getDeviceNo())) {
removePending(commandId);
}
}
}
private void handlePlainAck(String deviceNo, String payload) {
@@ -199,6 +251,7 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
deviceNo, commandId, pending.getDeviceNo());
return false;
}
notifyCommandAcknowledged(pending, ack);
deletePending(commandId);
RedisUtils.setCacheObject(ackKey(commandId), ack, Duration.ofSeconds(mqttProperties.getCommandAck().getAckTtlSeconds()));
return true;
@@ -257,6 +310,7 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
return;
}
if (command.getRetryCount() >= mqttProperties.getCommandAck().getMaxRetryCount()) {
notifyCommandExpired(command);
deletePending(commandId);
log.warn("[MQTT] 命令重试次数已达上限 设备编号={} 命令编号={}", command.getDeviceNo(), commandId);
return;
@@ -300,6 +354,9 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
}
private void refreshDeviceOnline(String deviceNo) {
if (deviceBindingService != null && !deviceBindingService.isActive(deviceNo)) {
return;
}
RLock lock = RedisUtils.getClient().getLock(DEVICE_STATUS_LOCK_PREFIX + deviceNo);
boolean locked = false;
try {
@@ -341,6 +398,23 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
pendingIds().remove(commandId);
}
private void notifyCommandAcknowledged(DeviceCommand command, DeviceCommandAck ack) {
for (IDeviceCommandLifecycleListener listener : lifecycleListeners) {
listener.onCommandAcknowledged(command, ack);
}
}
private void notifyCommandExpired(DeviceCommand command) {
for (IDeviceCommandLifecycleListener listener : lifecycleListeners) {
try {
listener.onCommandExpired(command);
} catch (RuntimeException e) {
log.error("[MQTT] 命令过期生命周期回调失败 设备编号={} 命令编号={}",
command.getDeviceNo(), command.getCommandId(), e);
}
}
}
private CommandLockResult withCommandLock(String commandId, long waitTimeMs, BooleanSupplier action) {
RLock lock = RedisUtils.getClient().getLock(mqttProperties.getCommandAck().getRetryLockKeyPrefix() + commandId);
boolean locked = false;

52
water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/MqttDeviceUnbindResultPublisher.java Normal file
View File

@@ -0,0 +1,52 @@
package org.dromara.mqtt;
import lombok.RequiredArgsConstructor;
import org.dromara.app.domain.mqtt.DeviceUnbindResult;
import org.dromara.app.service.IDeviceUnbindResultPublisher;
import org.dromara.common.core.exception.ServiceException;
import org.dromara.common.core.utils.StringUtils;
import org.dromara.common.json.utils.JsonUtils;
import org.dromara.mqtt.config.properties.MqttProperties;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Component;
import java.util.LinkedHashMap;
import java.util.Locale;
import java.util.Map;
@Component
@RequiredArgsConstructor
public class MqttDeviceUnbindResultPublisher implements IDeviceUnbindResultPublisher {
private final MqttProperties mqttProperties;
@Lazy
@Autowired
private MqttClientManager mqttClientManager;
@Override
public void publish(DeviceUnbindResult result) {
if (result == null || StringUtils.isBlank(result.deviceNo())) {
throw new ServiceException("解绑回执缺少设备编号");
}
Map<String, Object> payload = new LinkedHashMap<>();
payload.put("commandId", result.commandId());
payload.put("bindingId", result.bindingId());
payload.put("status", result.status());
payload.put("message", result.message());
mqttClientManager.publish(buildTopic(result.deviceNo()), JsonUtils.toJsonString(payload));
}
private String buildTopic(String deviceNo) {
String device = deviceNo.trim().toLowerCase(Locale.ROOT);
String prefix = mqttProperties.getTopics().getPublishPrefix();
if (StringUtils.isBlank(prefix)) {
return "/" + device + "/subscriber/unbind/ack";
}
String normalizedPrefix = prefix.startsWith("/") ? prefix : "/" + prefix;
if (normalizedPrefix.endsWith("/")) {
normalizedPrefix = normalizedPrefix.substring(0, normalizedPrefix.length() - 1);
}
return normalizedPrefix + "/" + device + "/subscriber/unbind/ack";
}
}

2
water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/config/properties/MqttProperties.java
View File

@@ -31,6 +31,7 @@ public class MqttProperties {
@Data
public static class Topics {
private List<String> subscribe = new ArrayList<>();
private String sharedGroup;
private String publishPrefix;
}
@@ -59,6 +60,7 @@ public class MqttProperties {
private long scanIntervalMs = 5000;
private long pendingTtlSeconds = 86400;
private long ackTtlSeconds = 86400;
private long startupCleanupWaitMs = 30000;
private String pendingKeyPrefix = "mqtt:command:pending:";
private String ackKeyPrefix = "mqtt:command:ack:";
private String pendingSetKey = "mqtt:command:pending:ids";

21
water-common/water-common-mqtt/src/test/java/org/dromara/mqtt/DeviceMqttCommandPublisherTest.java
View File

@@ -13,6 +13,7 @@ import org.springframework.context.support.GenericApplicationContext;
import java.util.function.Supplier;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
import static org.springframework.test.util.ReflectionTestUtils.invokeMethod;
@@ -63,4 +64,24 @@ class DeviceMqttCommandPublisherTest {
assertThat(command.getNextRetryAt() - command.getLastSentAt()).isEqualTo(10000);
}
@Test
void sendWaitsForStartupCleanupBeforeSavingPendingCommand() {
MqttClientManager clientManager = mock(MqttClientManager.class);
MqttCommandAckService ackService = mock(MqttCommandAckService.class);
DeviceMqttCommandPublisher publisher = new DeviceMqttCommandPublisher(
clientManager,
ackService,
new MqttProperties()
);
DeviceCommand command = new DeviceCommand();
command.setDeviceNo("D01");
command.setCommandType("switch");
publisher.send(command);
var order = inOrder(ackService, clientManager);
order.verify(ackService).awaitStartupCleanup();
order.verify(ackService).savePending(command);
}
}

38
water-common/water-common-mqtt/src/test/java/org/dromara/mqtt/MqttClientManagerTest.java Normal file
View File

@@ -0,0 +1,38 @@
package org.dromara.mqtt;
import org.dromara.common.core.exception.ServiceException;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@Tag("dev")
class MqttClientManagerTest {
@Test
void subscriptionFilterWrapsTopicInSharedGroup() {
assertThat(MqttClientManager.subscriptionFilter("/+/publish/register", "water-backend"))
.isEqualTo("$share/water-backend//+/publish/register");
}
@Test
void subscriptionFilterKeepsExistingSharedSubscription() {
String topic = "$share/existing//+/publish/register";
assertThat(MqttClientManager.subscriptionFilter(topic, "water-backend")).isEqualTo(topic);
}
@Test
void subscriptionFilterKeepsTopicWhenSharedGroupIsNotConfigured() {
assertThat(MqttClientManager.subscriptionFilter("/+/publish/register", " "))
.isEqualTo("/+/publish/register");
}
@Test
void subscriptionFilterRejectsInvalidSharedGroup() {
assertThatThrownBy(() -> MqttClientManager.subscriptionFilter("/+/publish/register", "water/backend"))
.isInstanceOf(ServiceException.class)
.hasMessageContaining("共享订阅组名称");
}
}

36
water-common/water-common-mqtt/src/test/java/org/dromara/mqtt/MqttCommandAckServiceTest.java
View File

@@ -213,6 +213,42 @@ class MqttCommandAckServiceTest {
}
}
@Test
void clearPendingCommandsByDeviceKeepsOtherDeviceCommands() throws Exception {
MqttProperties properties = new MqttProperties();
AppDeviceMapper appDeviceMapper = mock(AppDeviceMapper.class);
RedissonClient redissonClient = mock(RedissonClient.class);
RSet<String> pendingIds = mock(RSet.class);
RLock commandLock = mock(RLock.class);
when(redissonClient.<String>getSet("mqtt:command:pending:ids")).thenReturn(pendingIds);
when(pendingIds.readAll()).thenReturn(Set.of("cmd-1", "cmd-2"));
when(redissonClient.getLock("lock:mqtt:command:retry:cmd-1")).thenReturn(commandLock);
when(commandLock.tryLock(anyLong(), anyLong(), eq(TimeUnit.MILLISECONDS))).thenReturn(true);
when(commandLock.isHeldByCurrentThread()).thenReturn(true);
DeviceCommand first = new DeviceCommand();
first.setCommandId("cmd-1");
first.setDeviceNo("D01");
DeviceCommand second = new DeviceCommand();
second.setCommandId("cmd-2");
second.setDeviceNo("D02");
MqttCommandAckService service = new MqttCommandAckService(properties, appDeviceMapper);
try (MockedStatic<RedisUtils> redis = mockStatic(RedisUtils.class)) {
redis.when(RedisUtils::getClient).thenReturn(redissonClient);
redis.when(() -> RedisUtils.getCacheObject("mqtt:command:pending:cmd-1")).thenReturn(first);
redis.when(() -> RedisUtils.getCacheObject("mqtt:command:pending:cmd-2")).thenReturn(second);
service.clearPendingCommands(List.of("D01"));
redis.verify(() -> RedisUtils.deleteObject("mqtt:command:pending:cmd-1"));
redis.verify(() -> RedisUtils.deleteObject("mqtt:command:pending:cmd-2"), never());
verify(pendingIds).remove("cmd-1");
verify(pendingIds, never()).remove("cmd-2");
verify(commandLock).unlock();
}
}
@Test
void handleAckWaitsBrieflyForCommandLock() throws Exception {
MqttProperties properties = new MqttProperties();