+ * MySQL 为主数据源（@Primary，供 MyBatis-Plus 使用）， + * TDengine 为时序数据源，通过专属 {@code tdengineJdbcTemplate} 访问设备心跳数据。 + *
+ * 注意：Spring Boot 4.x 下 Druid 1.2.24 的 spring-boot-starter 存在兼容性问题 + * （依赖已移除的 DataSourceProperties 类），此处直接使用 DruidDataSource 手动配置。 + * + * @author TongTongStudio + */ +@Configuration +public class DataSourceConfig { + + // 1. MySQL 主数据源 (Druid) + @Primary + @Bean(name = "mysqlDataSource") + @ConfigurationProperties("spring.datasource.mysql") + public DataSource mysqlDataSource() { + return new DruidDataSource(); + } + + // 2. TDengine 时序数据源 (Druid 连接池 + Restful 驱动) + @Bean(name = "tdengineDataSource") + @ConfigurationProperties("spring.datasource.tdengine") + public DataSource tdengineDataSource() { + return new DruidDataSource(); + } + + // 3. 用于操作 MySQL 的 JdbcTemplate (可选) + @Primary + @Bean(name = "mysqlJdbcTemplate") + public JdbcTemplate mysqlJdbcTemplate(@Qualifier("mysqlDataSource") DataSource dataSource) { + return new JdbcTemplate(dataSource); + } + + // 4. 用于操作 TDengine 的 JdbcTemplate + @Bean(name = "tdengineJdbcTemplate") + public JdbcTemplate tdengineJdbcTemplate(@Qualifier("tdengineDataSource") DataSource dataSource) { + return new JdbcTemplate(dataSource); + } +} diff --git a/src/main/java/com/youlai/boot/config/MqttConfig.java b/src/main/java/com/youlai/boot/config/MqttConfig.java index 826941ee..3be1621a 100644 --- a/src/main/java/com/youlai/boot/config/MqttConfig.java +++ b/src/main/java/com/youlai/boot/config/MqttConfig.java @@ -1,53 +1,145 @@ package com.youlai.boot.config; import com.youlai.boot.config.property.MqttProperties; -import org.eclipse.paho.client.mqttv3.MqttConnectOptions; +import com.youlai.boot.support.mqtt.MqttMessageHandler; +import lombok.extern.slf4j.Slf4j; +import org.eclipse.paho.mqttv5.client.*; +import org.eclipse.paho.mqttv5.common.MqttException; +import org.eclipse.paho.mqttv5.common.MqttMessage; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; -import org.springframework.integration.annotation.ServiceActivator; -import org.springframework.integration.channel.DirectChannel; -import org.springframework.integration.core.MessageProducer; -import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory; -import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter; -import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler; -import org.springframework.messaging.MessageChannel; -import org.springframework.messaging.MessageHandler; + +import jakarta.annotation.PreDestroy; +import java.nio.charset.StandardCharsets; +import java.util.concurrent.TimeUnit; /** * MQTT 配置 *
- * 基于 spring-integration-mqtt (Eclipse Paho) 实现消息的发布与订阅。 + * 基于 Eclipse Paho MQTT v5 原生客户端 (org.eclipse.paho.mqttv5.client) 实现消息的发布与订阅。 * 通过 {@code mqtt.enabled=false} 可整体关闭 MQTT 功能。 * * @author TongTongStudio * @since 2026/8/8 */ +@Slf4j @Configuration @ConditionalOnProperty(prefix = "mqtt", name = "enabled", havingValue = "true") public class MqttConfig { /** - * MQTT 连接工厂 + * MQTT 客户端实例（保存引用，供 @PreDestroy 优雅关闭） */ - @Bean - public DefaultMqttPahoClientFactory mqttClientFactory(MqttProperties properties) { - MqttConnectOptions options = new MqttConnectOptions(); - options.setServerURIs(new String[]{normalizeServerUri(properties.getUrl())}); + private IMqttClient client; + + /** + * MQTT 客户端 + *
同一客户端同时用于发布与订阅。连接成功后立即订阅配置的主题。
+ */ + @Bean(destroyMethod = "disconnect") + public IMqttClient mqttClient(MqttProperties properties, MqttMessageHandler mqttMessageHandler) throws MqttException { + String serverUri = normalizeServerUri(properties.getUrl()); + String clientId = properties.getClientId(); + + client = new MqttClient(serverUri, clientId); + + MqttConnectionOptionsBuilder builder = new MqttConnectionOptionsBuilder() + .serverURI(serverUri) + .connectionTimeout(properties.getConnectionTimeout()) + .keepAliveInterval(properties.getKeepAliveInterval()) + .automaticReconnect(properties.isAutomaticReconnect()) + .sessionExpiryInterval(TimeUnit.DAYS.toSeconds(1)) + .cleanStart(false); + if (properties.getUsername() != null && !properties.getUsername().isEmpty()) { - options.setUserName(properties.getUsername()); + builder.username(properties.getUsername()); } if (properties.getPassword() != null && !properties.getPassword().isEmpty()) { - options.setPassword(properties.getPassword().toCharArray()); + builder.password(properties.getPassword().getBytes(StandardCharsets.UTF_8)); } - options.setConnectionTimeout(properties.getConnectionTimeout()); - options.setKeepAliveInterval(properties.getKeepAliveInterval()); - options.setAutomaticReconnect(properties.isAutomaticReconnect()); - options.setCleanSession(true); - DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory(); - factory.setConnectionOptions(options); - return factory; + MqttConnectionOptions options = builder.build(); + + // 消息到达回调：分发到业务处理器 + client.setCallback(new org.eclipse.paho.mqttv5.client.MqttCallback() { + @Override + public void messageArrived(String topic, MqttMessage message) { + String payload = new String(message.getPayload(), StandardCharsets.UTF_8); + mqttMessageHandler.handleMessage(topic, payload); + } + + @Override + public void deliveryComplete(IMqttToken iMqttToken) { + + } + + @Override + public void disconnected(org.eclipse.paho.mqttv5.client.MqttDisconnectResponse disconnectResponse) { + log.warn("MQTT 连接断开: reasonCode={}, reasonString={}", + disconnectResponse.getReturnCode(), disconnectResponse.getReasonString()); + } + + @Override + public void mqttErrorOccurred(MqttException exception) { + log.error("MQTT 发生异常", exception); + } + + @Override + public void connectComplete(boolean reconnect, String serverURI) { + log.info("MQTT 连接成功, reconnect={}, serverURI={}", reconnect, serverURI); + // 连接/重连成功后订阅主题 + subscribeTopics(client, properties); + } + + @Override + public void authPacketArrived(int i, org.eclipse.paho.mqttv5.common.packet.MqttProperties mqttProperties) { + + } + }); + + client.connect(options); + log.info("MQTT 客户端已启动, serverURI={}, clientId={}", serverUri, clientId); + + // 首次连接成功后订阅主题（重连场景由 connectComplete 处理） + if (client.isConnected()) { + subscribeTopics(client, properties); + } + + return client; + } + + /** + * 应用关闭时优雅停止 MQTT 客户端。 + *先 {@code disconnect} 断开连接，再 {@code close} 释放 Paho 后台线程池， + * 避免容器已销毁（Redisson 已 shutdown）后回调线程仍在处理消息而报错。
+ */ + @PreDestroy + public void destroy() { + try { + if (client != null) { + if (client.isConnected()) { + client.disconnect(); + } + client.close(); + } + } catch (MqttException e) { + log.warn("关闭 MQTT 客户端失败", e); + } + } + + /** + * 订阅配置中的所有主题 + */ + private void subscribeTopics(IMqttClient client, MqttProperties properties) { + for (String topic : properties.getTopics()) { + try { + client.subscribe(topic, properties.getQos()); + log.info("已订阅 MQTT 主题: {}", topic); + } catch (MqttException e) { + log.error("订阅 MQTT 主题失败: {}", topic, e); + } + } } /** @@ -67,51 +159,4 @@ public class MqttConfig { } return uri; } - - /* ============================ 发布 (Publish) ============================ */ - - /** - * 发布消息的出站通道 - */ - @Bean - public MessageChannel mqttOutboundChannel() { - return new DirectChannel(); - } - - /** - * 发布消息处理器，通过 {@link org.springframework.messaging.support.MessageBuilder} 发送到出站通道即可发布 - */ - @Bean - @ServiceActivator(inputChannel = "mqttOutboundChannel") - public MessageHandler mqttOutbound(DefaultMqttPahoClientFactory factory, MqttProperties properties) { - MqttPahoMessageHandler handler = new MqttPahoMessageHandler(properties.getClientId() + "-pub", factory); - handler.setAsync(false); - handler.setDefaultQos(properties.getQos()); - return handler; - } - - /* ============================ 订阅 (Subscribe) ============================ */ - - /** - * 订阅消息的入站通道 - */ - @Bean - public MessageChannel mqttInboundChannel() { - return new DirectChannel(); - } - - /** - * 入站订阅适配器：订阅 {@link MqttProperties#getTopics()} 中配置的主题 - */ - @Bean - public MessageProducer mqttInbound(DefaultMqttPahoClientFactory factory, MqttProperties properties) { - String[] topics = properties.getTopics().toArray(new String[0]); - MqttPahoMessageDrivenChannelAdapter adapter = - new MqttPahoMessageDrivenChannelAdapter( - properties.getClientId() + "-sub", factory, topics); - adapter.setCompletionTimeout(5000L); - adapter.setQos(properties.getQos()); - adapter.setOutputChannel(mqttInboundChannel()); - return adapter; - } } diff --git a/src/main/java/com/youlai/boot/device/controller/DeviceController.java b/src/main/java/com/youlai/boot/device/controller/DeviceController.java index 69cef2f2..cb545bd4 100644 --- a/src/main/java/com/youlai/boot/device/controller/DeviceController.java +++ b/src/main/java/com/youlai/boot/device/controller/DeviceController.java @@ -30,6 +30,7 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.PathVariable; import org.springframework.web.bind.annotation.PostMapping; +import org.springframework.web.bind.annotation.PutMapping; import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; @@ -82,6 +83,18 @@ public class DeviceController { return Result.success(); } + @Operation(summary = "修改设备（仅允许修改设备昵称）") + @PutMapping("/{id}") + public Result- * 消费 {@code mqttInboundChannel} 通道中的消息，根据主题分发处理。 + * 由 {@link com.youlai.boot.config.MqttConfig} 中 v5 客户端的 + * {@code messageArrived} 回调调用，根据主题分发处理消息。 * * @author TongTongStudio * @since 2026/8/8 */ @Slf4j @Component -@ConditionalOnProperty(prefix = "mqtt", name = "enabled", havingValue = "true") -public class MqttMessageHandler implements MessageHandler { +public class MqttMessageHandler { private final DeviceOnlineService deviceOnlineService; private final MqttProperties mqttProperties; @@ -45,16 +37,10 @@ public class MqttMessageHandler implements MessageHandler { /** * 处理从 MQTT Broker 订阅到的消息 * - * @param message MQTT 入站消息，可通过 {@link MqttHeaders#RECEIVED_TOPIC} 获取主题 + * @param topic 消息主题 + * @param payload 消息内容（UTF-8 字符串） */ - @ServiceActivator(inputChannel = "mqttInboundChannel") - @Override - public void handleMessage(Message message) throws MessagingException { - String topic = message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC, String.class); - String payload = message.getPayload() instanceof byte[] - ? new String((byte[]) message.getPayload(), StandardCharsets.UTF_8) - : String.valueOf(message.getPayload()); - + public void handleMessage(String topic, String payload) { log.info("收到 MQTT 消息, topic={}, payload={}", topic, payload); // 根据业务需要，按主题分发处理 @@ -69,7 +55,7 @@ public class MqttMessageHandler implements MessageHandler { /** * 处理设备状态上报（心跳/屏幕状态），并同步更新 Redis 中的在线状态 *
- * 消息体为 JSON，格式：{"sn":"xxx", "status":"online|offline", "ip":"1.2.3.4", "screenOn":true|false} + * 消息体为 JSON，格式：{"sn":"xxx", "online":"true online|false offline", "ip":"1.2.3.4", "screenOn":true|false} * 其中 screenOn 为可选字段，表示设备亮屏(亮屏=屏幕点亮)状态。 */ private void handleDeviceStatus(String payload) { @@ -84,7 +70,7 @@ public class MqttMessageHandler implements MessageHandler { log.warn("设备状态上报缺少 sn 字段, payload={}", payload); return; } - String status = json.getStr("status"); + boolean online = json.getBool("online"); String ip = json.getStr("ip"); // 解析屏幕状态（可选）：true=亮屏, false=熄屏, 缺省=null（未知） @@ -93,7 +79,7 @@ public class MqttMessageHandler implements MessageHandler { screenState = json.getBool("screenOn") ? 1 : 0; } - if ("offline".equalsIgnoreCase(status)) { + if (!online) { // 设备主动上报离线（如关机/退出），立即标记离线 deviceOnlineService.markOffline(sn); } else { diff --git a/src/main/java/com/youlai/boot/support/mqtt/MqttProducer.java b/src/main/java/com/youlai/boot/support/mqtt/MqttProducer.java index 129fbef3..b1bcef38 100644 --- a/src/main/java/com/youlai/boot/support/mqtt/MqttProducer.java +++ b/src/main/java/com/youlai/boot/support/mqtt/MqttProducer.java @@ -1,19 +1,20 @@ package com.youlai.boot.support.mqtt; +import com.youlai.boot.config.property.MqttProperties; import lombok.extern.slf4j.Slf4j; +import org.eclipse.paho.mqttv5.client.IMqttClient; +import org.eclipse.paho.mqttv5.common.MqttException; +import org.eclipse.paho.mqttv5.common.MqttMessage; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; -import org.springframework.integration.core.MessagingTemplate; -import org.springframework.integration.mqtt.support.MqttHeaders; -import org.springframework.messaging.Message; -import org.springframework.messaging.MessageChannel; -import org.springframework.messaging.support.MessageBuilder; import org.springframework.stereotype.Component; +import java.nio.charset.StandardCharsets; + /** * MQTT 消息发布服务 *
- * 提供向指定主题发布消息的能力，消息经过 {@code mqttOutboundChannel} 通道发送到 MQTT Broker。
+ * 基于 Eclipse Paho MQTT v5 原生客户端发布消息，直接通过 {@link IMqttClient} 发送到 Broker。
* 当 {@code mqtt.enabled=false} 时该 Bean 不创建。
*
* @author TongTongStudio
@@ -24,15 +25,17 @@ import org.springframework.stereotype.Component;
@ConditionalOnProperty(prefix = "mqtt", name = "enabled", havingValue = "true")
public class MqttProducer {
- private final MessagingTemplate messagingTemplate;
+ private final IMqttClient mqttClient;
+ private final MqttProperties mqttProperties;
@Autowired
- public MqttProducer(MessageChannel mqttOutboundChannel) {
- this.messagingTemplate = new MessagingTemplate(mqttOutboundChannel);
+ public MqttProducer(IMqttClient mqttClient, MqttProperties mqttProperties) {
+ this.mqttClient = mqttClient;
+ this.mqttProperties = mqttProperties;
}
/**
- * 发布消息
+ * 发布消息（使用默认 QoS）
*
* @param topic 主题
* @param payload 消息内容(字符串)
@@ -49,14 +52,19 @@ public class MqttProducer {
* @param qos QoS 级别，为 null 时使用默认 QoS
*/
public void publish(String topic, String payload, Integer qos) {
- MessageBuilder 预热失败仅记录告警，不阻断应用启动（避免数据库临时不可用时整站起不来）。