mqtt对接嵌入式
This commit is contained in:
yuhaiming
@@ -6,7 +6,7 @@
|
||||
<parent>
|
||||
<groupId>org.dromara</groupId>
|
||||
<artifactId>water-common</artifactId>
|
||||
<version>5.6.0</version>
|
||||
<version>${revision}</version>
|
||||
</parent>
|
||||
|
||||
<artifactId>water-common-mqtt</artifactId>
|
||||
@@ -40,4 +40,4 @@
|
||||
<artifactId>water-app</artifactId>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
</project>
|
||||
</project>
|
||||
|
||||
@@ -0,0 +1,52 @@
|
||||
package org.dromara.mqtt;
|
||||
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import org.dromara.app.domain.mqtt.DeviceCommand;
|
||||
import org.dromara.app.service.IDeviceCommandPublisher;
|
||||
import org.dromara.common.core.utils.StringUtils;
|
||||
import org.dromara.common.json.utils.JsonUtils;
|
||||
import org.dromara.mqtt.config.properties.MqttProperties;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.util.UUID;
|
||||
|
||||
@Component
|
||||
@RequiredArgsConstructor
|
||||
public class DeviceMqttCommandPublisher implements IDeviceCommandPublisher {
|
||||
|
||||
private final MqttClientManager mqttClientManager;
|
||||
private final MqttCommandAckService ackService;
|
||||
private final MqttProperties mqttProperties;
|
||||
|
||||
@Override
|
||||
public String send(DeviceCommand command) {
|
||||
long now = System.currentTimeMillis();
|
||||
if (StringUtils.isBlank(command.getCommandId())) {
|
||||
command.setCommandId(UUID.randomUUID().toString().replace("-", ""));
|
||||
}
|
||||
if (StringUtils.isBlank(command.getTopic())) {
|
||||
command.setTopic(buildCommandTopic(command.getDeviceNo()));
|
||||
}
|
||||
command.setCreatedAt(now);
|
||||
command.setLastSentAt(now);
|
||||
command.setNextRetryAt(now + mqttProperties.getCommandAck().getRetryIntervalMs());
|
||||
command.getPayload().put("commandId", command.getCommandId());
|
||||
command.getPayload().put("commandType", command.getCommandType());
|
||||
|
||||
ackService.savePending(command);
|
||||
mqttClientManager.publish(command.getTopic(), JsonUtils.toJsonString(command.getPayload()));
|
||||
return command.getCommandId();
|
||||
}
|
||||
|
||||
private String buildCommandTopic(String deviceNo) {
|
||||
String prefix = mqttProperties.getTopics().getPublishPrefix();
|
||||
if (StringUtils.isBlank(prefix)) {
|
||||
return "/" + deviceNo + "/subscriber/cmd";
|
||||
}
|
||||
String normalizedPrefix = prefix.startsWith("/") ? prefix : "/" + prefix;
|
||||
if (normalizedPrefix.endsWith("/")) {
|
||||
normalizedPrefix = normalizedPrefix.substring(0, normalizedPrefix.length() - 1);
|
||||
}
|
||||
return normalizedPrefix + "/" + deviceNo + "/subscriber/cmd";
|
||||
}
|
||||
}
|
||||
@@ -4,91 +4,278 @@ import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.dromara.app.mqtt.MqttMessageDispatcher;
|
||||
import org.dromara.mqtt.config.properties.MqttProperties;
|
||||
import org.eclipse.paho.client.mqttv3.*;
|
||||
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.persist.MemoryPersistence;
|
||||
import org.springframework.beans.factory.DisposableBean;
|
||||
import org.springframework.beans.factory.InitializingBean;
|
||||
import org.springframework.boot.autoconfigure.AutoConfiguration;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||
import org.springframework.boot.context.properties.EnableConfigurationProperties;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.scheduling.annotation.EnableScheduling;
|
||||
|
||||
import javax.net.ssl.SSLContext;
|
||||
import javax.net.ssl.TrustManager;
|
||||
import javax.net.ssl.X509TrustManager;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
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.atomic.AtomicLong;
|
||||
|
||||
@Slf4j
|
||||
@AutoConfiguration
|
||||
@EnableScheduling
|
||||
@EnableConfigurationProperties(MqttProperties.class)
|
||||
@RequiredArgsConstructor
|
||||
public class MqttClientManager implements InitializingBean, DisposableBean {
|
||||
public class MqttClientManager implements DisposableBean {
|
||||
|
||||
private final MqttProperties props;
|
||||
private final MqttMessageDispatcher dispatcher;
|
||||
private MqttClient client;
|
||||
|
||||
@Override
|
||||
public void afterPropertiesSet() throws Exception {
|
||||
// mqttConnect();
|
||||
}
|
||||
private MqttAsyncClient client;
|
||||
private ThreadPoolExecutor consumerExecutor;
|
||||
private BlockingQueue<InboundMessage> messageQueue;
|
||||
private volatile boolean running;
|
||||
|
||||
@Bean
|
||||
private MqttClient mqttConnect() throws MqttException {
|
||||
client = new MqttClient(props.getBrokerUrl(), props.getClientId(), new MemoryPersistence());
|
||||
@ConditionalOnProperty(prefix = "mqtt", name = "enabled", havingValue = "true")
|
||||
public MqttAsyncClient mqttConnect() throws MqttException {
|
||||
messageQueue = new ArrayBlockingQueue<>(Math.max(1, props.getAsync().getQueueCapacity()));
|
||||
consumerExecutor = createConsumerExecutor();
|
||||
running = true;
|
||||
startConsumers();
|
||||
client = new MqttAsyncClient(props.getBrokerUrl(), props.getClientId(), new MemoryPersistence());
|
||||
|
||||
MqttConnectOptions options = new MqttConnectOptions();
|
||||
options.setUserName(props.getUsername());
|
||||
options.setPassword(props.getPassword().toCharArray());
|
||||
if (props.getPassword() != null) {
|
||||
options.setPassword(props.getPassword().toCharArray());
|
||||
}
|
||||
options.setKeepAliveInterval(props.getKeepAlive());
|
||||
options.setConnectionTimeout(props.getConnectionTimeout());
|
||||
options.setAutomaticReconnect(true);
|
||||
options.setCleanSession(false);
|
||||
options.setAutomaticReconnect(props.isAutomaticReconnect());
|
||||
options.setCleanSession(props.isCleanSession());
|
||||
options.setMaxInflight(props.getMaxInflight());
|
||||
configureTls(options);
|
||||
|
||||
client.setCallback(new MqttCallback() {
|
||||
client.setCallback(new MqttCallbackExtended() {
|
||||
@Override
|
||||
public void connectionLost(Throwable cause) {
|
||||
log.warn("[MQTT] 连接断开: {}", cause.getMessage());
|
||||
log.warn("[MQTT] connection lost: {}", cause == null ? "" : cause.getMessage());
|
||||
}
|
||||
|
||||
@Override
|
||||
public void connectComplete(boolean reconnect, String serverURI) {
|
||||
if (reconnect) {
|
||||
log.info("[MQTT] reconnected: {}", serverURI);
|
||||
subscribeTopics();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void messageArrived(String topic, MqttMessage message) {
|
||||
dispatcher.dispatch(topic, new String(message.getPayload()));
|
||||
String payload = new String(message.getPayload(), StandardCharsets.UTF_8);
|
||||
enqueue(topic, payload);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void deliveryComplete(IMqttDeliveryToken token) {}
|
||||
public void deliveryComplete(IMqttDeliveryToken token) {
|
||||
}
|
||||
});
|
||||
|
||||
client.connect(options);
|
||||
log.info("[MQTT] 连接成功: {}", props.getBrokerUrl());
|
||||
|
||||
// 订阅所有配置的主题
|
||||
for (String topic : props.getTopics().getSubscribe()) {
|
||||
client.subscribe(topic, props.getQos());
|
||||
// log.info("[MQTT] 订阅主题: {}", topic);
|
||||
log.info("[MQTT] 订阅成功: {} qos={}", topic, props.getQos()); // ← 加这行
|
||||
}
|
||||
|
||||
client.connect(options).waitForCompletion();
|
||||
log.info("[MQTT] connected: {}", props.getBrokerUrl());
|
||||
subscribeTopics();
|
||||
return client;
|
||||
}
|
||||
|
||||
/**
|
||||
* 发布消息
|
||||
*/
|
||||
private void configureTls(MqttConnectOptions options) throws MqttException {
|
||||
if (!isTlsEnabled()) {
|
||||
return;
|
||||
}
|
||||
|
||||
options.setHttpsHostnameVerificationEnabled(!props.getTls().isSkipVerify());
|
||||
if (props.getTls().isSkipVerify()) {
|
||||
try {
|
||||
options.setSocketFactory(createTrustAllSslContext().getSocketFactory());
|
||||
options.setSSLHostnameVerifier((hostname, session) -> true);
|
||||
} catch (GeneralSecurityException e) {
|
||||
throw new MqttException(e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private boolean isTlsEnabled() {
|
||||
return props.getTls() != null
|
||||
&& props.getTls().isEnabled()
|
||||
&& props.getBrokerUrl() != null
|
||||
&& props.getBrokerUrl().startsWith("ssl://");
|
||||
}
|
||||
|
||||
private SSLContext createTrustAllSslContext() throws GeneralSecurityException {
|
||||
TrustManager[] trustManagers = new TrustManager[] {
|
||||
new X509TrustManager() {
|
||||
@Override
|
||||
public void checkClientTrusted(X509Certificate[] chain, String authType) {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void checkServerTrusted(X509Certificate[] chain, String authType) {
|
||||
}
|
||||
|
||||
@Override
|
||||
public X509Certificate[] getAcceptedIssuers() {
|
||||
return new X509Certificate[0];
|
||||
}
|
||||
}
|
||||
};
|
||||
SSLContext sslContext = SSLContext.getInstance("TLS");
|
||||
sslContext.init(null, trustManagers, new SecureRandom());
|
||||
return sslContext;
|
||||
}
|
||||
|
||||
private ThreadPoolExecutor createConsumerExecutor() {
|
||||
MqttProperties.Async async = props.getAsync();
|
||||
int corePoolSize = Math.max(1, async.getCorePoolSize());
|
||||
int maxPoolSize = Math.max(corePoolSize, async.getMaxPoolSize());
|
||||
|
||||
ThreadPoolExecutor executor = new ThreadPoolExecutor(
|
||||
corePoolSize,
|
||||
maxPoolSize,
|
||||
60L,
|
||||
TimeUnit.SECONDS,
|
||||
new ArrayBlockingQueue<>(maxPoolSize),
|
||||
mqttThreadFactory(),
|
||||
new ThreadPoolExecutor.CallerRunsPolicy()
|
||||
);
|
||||
executor.allowCoreThreadTimeOut(true);
|
||||
return executor;
|
||||
}
|
||||
|
||||
private ThreadFactory mqttThreadFactory() {
|
||||
AtomicLong threadNo = new AtomicLong();
|
||||
return runnable -> {
|
||||
Thread thread = new Thread(runnable);
|
||||
thread.setName("mqtt-consumer-" + threadNo.incrementAndGet());
|
||||
thread.setDaemon(true);
|
||||
return thread;
|
||||
};
|
||||
}
|
||||
|
||||
private void startConsumers() {
|
||||
int consumerCount = Math.max(1, props.getAsync().getConsumerCount());
|
||||
for (int i = 0; i < consumerCount; i++) {
|
||||
consumerExecutor.execute(this::consumeLoop);
|
||||
}
|
||||
}
|
||||
|
||||
private void enqueue(String topic, String payload) {
|
||||
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);
|
||||
}
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
log.warn("[MQTT] enqueue interrupted topic={}", topic);
|
||||
}
|
||||
}
|
||||
|
||||
private void consumeLoop() {
|
||||
MqttProperties.Async async = props.getAsync();
|
||||
int batchSize = Math.max(1, async.getBatchSize());
|
||||
long pollTimeoutMs = Math.max(1, async.getPollTimeoutMs());
|
||||
List<InboundMessage> batch = new java.util.ArrayList<>(batchSize);
|
||||
|
||||
while (running || !messageQueue.isEmpty()) {
|
||||
try {
|
||||
InboundMessage first = messageQueue.poll(pollTimeoutMs, TimeUnit.MILLISECONDS);
|
||||
if (first == null) {
|
||||
continue;
|
||||
}
|
||||
batch.add(first);
|
||||
messageQueue.drainTo(batch, batchSize - 1);
|
||||
for (InboundMessage message : batch) {
|
||||
dispatch(message);
|
||||
}
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
break;
|
||||
} finally {
|
||||
batch.clear();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void dispatch(InboundMessage message) {
|
||||
try {
|
||||
dispatcher.dispatch(message.topic(), message.payload());
|
||||
} catch (Exception e) {
|
||||
log.error("[MQTT] dispatch failed topic={}", message.topic(), e);
|
||||
}
|
||||
}
|
||||
|
||||
private void subscribeTopics() {
|
||||
try {
|
||||
if (client == null || !client.isConnected()) {
|
||||
return;
|
||||
}
|
||||
List<String> topics = props.getTopics() == null ? List.of() : props.getTopics().getSubscribe();
|
||||
if (topics == null || topics.isEmpty()) {
|
||||
log.warn("[MQTT] no subscribe topics configured");
|
||||
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()));
|
||||
} catch (MqttException e) {
|
||||
log.error("[MQTT] subscribe failed", 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;
|
||||
}
|
||||
MqttMessage message = new MqttMessage(payload.getBytes(StandardCharsets.UTF_8));
|
||||
message.setQos(props.getQos());
|
||||
message.setRetained(false);
|
||||
client.publish(topic, message);
|
||||
log.info("[MQTT] 发布消息 topic={} payload={}", topic, payload);
|
||||
log.info("[MQTT] published topic={} payload={}", topic, payload);
|
||||
} catch (MqttException e) {
|
||||
log.error("[MQTT] 发布失败 topic={}", topic, e);
|
||||
log.error("[MQTT] publish failed topic={}", topic, e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void destroy() throws Exception {
|
||||
if (client != null && client.isConnected()) {
|
||||
client.disconnect();
|
||||
if (client != null) {
|
||||
if (client.isConnected()) {
|
||||
client.disconnect().waitForCompletion(5000);
|
||||
}
|
||||
client.close();
|
||||
}
|
||||
if (consumerExecutor != null) {
|
||||
running = false;
|
||||
consumerExecutor.shutdown();
|
||||
if (!consumerExecutor.awaitTermination(10, TimeUnit.SECONDS)) {
|
||||
consumerExecutor.shutdownNow();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private record InboundMessage(String topic, String payload) {
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,199 @@
|
||||
package org.dromara.mqtt;
|
||||
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.dromara.app.domain.mqtt.DeviceCommand;
|
||||
import org.dromara.app.domain.mqtt.DeviceCommandAck;
|
||||
import org.dromara.app.service.IDeviceCommandAckHandler;
|
||||
import org.dromara.common.core.utils.StringUtils;
|
||||
import org.dromara.common.json.utils.JsonUtils;
|
||||
import org.dromara.common.redis.utils.RedisUtils;
|
||||
import org.dromara.mqtt.config.properties.MqttProperties;
|
||||
import org.redisson.api.RLock;
|
||||
import org.redisson.api.RSet;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.context.annotation.Lazy;
|
||||
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.concurrent.TimeUnit;
|
||||
|
||||
@Slf4j
|
||||
@Service
|
||||
@RequiredArgsConstructor
|
||||
public class MqttCommandAckService implements IDeviceCommandAckHandler {
|
||||
|
||||
// 字段注入 + @Lazy 打破 MqttClientManager → MqttMessageDispatcher → MqttCommandAckService → MqttClientManager 循环
|
||||
// publish() 仅在 retryExpiredCommands() 运行时调用，构造阶段无需立即解析
|
||||
// 注意：不能声明为 final，否则 Lombok @RequiredArgsConstructor 会放入构造器，@Lazy 失效
|
||||
@Lazy
|
||||
@Autowired
|
||||
private MqttClientManager mqttClientManager;
|
||||
private final MqttProperties mqttProperties;
|
||||
|
||||
public void savePending(DeviceCommand command) {
|
||||
RedisUtils.setCacheObject(pendingKey(command.getCommandId()), command, Duration.ofSeconds(mqttProperties.getCommandAck().getPendingTtlSeconds()));
|
||||
pendingIds().add(command.getCommandId());
|
||||
}
|
||||
|
||||
@Override
|
||||
public void handleAck(String deviceNo, String payload) {
|
||||
if (StringUtils.isBlank(payload)) {
|
||||
log.warn("[MQTT] 收到空 ACK deviceNo={}", deviceNo);
|
||||
return;
|
||||
}
|
||||
|
||||
String trimmedPayload = payload.trim();
|
||||
if (!trimmedPayload.startsWith("{")) {
|
||||
handlePlainAck(deviceNo, trimmedPayload);
|
||||
return;
|
||||
}
|
||||
|
||||
DeviceCommandAck ack;
|
||||
try {
|
||||
ack = JsonUtils.parseObject(trimmedPayload, DeviceCommandAck.class);
|
||||
} catch (RuntimeException e) {
|
||||
log.warn("[MQTT] ACK JSON 格式错误 deviceNo={} payload={}", deviceNo, payload, e);
|
||||
return;
|
||||
}
|
||||
if (ack == null || ack.getCommandId() == null) {
|
||||
log.warn("[MQTT] ACK 缺少 commandId deviceNo={} payload={}", deviceNo, payload);
|
||||
return;
|
||||
}
|
||||
ack.setDeviceNo(deviceNo);
|
||||
RedisUtils.deleteObject(pendingKey(ack.getCommandId()));
|
||||
pendingIds().remove(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());
|
||||
}
|
||||
|
||||
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);
|
||||
return;
|
||||
}
|
||||
|
||||
DeviceCommand command = pendingCommands.get(0);
|
||||
String commandId = command.getCommandId();
|
||||
DeviceCommandAck ack = new DeviceCommandAck();
|
||||
ack.setDeviceNo(deviceNo);
|
||||
ack.setCommandId(commandId);
|
||||
ack.setStatus("1");
|
||||
ack.setMessage(payload);
|
||||
|
||||
RedisUtils.deleteObject(pendingKey(commandId));
|
||||
pendingIds().remove(commandId);
|
||||
RedisUtils.setCacheObject(ackKey(commandId), ack, Duration.ofSeconds(mqttProperties.getCommandAck().getAckTtlSeconds()));
|
||||
log.info("[MQTT] plain command ack received deviceNo={} commandId={} payload={}", deviceNo, commandId, payload);
|
||||
}
|
||||
|
||||
private List<DeviceCommand> findPendingCommandsByDeviceNo(String deviceNo) {
|
||||
List<DeviceCommand> commands = new ArrayList<>();
|
||||
for (String commandId : pendingIds().readAll()) {
|
||||
DeviceCommand command = RedisUtils.getCacheObject(pendingKey(commandId));
|
||||
if (command != null && deviceNo.equals(command.getDeviceNo())) {
|
||||
commands.add(command);
|
||||
}
|
||||
}
|
||||
return commands;
|
||||
}
|
||||
|
||||
public void retryExpiredCommands() {
|
||||
long now = System.currentTimeMillis();
|
||||
Collection<String> commandIds = pendingIds().readAll();
|
||||
for (String commandId : commandIds) {
|
||||
retryCommandIfLocked(commandId, now);
|
||||
}
|
||||
}
|
||||
|
||||
private void retryCommandIfLocked(String commandId, long now) {
|
||||
RLock lock = RedisUtils.getClient().getLock(mqttProperties.getCommandAck().getRetryLockKeyPrefix() + commandId);
|
||||
boolean locked = false;
|
||||
try {
|
||||
locked = lock.tryLock(0, mqttProperties.getCommandAck().getRetryLockTtlMs(), TimeUnit.MILLISECONDS);
|
||||
if (!locked) {
|
||||
return;
|
||||
}
|
||||
retryCommand(commandId, now);
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
log.warn("[MQTT] 命令重试锁等待被中断 commandId={}", commandId);
|
||||
} finally {
|
||||
if (locked && lock.isHeldByCurrentThread()) {
|
||||
lock.unlock();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void retryCommand(String commandId, long now) {
|
||||
DeviceCommand command = RedisUtils.getCacheObject(pendingKey(commandId));
|
||||
if (command == null) {
|
||||
pendingIds().remove(commandId);
|
||||
return;
|
||||
}
|
||||
if (command.getNextRetryAt() > now) {
|
||||
return;
|
||||
}
|
||||
if (command.getRetryCount() >= mqttProperties.getCommandAck().getMaxRetryCount()) {
|
||||
RedisUtils.deleteObject(pendingKey(commandId));
|
||||
pendingIds().remove(commandId);
|
||||
log.warn("[MQTT] 命令重试次数已达上限 deviceNo={} commandId={}", command.getDeviceNo(), commandId);
|
||||
return;
|
||||
}
|
||||
if (!isDeviceOnline(command.getDeviceNo())) {
|
||||
command.setNextRetryAt(now + mqttProperties.getCommandAck().getRetryIntervalMs());
|
||||
savePending(command);
|
||||
log.debug("[MQTT] command retry delayed, device offline deviceNo={} commandId={}", command.getDeviceNo(), commandId);
|
||||
return;
|
||||
}
|
||||
|
||||
command.setRetryCount(command.getRetryCount() + 1);
|
||||
command.setLastSentAt(now);
|
||||
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());
|
||||
}
|
||||
|
||||
private boolean isDeviceOnline(String deviceNo) {
|
||||
Map<String, Object> status = RedisUtils.getCacheObject(mqttProperties.getCommandAck().getDeviceStatusCachePrefix() + deviceNo);
|
||||
if (status == null) {
|
||||
return false;
|
||||
}
|
||||
Object value = status.get("status");
|
||||
return value == null || !"0".equals(String.valueOf(value));
|
||||
}
|
||||
|
||||
private void refreshDeviceOnline(String deviceNo) {
|
||||
Map<String, Object> statusCache = new HashMap<>();
|
||||
statusCache.put("deviceNo", deviceNo);
|
||||
statusCache.put("status", "1");
|
||||
statusCache.put("lastReportTime", Instant.now().toString());
|
||||
RedisUtils.setCacheObject(
|
||||
mqttProperties.getCommandAck().getDeviceStatusCachePrefix() + deviceNo,
|
||||
statusCache,
|
||||
Duration.ofSeconds(mqttProperties.getCommandAck().getDeviceStatusCacheTtlSeconds())
|
||||
);
|
||||
}
|
||||
|
||||
private RSet<String> pendingIds() {
|
||||
return RedisUtils.getClient().getSet(mqttProperties.getCommandAck().getPendingSetKey());
|
||||
}
|
||||
|
||||
private String pendingKey(String commandId) {
|
||||
return mqttProperties.getCommandAck().getPendingKeyPrefix() + commandId;
|
||||
}
|
||||
|
||||
private String ackKey(String commandId) {
|
||||
return mqttProperties.getCommandAck().getAckKeyPrefix() + commandId;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,21 @@
|
||||
package org.dromara.mqtt;
|
||||
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import org.dromara.mqtt.config.properties.MqttProperties;
|
||||
import org.springframework.scheduling.annotation.Scheduled;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
@Component
|
||||
@RequiredArgsConstructor
|
||||
public class MqttCommandRetryTask {
|
||||
|
||||
private final MqttCommandAckService ackService;
|
||||
private final MqttProperties mqttProperties;
|
||||
|
||||
@Scheduled(fixedDelayString = "${mqtt.command-ack.scan-interval-ms:5000}")
|
||||
public void retryExpiredCommands() {
|
||||
if (mqttProperties.getCommandAck().isEnabled()) {
|
||||
ackService.retryExpiredCommands();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -3,8 +3,8 @@ package org.dromara.mqtt.config.properties;
|
||||
|
||||
import lombok.Data;
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
@Data
|
||||
@@ -12,6 +12,7 @@ import java.util.List;
|
||||
@ConfigurationProperties(prefix = "mqtt")
|
||||
public class MqttProperties {
|
||||
|
||||
private boolean enabled = false;
|
||||
private String brokerUrl;
|
||||
private String username;
|
||||
private String password;
|
||||
@@ -19,11 +20,52 @@ public class MqttProperties {
|
||||
private int qos = 1;
|
||||
private int keepAlive = 60;
|
||||
private int connectionTimeout = 30;
|
||||
private Topics topics;
|
||||
private int maxInflight = 1000;
|
||||
private boolean cleanSession = false;
|
||||
private boolean automaticReconnect = true;
|
||||
private Topics topics = new Topics();
|
||||
private Async async = new Async();
|
||||
private Tls tls = new Tls();
|
||||
private CommandAck commandAck = new CommandAck();
|
||||
|
||||
@Data
|
||||
public static class Topics {
|
||||
private List<String> subscribe;
|
||||
private List<String> subscribe = new ArrayList<>();
|
||||
private String publishPrefix;
|
||||
}
|
||||
|
||||
@Data
|
||||
public static class Async {
|
||||
private int corePoolSize = 8;
|
||||
private int maxPoolSize = 32;
|
||||
private int consumerCount = 16;
|
||||
private int queueCapacity = 5000;
|
||||
private int batchSize = 100;
|
||||
private long offerTimeoutMs = 50;
|
||||
private long pollTimeoutMs = 100;
|
||||
}
|
||||
|
||||
@Data
|
||||
public static class Tls {
|
||||
private boolean enabled = true;
|
||||
private boolean skipVerify = false;
|
||||
}
|
||||
|
||||
@Data
|
||||
public static class CommandAck {
|
||||
private boolean enabled = true;
|
||||
private int maxRetryCount = 3;
|
||||
private long retryIntervalMs = 5000;
|
||||
private long scanIntervalMs = 5000;
|
||||
private long pendingTtlSeconds = 86400;
|
||||
private long ackTtlSeconds = 86400;
|
||||
private String pendingKeyPrefix = "mqtt:command:pending:";
|
||||
private String ackKeyPrefix = "mqtt:command:ack:";
|
||||
private String pendingSetKey = "mqtt:command:pending:ids";
|
||||
private String retryLockKeyPrefix = "lock:mqtt:command:retry:";
|
||||
private long retryLockTtlMs = 30000;
|
||||
private String deviceStatusCachePrefix = "mqtt:device:status:";
|
||||
private long deviceStatusCacheTtlSeconds = 300;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -1,2 +1,5 @@
|
||||
org.dromara.mqtt.MqttClientManager
|
||||
org.dromara.app.mqtt.MqttMessageDispatcher
|
||||
org.dromara.mqtt.MqttCommandAckService
|
||||
org.dromara.mqtt.DeviceMqttCommandPublisher
|
||||
org.dromara.mqtt.MqttCommandRetryTask
|
||||
org.dromara.app.mqtt.MqttMessageDispatcher
|
||||
|
||||
Reference in New Issue
Block a user