package cn.iocoder.yudao.module.mes.framework.collect.mqtt;
|
|
import cn.hutool.core.util.StrUtil;
|
import cn.iocoder.yudao.module.mes.framework.collect.config.MesCollectProperties;
|
import cn.iocoder.yudao.module.mes.service.collect.MesCollectMessageDispatcher;
|
import jakarta.annotation.Resource;
|
import lombok.extern.slf4j.Slf4j;
|
import org.eclipse.paho.client.mqttv3.*;
|
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
|
import org.springframework.context.SmartLifecycle;
|
import org.springframework.stereotype.Component;
|
|
import java.net.InetAddress;
|
import java.nio.charset.StandardCharsets;
|
import java.util.Arrays;
|
import java.util.Collections;
|
import java.util.List;
|
import java.util.concurrent.Executors;
|
import java.util.concurrent.ScheduledExecutorService;
|
import java.util.concurrent.TimeUnit;
|
|
/**
|
* MES 数采 MQTT 客户端
|
*
|
* @author 超级管理员
|
*/
|
@Component
|
@Slf4j
|
public class MesCollectMqttClient implements SmartLifecycle {
|
|
private static final long RECONNECT_DELAY_SECONDS = 30L;
|
|
@Resource
|
private MesCollectProperties properties;
|
@Resource
|
private MesCollectMessageDispatcher dispatcher;
|
|
private volatile boolean running = false;
|
private MqttClient client;
|
private ScheduledExecutorService scheduler;
|
|
@Override
|
public void start() {
|
MesCollectProperties.Mqtt mqtt = properties.getMqtt();
|
if (!Boolean.TRUE.equals(mqtt.getEnabled())) {
|
log.info("[start][数采 MQTT 未启用]");
|
return;
|
}
|
if (StrUtil.isBlank(mqtt.getUrl())) {
|
log.warn("[start][数采 MQTT 已启用但 url 为空，跳过连接]");
|
return;
|
}
|
running = true;
|
scheduler = Executors.newSingleThreadScheduledExecutor(runnable -> {
|
Thread thread = new Thread(runnable, "mes-collect-mqtt-reconnect");
|
thread.setDaemon(true);
|
return thread;
|
});
|
scheduler.scheduleWithFixedDelay(this::ensureConnected, 0, RECONNECT_DELAY_SECONDS, TimeUnit.SECONDS);
|
}
|
|
private void ensureConnected() {
|
if (!running) {
|
return;
|
}
|
MesCollectProperties.Mqtt mqtt = properties.getMqtt();
|
try {
|
if (client == null) {
|
client = new MqttClient(mqtt.getUrl(), buildClientId(), new MemoryPersistence());
|
client.setCallback(new MqttCallbackExtended() {
|
@Override
|
public void connectComplete(boolean reconnect, String serverURI) {
|
subscribe();
|
}
|
|
@Override
|
public void connectionLost(Throwable cause) {
|
log.warn("[connectionLost][数采 MQTT 连接断开：{}]", cause == null ? "未知原因" : cause.getMessage());
|
}
|
|
@Override
|
public void messageArrived(String topic, MqttMessage message) {
|
String payload = new String(message.getPayload(), StandardCharsets.UTF_8);
|
try {
|
dispatcher.dispatch(topic, payload);
|
} catch (Exception e) {
|
log.error("[messageArrived][主题({}) 入队失败]", topic, e);
|
}
|
}
|
|
@Override
|
public void deliveryComplete(IMqttDeliveryToken token) {
|
// 当前只订阅，不发布
|
}
|
});
|
}
|
if (!client.isConnected()) {
|
MqttConnectOptions options = new MqttConnectOptions();
|
options.setAutomaticReconnect(true);
|
options.setCleanSession(true);
|
options.setKeepAliveInterval(mqtt.getKeepAlive());
|
options.setConnectionTimeout(mqtt.getConnectionTimeout());
|
if (StrUtil.isNotBlank(mqtt.getUsername())) {
|
options.setUserName(mqtt.getUsername());
|
}
|
if (mqtt.getPassword() != null) {
|
options.setPassword(mqtt.getPassword().toCharArray());
|
}
|
client.connect(options);
|
log.info("[ensureConnected][数采 MQTT 已连接：{}]", mqtt.getUrl());
|
}
|
} catch (Exception e) {
|
log.warn("[ensureConnected][数采 MQTT 连接/订阅失败，{} 秒后重试：{}]",
|
RECONNECT_DELAY_SECONDS, e.getMessage());
|
}
|
}
|
|
private void subscribe() {
|
MesCollectProperties.Mqtt mqtt = properties.getMqtt();
|
for (String topic : resolveSubscribeTopics(mqtt)) {
|
try {
|
client.subscribe(topic, mqtt.getQos());
|
log.info("[subscribe][数采 MQTT 订阅成功：{}]", topic);
|
} catch (Exception e) {
|
log.error("[subscribe][数采 MQTT 订阅失败：{}]", topic, e);
|
}
|
}
|
}
|
|
/**
|
* 解析需要订阅的主题。
|
*
|
* <p>显式配置 topics 优先；网关格式订阅整个前缀（gateway/#），
|
* 因为网关除 attributes 外还会发 connect / disconnect，后续也可能新增主题；
|
* 其余格式沿用 /YLCY/IMM/+/+ 两级通配。</p>
|
*/
|
private List<String> resolveSubscribeTopics(MesCollectProperties.Mqtt mqtt) {
|
if (StrUtil.isNotBlank(mqtt.getTopics())) {
|
return Arrays.stream(mqtt.getTopics().split(","))
|
.map(String::trim)
|
.filter(StrUtil::isNotBlank)
|
.toList();
|
}
|
String prefix = normalizeTopicPrefix(mqtt.getTopicPrefix());
|
return Collections.singletonList(mqtt.isGatewayFormat() ? prefix + "#" : prefix + "+/+");
|
}
|
|
private String normalizeTopicPrefix(String prefix) {
|
if (StrUtil.isBlank(prefix)) {
|
return "";
|
}
|
return prefix.endsWith("/") ? prefix : prefix + "/";
|
}
|
|
private String buildClientId() {
|
String host;
|
try {
|
host = InetAddress.getLocalHost().getHostName();
|
} catch (Exception e) {
|
host = "host";
|
}
|
host = host.replaceAll("[^a-zA-Z0-9_-]", "_");
|
String clientId = properties.getMqtt().getClientIdPrefix() + "-" + host + "-"
|
+ Long.toHexString(System.nanoTime());
|
return StrUtil.maxLength(clientId, 64);
|
}
|
|
@Override
|
public void stop() {
|
running = false;
|
if (scheduler != null) {
|
scheduler.shutdownNow();
|
scheduler = null;
|
}
|
if (client != null) {
|
try {
|
if (client.isConnected()) {
|
client.disconnect();
|
}
|
} catch (Exception e) {
|
log.warn("[stop][数采 MQTT 断开失败：{}]", e.getMessage());
|
}
|
try {
|
client.close(true);
|
} catch (Exception e) {
|
log.warn("[stop][数采 MQTT 关闭失败：{}]", e.getMessage());
|
}
|
client = null;
|
}
|
log.info("[stop][数采 MQTT 已停止]");
|
}
|
|
@Override
|
public boolean isRunning() {
|
return running;
|
}
|
|
}
|
