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

设备心跳调整 app新增增设备时绑定联网设备

Browse Source
This commit is contained in:
yuhaiming
2026-06-13 10:51:45 +08:00
parent c7f9df980a
commit c74363afa4
65 changed files with 1622 additions and 592 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
water-common/water-common-mqtt/pom.xml
View File

@@ -35,6 +35,10 @@
<groupId>org.dromara</groupId>
<artifactId>water-common-json</artifactId>
</dependency>
<dependency>
<groupId>org.dromara</groupId>
<artifactId>water-common-core</artifactId>
</dependency>
<dependency>
<groupId>org.dromara</groupId>
<artifactId>water-app</artifactId>

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

@@ -34,7 +34,12 @@ public class DeviceMqttCommandPublisher implements IDeviceCommandPublisher {
command.getPayload().put("commandType", command.getCommandType());
ackService.savePending(command);
mqttClientManager.publish(command.getTopic(), JsonUtils.toJsonString(command.getPayload()));
try {
mqttClientManager.publish(command.getTopic(), JsonUtils.toJsonString(command.getPayload()));
} catch (RuntimeException e) {
ackService.removePending(command.getCommandId());
throw e;
}
return command.getCommandId();
}

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

@@ -3,13 +3,10 @@ package org.dromara.mqtt;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.dromara.app.mqtt.MqttMessageDispatcher;
import org.dromara.common.core.exception.ServiceException;
import org.dromara.common.core.utils.StringUtils;
import org.dromara.mqtt.config.properties.MqttProperties;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.boot.autoconfigure.AutoConfiguration;
@@ -26,11 +23,7 @@ import java.security.GeneralSecurityException;
import java.security.SecureRandom;
import java.security.cert.X509Certificate;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
@Slf4j
@@ -50,6 +43,12 @@ public class MqttClientManager implements DisposableBean {
@Bean
@ConditionalOnProperty(prefix = "mqtt", name = "enabled", havingValue = "true")
public MqttAsyncClient mqttConnect() throws MqttException {
if (StringUtils.isBlank(props.getBrokerUrl())) {
throw new ServiceException("启用 MQTT 时必须配置 broker-url");
}
if (StringUtils.isBlank(props.getClientId())) {
throw new ServiceException("启用 MQTT 时必须配置 client-id");
}
messageQueue = new ArrayBlockingQueue<>(Math.max(1, props.getAsync().getQueueCapacity()));
consumerExecutor = createConsumerExecutor();
running = true;
@@ -71,13 +70,13 @@ public class MqttClientManager implements DisposableBean {
client.setCallback(new MqttCallbackExtended() {
@Override
public void connectionLost(Throwable cause) {
log.warn("[MQTT] connection lost: {}", cause == null ? "" : cause.getMessage());
log.warn("[MQTT] 连接已断开：{}", cause == null ? "" : cause.getMessage());
}
@Override
public void connectComplete(boolean reconnect, String serverURI) {
if (reconnect) {
log.info("[MQTT] reconnected: {}", serverURI);
log.info("[MQTT] 已重新连接：{}", serverURI);
subscribeTopics();
}
}
@@ -94,7 +93,7 @@ public class MqttClientManager implements DisposableBean {
});
client.connect(options).waitForCompletion();
log.info("[MQTT] connected: {}", props.getBrokerUrl());
log.info("[MQTT] 已连接：{}", props.getBrokerUrl());
subscribeTopics();
return client;
}
@@ -183,11 +182,11 @@ public class MqttClientManager implements DisposableBean {
InboundMessage inboundMessage = new InboundMessage(topic, payload);
try {
if (!messageQueue.offer(inboundMessage, props.getAsync().getOfferTimeoutMs(), TimeUnit.MILLISECONDS)) {
log.warn("[MQTT] inbound queue full, dropped topic={}", topic);
log.warn("[MQTT] 上行消息队列已满，丢弃 主题={}", topic);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
log.warn("[MQTT] enqueue interrupted topic={}", topic);
log.warn("[MQTT] 上行消息入队被中断 主题={}", topic);
}
}
@@ -221,7 +220,7 @@ public class MqttClientManager implements DisposableBean {
try {
dispatcher.dispatch(message.topic(), message.payload());
} catch (Exception e) {
log.error("[MQTT] dispatch failed topic={}", message.topic(), e);
log.error("[MQTT] 消息分发失败 主题={}", message.topic(), e);
}
}
@@ -232,30 +231,31 @@ public class MqttClientManager implements DisposableBean {
}
List<String> topics = props.getTopics() == null ? List.of() : props.getTopics().getSubscribe();
if (topics == null || topics.isEmpty()) {
log.warn("[MQTT] no subscribe topics configured");
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] subscribed: {} qos={}", topic, props.getQos()));
topics.forEach(topic -> log.info("[MQTT] 已订阅 主题={} 服务质量等级={}", topic, props.getQos()));
} catch (MqttException e) {
log.error("[MQTT] subscribe failed", e);
log.error("[MQTT] 订阅失败", e);
}
}
public void publish(String topic, String payload) {
try {
if (client == null || !client.isConnected()) {
log.warn("[MQTT] publish skipped, client not connected topic={}", topic);
return;
throw new ServiceException("MQTT客户端未连接");
}
MqttMessage message = new MqttMessage(payload.getBytes(StandardCharsets.UTF_8));
message.setQos(props.getQos());
message.setRetained(false);
client.publish(topic, message);
log.info("[MQTT] published topic={} payload={}", topic, payload);
IMqttDeliveryToken token = client.publish(topic, message);
token.waitForCompletion(Math.max(1000L, props.getConnectionTimeout() * 1000L));
log.info("[MQTT] 消息已发布 主题={} 消息体={}", topic, payload);
} catch (MqttException e) {
log.error("[MQTT] publish failed topic={}", topic, e);
log.error("[MQTT] 消息发布失败 主题={}", topic, e);
throw new ServiceException("MQTT 消息发布失败：{}", e.getMessage());
}
}

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

@@ -1,9 +1,12 @@
package org.dromara.mqtt;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
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.IDeviceCommandAckHandler;
import org.dromara.common.core.utils.StringUtils;
import org.dromara.common.json.utils.JsonUtils;
@@ -17,11 +20,7 @@ import org.springframework.stereotype.Service;
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.*;
import java.util.concurrent.TimeUnit;
@Slf4j
@@ -36,16 +35,24 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
@Autowired
private MqttClientManager mqttClientManager;
private final MqttProperties mqttProperties;
private final AppDeviceMapper appDeviceMapper;
public void savePending(DeviceCommand command) {
RedisUtils.setCacheObject(pendingKey(command.getCommandId()), command, Duration.ofSeconds(mqttProperties.getCommandAck().getPendingTtlSeconds()));
pendingIds().add(command.getCommandId());
}
public void removePending(String commandId) {
if (StringUtils.isBlank(commandId)) {
return;
}
deletePending(commandId);
}
@Override
public void handleAck(String deviceNo, String payload) {
if (StringUtils.isBlank(payload)) {
log.warn("[MQTT] 收到空 ACK deviceNo={}", deviceNo);
log.warn("[MQTT] 收到空 ACK 设备编号={}", deviceNo);
return;
}
@@ -59,26 +66,25 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
try {
ack = JsonUtils.parseObject(trimmedPayload, DeviceCommandAck.class);
} catch (RuntimeException e) {
log.warn("[MQTT] ACK JSON 格式错误 deviceNo={} payload={}", deviceNo, payload, e);
log.warn("[MQTT] ACK JSON 格式错误 设备编号={} 消息体={}", deviceNo, payload, e);
return;
}
if (ack == null || ack.getCommandId() == null) {
log.warn("[MQTT] ACK 缺少 commandId deviceNo={} payload={}", deviceNo, payload);
log.warn("[MQTT] ACK 缺少命令编号 设备编号={} 消息体={}", deviceNo, payload);
return;
}
ack.setDeviceNo(deviceNo);
RedisUtils.deleteObject(pendingKey(ack.getCommandId()));
pendingIds().remove(ack.getCommandId());
deletePending(ack.getCommandId());
RedisUtils.setCacheObject(ackKey(ack.getCommandId()), ack, Duration.ofSeconds(mqttProperties.getCommandAck().getAckTtlSeconds()));
refreshDeviceOnline(deviceNo);
log.info("[MQTT] command ack received deviceNo={} commandId={} status={}", deviceNo, ack.getCommandId(), ack.getStatus());
log.info("[MQTT] 收到命令确认 设备编号={} 命令编号={} 状态={}", deviceNo, ack.getCommandId(), ack.getStatus());
}
private void handlePlainAck(String deviceNo, String payload) {
List<DeviceCommand> pendingCommands = findPendingCommandsByDeviceNo(deviceNo);
refreshDeviceOnline(deviceNo);
if (pendingCommands.size() != 1) {
log.warn("[MQTT] 非 JSON ACK 无法唯一匹配待确认命令 deviceNo={} pendingCount={} payload={}", deviceNo, pendingCommands.size(), payload);
log.warn("[MQTT] 非 JSON ACK 无法唯一匹配待确认命令 设备编号={} 待确认数量={} 消息体={}", deviceNo, pendingCommands.size(), payload);
return;
}
@@ -90,10 +96,9 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
ack.setStatus("1");
ack.setMessage(payload);
RedisUtils.deleteObject(pendingKey(commandId));
pendingIds().remove(commandId);
deletePending(commandId);
RedisUtils.setCacheObject(ackKey(commandId), ack, Duration.ofSeconds(mqttProperties.getCommandAck().getAckTtlSeconds()));
log.info("[MQTT] plain command ack received deviceNo={} commandId={} payload={}", deviceNo, commandId, payload);
log.info("[MQTT] 收到非 JSON 命令确认 设备编号={} 命令编号={} 消息体={}", deviceNo, commandId, payload);
}
private List<DeviceCommand> findPendingCommandsByDeviceNo(String deviceNo) {
@@ -126,7 +131,9 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
retryCommand(commandId, now);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
log.warn("[MQTT] 命令重试锁等待被中断 commandId={}", commandId);
log.warn("[MQTT] 命令重试锁等待被中断 命令编号={}", commandId);
} catch (RuntimeException e) {
log.error("[MQTT] 命令重试失败 命令编号={}", commandId, e);
} finally {
if (locked && lock.isHeldByCurrentThread()) {
lock.unlock();
@@ -144,15 +151,20 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
return;
}
if (command.getRetryCount() >= mqttProperties.getCommandAck().getMaxRetryCount()) {
RedisUtils.deleteObject(pendingKey(commandId));
pendingIds().remove(commandId);
log.warn("[MQTT] 命令重试次数已达上限 deviceNo={} commandId={}", command.getDeviceNo(), commandId);
deletePending(commandId);
log.warn("[MQTT] 命令重试次数已达上限 设备编号={} 命令编号={}", command.getDeviceNo(), commandId);
return;
}
if (!isDeviceOnline(command.getDeviceNo())) {
command.setRetryCount(command.getRetryCount() + 1);
if (command.getRetryCount() >= mqttProperties.getCommandAck().getMaxRetryCount()) {
deletePending(commandId);
log.warn("[MQTT] 设备离线，命令达到最大等待次数，已删除待确认命令 设备编号={} 命令编号={}", command.getDeviceNo(), commandId);
return;
}
command.setNextRetryAt(now + mqttProperties.getCommandAck().getRetryIntervalMs());
savePending(command);
log.debug("[MQTT] command retry delayed, device offline deviceNo={} commandId={}", command.getDeviceNo(), commandId);
log.debug("[MQTT] 设备离线，命令延后重试 设备编号={} 命令编号={} 等待次数={}", command.getDeviceNo(), commandId, command.getRetryCount());
return;
}
@@ -161,7 +173,7 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
command.setNextRetryAt(now + mqttProperties.getCommandAck().getRetryIntervalMs());
mqttClientManager.publish(command.getTopic(), JsonUtils.toJsonString(command.getPayload()));
savePending(command);
log.info("[MQTT] command resent deviceNo={} commandId={} retry={}", command.getDeviceNo(), commandId, command.getRetryCount());
log.info("[MQTT] 命令已重新下发 设备编号={} 命令编号={} 重试次数={}", command.getDeviceNo(), commandId, command.getRetryCount());
}
private boolean isDeviceOnline(String deviceNo) {
@@ -183,12 +195,22 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
statusCache,
Duration.ofSeconds(mqttProperties.getCommandAck().getDeviceStatusCacheTtlSeconds())
);
appDeviceMapper.update(null,
new LambdaUpdateWrapper<AppDevice>()
.set(AppDevice::getStatus, "1")
.eq(AppDevice::getDeviceNo, deviceNo)
);
}
private RSet<String> pendingIds() {
return RedisUtils.getClient().getSet(mqttProperties.getCommandAck().getPendingSetKey());
}
private void deletePending(String commandId) {
RedisUtils.deleteObject(pendingKey(commandId));
pendingIds().remove(commandId);
}
private String pendingKey(String commandId) {
return mqttProperties.getCommandAck().getPendingKeyPrefix() + commandId;
}

3
water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/MqttCommandRetryTask.java
View File

@@ -1,10 +1,12 @@
package org.dromara.mqtt;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.dromara.mqtt.config.properties.MqttProperties;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
@Slf4j
@Component
@RequiredArgsConstructor
public class MqttCommandRetryTask {
@@ -15,6 +17,7 @@ public class MqttCommandRetryTask {
@Scheduled(fixedDelayString = "${mqtt.command-ack.scan-interval-ms:5000}")
public void retryExpiredCommands() {
if (mqttProperties.getCommandAck().isEnabled()) {
// log.info("[设备命令重新下发] 定时器调用成功 ");
ackService.retryExpiredCommands();
}
}