|
|
|
|
@ -0,0 +1,152 @@
|
|
|
|
|
package jnpf.utils;
|
|
|
|
|
|
|
|
|
|
import lombok.extern.slf4j.Slf4j;
|
|
|
|
|
import org.eclipse.paho.mqttv5.client.MqttClient;
|
|
|
|
|
import org.eclipse.paho.mqttv5.client.MqttConnectionOptions;
|
|
|
|
|
import org.eclipse.paho.mqttv5.client.MqttTopic;
|
|
|
|
|
import org.eclipse.paho.mqttv5.client.persist.MemoryPersistence;
|
|
|
|
|
import org.eclipse.paho.mqttv5.common.MqttException;
|
|
|
|
|
import org.eclipse.paho.mqttv5.common.MqttMessage;
|
|
|
|
|
import org.springframework.beans.factory.annotation.Value;
|
|
|
|
|
import org.springframework.stereotype.Component;
|
|
|
|
|
|
|
|
|
|
import javax.annotation.PostConstruct;
|
|
|
|
|
import javax.annotation.Resource;
|
|
|
|
|
import java.nio.charset.StandardCharsets;
|
|
|
|
|
import java.util.Map;
|
|
|
|
|
|
|
|
|
|
@Slf4j
|
|
|
|
|
@Component
|
|
|
|
|
public class CustomerMqttClient {
|
|
|
|
|
@Resource
|
|
|
|
|
private MyMqttCallback callback;
|
|
|
|
|
private MqttClient client;
|
|
|
|
|
|
|
|
|
|
@Value("${mqtt.client.host}")
|
|
|
|
|
private String host;
|
|
|
|
|
@Value("${mqtt.client.clientId}")
|
|
|
|
|
private String clientId;
|
|
|
|
|
@Value("${mqtt.client.username}")
|
|
|
|
|
private String userName;
|
|
|
|
|
@Value("${mqtt.client.password}")
|
|
|
|
|
private String password;
|
|
|
|
|
@Value("${mqtt.client.cleanSession}")
|
|
|
|
|
private boolean cleanSession;
|
|
|
|
|
@Value("${mqtt.client.timeout}")
|
|
|
|
|
private int timeout;
|
|
|
|
|
@Value("${mqtt.client.qos}")
|
|
|
|
|
private int qos;
|
|
|
|
|
@Value("${mqtt.client.keepAlive}")
|
|
|
|
|
private int keepAlive;
|
|
|
|
|
// @Value("${mqtt.client.topics}")
|
|
|
|
|
private Map<String, Integer> topics;
|
|
|
|
|
|
|
|
|
|
@PostConstruct
|
|
|
|
|
public void init() {
|
|
|
|
|
try {
|
|
|
|
|
connect();
|
|
|
|
|
} catch (MqttException e) {
|
|
|
|
|
log.error("连接MQTT服务失败", e);
|
|
|
|
|
throw new RuntimeException("连接MQTT服务失败");
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
public void connect() throws MqttException {
|
|
|
|
|
client = new MqttClient(host, clientId, new MemoryPersistence());
|
|
|
|
|
MqttConnectionOptions options = new MqttConnectionOptions();
|
|
|
|
|
options.setUserName(userName);
|
|
|
|
|
options.setPassword(password.getBytes(StandardCharsets.UTF_8));
|
|
|
|
|
options.setCleanStart(cleanSession);
|
|
|
|
|
options.setKeepAliveInterval(keepAlive);
|
|
|
|
|
options.setConnectionTimeout(timeout);
|
|
|
|
|
options.setAutomaticReconnect(true);
|
|
|
|
|
options.setAutomaticReconnectDelay(1, 3);
|
|
|
|
|
options.setSessionExpiryInterval(0L);//会话过期时间
|
|
|
|
|
client.setCallback(callback);
|
|
|
|
|
if (!client.isConnected()) {
|
|
|
|
|
client.connect(options);
|
|
|
|
|
log.info("连接MQTT服务成功 success");
|
|
|
|
|
} else {
|
|
|
|
|
client.disconnect();
|
|
|
|
|
client.connect(options);
|
|
|
|
|
log.info("重新连接MQTT服务成功 success");
|
|
|
|
|
}
|
|
|
|
|
callback.registerClient(client);
|
|
|
|
|
// initSubscribe();
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
public void disconnect() throws MqttException {
|
|
|
|
|
client.disconnect();
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
public boolean isConnected() {
|
|
|
|
|
return client.isConnected();
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
public void publish(String topic, String msg) {
|
|
|
|
|
publish(topic, msg, qos, false);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
public void publish(String topic, String msg, int qos, boolean retained) {
|
|
|
|
|
MqttMessage message = new MqttMessage();
|
|
|
|
|
message.setPayload(msg.getBytes());
|
|
|
|
|
message.setQos(qos);
|
|
|
|
|
message.setRetained(retained);
|
|
|
|
|
MqttTopic mqttTopic = client.getTopic(topic);
|
|
|
|
|
if (null == mqttTopic) {
|
|
|
|
|
log.error("发布失败，主题不存在！！topic: {} , qos: {} ", topic, qos);
|
|
|
|
|
}
|
|
|
|
|
try {
|
|
|
|
|
client.publish(topic, message);
|
|
|
|
|
log.info("发布成功 topic: {} , qos: {} , message: {}", topic, qos, msg);
|
|
|
|
|
} catch (MqttException e) {
|
|
|
|
|
log.info("发布失败 topic: {} , qos: {} , message: {}", topic, qos, msg);
|
|
|
|
|
e.printStackTrace();
|
|
|
|
|
throw new RuntimeException("mqtt 发布主题失败，发布主题：" + topic + "，发布消息：" + msg);
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
public void subscribe(String topicFilter, int qos) {
|
|
|
|
|
if (client != null && client.isConnected()) {
|
|
|
|
|
try {
|
|
|
|
|
client.subscribe(topicFilter, qos);
|
|
|
|
|
log.info("订阅主题成功,topic: {} , qos: {} ", topicFilter, qos);
|
|
|
|
|
} catch (MqttException e) {
|
|
|
|
|
log.error("订阅主题失败 topic:{}, qos: {} ", topicFilter, qos);
|
|
|
|
|
e.printStackTrace();
|
|
|
|
|
}
|
|
|
|
|
} else {
|
|
|
|
|
log.error("客户端连接状态已断开,订阅主题失败 topic:{}, qos: {} ", topicFilter, qos);
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
public void unsubscribe(String topic) {
|
|
|
|
|
if (client != null && client.isConnected()) {
|
|
|
|
|
try {
|
|
|
|
|
client.unsubscribe(topic);
|
|
|
|
|
log.info("取消订阅成功 topic success:{}", topic);
|
|
|
|
|
} catch (MqttException e) {
|
|
|
|
|
log.error("取消订阅失败 topic:{}", topic);
|
|
|
|
|
e.printStackTrace();
|
|
|
|
|
}
|
|
|
|
|
} else {
|
|
|
|
|
log.error("客户端连接状态已断开,取消订阅失败 topic:{}", topic);
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
public void initSubscribe() {
|
|
|
|
|
if (client != null && client.isConnected()) {
|
|
|
|
|
if (topics == null || topics.size() == 0) {
|
|
|
|
|
log.info("未配置初始化订阅主题!!!");
|
|
|
|
|
}
|
|
|
|
|
for (Map.Entry<String, Integer> entry : topics.entrySet()) {
|
|
|
|
|
String topic = entry.getKey();
|
|
|
|
|
int qos = entry.getValue();
|
|
|
|
|
log.info("初始化订阅主题 topic: {} , qos: {} ", topic, qos);
|
|
|
|
|
subscribe(topic, qos);
|
|
|
|
|
}
|
|
|
|
|
} else {
|
|
|
|
|
log.error("客户端未连接,初始化订阅失败!!!");
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
guochaojie
4 months ago