package com.dingzhuo.energy.project.conglomeratepush.mqtt;
|
|
import lombok.extern.slf4j.Slf4j;
|
import org.eclipse.paho.client.mqttv3.*;
|
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
|
import org.springframework.beans.factory.annotation.Value;
|
import org.springframework.stereotype.Component;
|
|
/**
|
* MQTT工具类操作
|
*
|
* @author zhw
|
* @since 2022/04/05
|
*/
|
@Slf4j
|
@Component
|
public class MQTTConnect {
|
|
@Value("${mqtt.host}")
|
private String HOST;
|
private final String clientId = "ClientHBT" + (int) (Math.random() * 100000000);
|
private MqttClient mqttClient;
|
|
/**
|
* 客户端connect连接mqtt服务器
|
*
|
* @param username 用户名
|
* @param password 密码
|
* @param mqttCallback 回调函数
|
**/
|
public void setMqttClient(String username, String password, MqttCallback mqttCallback)
|
throws MqttException {
|
System.out.println("MQTT服务器连接开始...........");
|
MqttConnectOptions options = mqttConnectOptions(username, password);
|
mqttClient.setCallback(mqttCallback);
|
mqttClient.connect(options);
|
}
|
|
/**
|
* MQTT连接参数设置
|
*/
|
private MqttConnectOptions mqttConnectOptions(String userName, String passWord)
|
throws MqttException {
|
mqttClient = new MqttClient(HOST, clientId, new MemoryPersistence());
|
MqttConnectOptions options = new MqttConnectOptions();
|
options.setUserName(userName);
|
options.setPassword(passWord.toCharArray());
|
options.setConnectionTimeout(10);///默认：30
|
options.setAutomaticReconnect(true);//默认：false
|
options.setCleanSession(false);//默认：true
|
//options.setKeepAliveInterval(20);//默认：60
|
return options;
|
}
|
|
/**
|
* 关闭MQTT连接
|
*/
|
public void close() throws MqttException {
|
try {
|
if(mqttClient!=null){
|
if(mqttClient.isConnected()){
|
mqttClient.unsubscribe(clientId);
|
}
|
mqttClient.disconnect();
|
mqttClient.close();
|
}
|
|
}catch (Exception e){
|
e.printStackTrace();
|
}
|
|
}
|
|
/**
|
* 向某个主题发布消息 默认qos：1
|
*/
|
public void pub(String topic, String msg) throws MqttException {
|
MqttMessage mqttMessage = new MqttMessage();
|
//mqttMessage.setQos(2);
|
mqttMessage.setPayload(msg.getBytes());
|
MqttTopic mqttTopic = mqttClient.getTopic(topic);
|
MqttDeliveryToken token = mqttTopic.publish(mqttMessage);
|
token.waitForCompletion();
|
}
|
|
/**
|
* 向某个主题发布消息
|
*
|
* @param topic: 发布的主题
|
* @param msg: 发布的消息
|
* @param qos: 消息质量 Qos：0、1、2
|
*/
|
public void pub(String topic, String msg, int qos) throws MqttException {
|
MqttMessage mqttMessage = new MqttMessage();
|
mqttMessage.setQos(qos);
|
mqttMessage.setPayload(msg.getBytes());
|
MqttTopic mqttTopic = mqttClient.getTopic(topic);
|
MqttDeliveryToken token = mqttTopic.publish(mqttMessage);
|
token.waitForCompletion();
|
}
|
|
/**
|
* 订阅某一个主题 ，此方法默认的的Qos等级为：1
|
*
|
* @param topic 主题
|
*/
|
public void sub(String topic) throws MqttException {
|
mqttClient.subscribe(topic);
|
}
|
|
/**
|
* 订阅某一个主题，可携带Qos
|
*
|
* @param topic 所要订阅的主题
|
* @param qos 消息质量：0、1、2
|
*/
|
public void sub(String topic, int qos) throws MqttException {
|
mqttClient.subscribe(topic, qos);
|
}
|
|
public static void main(String[] args) throws MqttException {
|
//测试连接及主动订阅信息 输出 通过
|
//要做推送 直接在计划任务中连接推送即可，做全局conn对象，然后 每次计划任务轮询发送前判断
|
//对象是否存在且 没有关闭 要进行处理
|
// MQTTConnect mqttConnect = new MQTTConnect();
|
// String msg = "Mr.Qu" + (int) (Math.random() * 100000000);
|
// mqttConnect.setMqttClient("dingzhuo", "dingzhuo@2022", new InitCallback());
|
// mqttConnect.sub("iot-2/evt/waconn/#",0);
|
// try {
|
// Thread.sleep(5000);
|
// } catch (InterruptedException e) {
|
// e.printStackTrace();
|
// }
|
// mqttConnect.close();
|
}
|
}
|
