mqtt 模块添加
This commit is contained in:
yuhaiming
43
water-common/water-common-mqtt/pom.xml
Normal file
43
water-common/water-common-mqtt/pom.xml
Normal file
@@ -0,0 +1,43 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
<parent>
|
||||
<groupId>org.dromara</groupId>
|
||||
<artifactId>water-common</artifactId>
|
||||
<version>5.6.0</version>
|
||||
</parent>
|
||||
|
||||
<artifactId>water-common-mqtt</artifactId>
|
||||
|
||||
<properties>
|
||||
<maven.compiler.source>17</maven.compiler.source>
|
||||
<maven.compiler.target>17</maven.compiler.target>
|
||||
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
|
||||
</properties>
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.eclipse.paho</groupId>
|
||||
<artifactId>org.eclipse.paho.client.mqttv3</artifactId>
|
||||
<version>1.2.5</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.projectlombok</groupId>
|
||||
<artifactId>lombok</artifactId>
|
||||
<scope>provided</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework</groupId>
|
||||
<artifactId>spring-context</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.dromara</groupId>
|
||||
<artifactId>water-common-json</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.dromara</groupId>
|
||||
<artifactId>water-app</artifactId>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
</project>
|
||||
@@ -0,0 +1,94 @@
|
||||
package org.dromara.mqtt;
|
||||
|
||||
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.persist.MemoryPersistence;
|
||||
import org.springframework.beans.factory.DisposableBean;
|
||||
import org.springframework.beans.factory.InitializingBean;
|
||||
import org.springframework.boot.autoconfigure.AutoConfiguration;
|
||||
import org.springframework.boot.context.properties.EnableConfigurationProperties;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
|
||||
import java.nio.charset.StandardCharsets;
|
||||
|
||||
@Slf4j
|
||||
@AutoConfiguration
|
||||
@EnableConfigurationProperties(MqttProperties.class)
|
||||
@RequiredArgsConstructor
|
||||
public class MqttClientManager implements InitializingBean, DisposableBean {
|
||||
|
||||
private final MqttProperties props;
|
||||
private final MqttMessageDispatcher dispatcher;
|
||||
private MqttClient client;
|
||||
|
||||
@Override
|
||||
public void afterPropertiesSet() throws Exception {
|
||||
// mqttConnect();
|
||||
}
|
||||
|
||||
@Bean
|
||||
private MqttClient mqttConnect() throws MqttException {
|
||||
client = new MqttClient(props.getBrokerUrl(), props.getClientId(), new MemoryPersistence());
|
||||
|
||||
MqttConnectOptions options = new MqttConnectOptions();
|
||||
options.setUserName(props.getUsername());
|
||||
options.setPassword(props.getPassword().toCharArray());
|
||||
options.setKeepAliveInterval(props.getKeepAlive());
|
||||
options.setConnectionTimeout(props.getConnectionTimeout());
|
||||
options.setAutomaticReconnect(true);
|
||||
options.setCleanSession(false);
|
||||
|
||||
client.setCallback(new MqttCallback() {
|
||||
@Override
|
||||
public void connectionLost(Throwable cause) {
|
||||
log.warn("[MQTT] 连接断开: {}", cause.getMessage());
|
||||
}
|
||||
|
||||
@Override
|
||||
public void messageArrived(String topic, MqttMessage message) {
|
||||
dispatcher.dispatch(topic, new String(message.getPayload()));
|
||||
}
|
||||
|
||||
@Override
|
||||
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()); // ← 加这行
|
||||
}
|
||||
|
||||
return client;
|
||||
}
|
||||
|
||||
/**
|
||||
* 发布消息
|
||||
*/
|
||||
public void publish(String topic, String payload) {
|
||||
try {
|
||||
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);
|
||||
} catch (MqttException e) {
|
||||
log.error("[MQTT] 发布失败 topic={}", topic, e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void destroy() throws Exception {
|
||||
if (client != null && client.isConnected()) {
|
||||
client.disconnect();
|
||||
client.close();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,29 @@
|
||||
package org.dromara.mqtt.config.properties;
|
||||
|
||||
|
||||
import lombok.Data;
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
@Data
|
||||
//@Component
|
||||
@ConfigurationProperties(prefix = "mqtt")
|
||||
public class MqttProperties {
|
||||
|
||||
private String brokerUrl;
|
||||
private String username;
|
||||
private String password;
|
||||
private String clientId;
|
||||
private int qos = 1;
|
||||
private int keepAlive = 60;
|
||||
private int connectionTimeout = 30;
|
||||
private Topics topics;
|
||||
|
||||
@Data
|
||||
public static class Topics {
|
||||
private List<String> subscribe;
|
||||
private String publishPrefix;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,2 @@
|
||||
org.dromara.mqtt.MqttClientManager
|
||||
org.dromara.app.mqtt.MqttMessageDispatcher
|
||||
Reference in New Issue
Block a user