MQTT工具类

**MQTT工具类**

MQTT（Message Queuing Telemetry Transport）是一种轻量级的机器间通信协议，广泛应用于物联网、工业自动化等领域。下面是关于MQTT的一些基本概念和工具类实现。

### MQTT基本概念* **Broker**: MQTT中的消息代理服务器，负责接收、存储和转发消息。
* **Client**: MQTT客户端，连接到broker并发送或接收消息。
* **Topic**: MQTT中用于发布订阅的主题名称。
* **QoS**: MQTT中用于保证消息传递可靠性的服务质量等级，有三种：0、1和2。

### MQTT工具类实现下面是基于Java语言的一个简单的MQTT工具类实现：

javaimport org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;

public class MqttUtil {

 private static final String BROKER_URL = "tcp://localhost:1883";
 private static final String CLIENT_ID = "client1";

 public void connect() {
 try {
 MqttClient client = new MqttClient(BROKER_URL, CLIENT_ID);
 MqttConnectOptions options = new MqttConnectOptions();
 client.connect(options);
 System.out.println("Connected to MQTT broker");
 } catch (MqttException e) {
 System.err.println("Failed to connect to MQTT broker: " + e.getMessage());
 }
 }

 public void publish(String topic, String message) {
 try {
 MqttClient client = new MqttClient(BROKER_URL, CLIENT_ID);
 client.connect();
 client.publish(topic, message.getBytes());
 System.out.println("Published message to topic: " + topic);
 } catch (MqttException e) {
 System.err.println("Failed to publish message: " + e.getMessage());
 }
 }

 public void subscribe(String topic) {
 try {
 MqttClient client = new MqttClient(BROKER_URL, CLIENT_ID);
 client.connect();
 client.subscribe(topic);
 System.out.println("Subscribed to topic: " + topic);
 } catch (MqttException e) {
 System.err.println("Failed to subscribe to topic: " + e.getMessage());
 }
 }

 public void unsubscribe(String topic) {
 try {
 MqttClient client = new MqttClient(BROKER_URL, CLIENT_ID);
 client.connect();
 client.unsubscribe(topic);
 System.out.println("Unsubscribed from topic: " + topic);
 } catch (MqttException e) {
 System.err.println("Failed to unsubscribe from topic: " + e.getMessage());
 }
 }

 public void disconnect() {
 try {
 MqttClient client = new MqttClient(BROKER_URL, CLIENT_ID);
 client.disconnect();
 System.out.println("Disconnected from MQTT broker");
 } catch (MqttException e) {
 System.err.println("Failed to disconnect from MQTT broker: " + e.getMessage());
 }
 }

 public void setCallback(MqttCallback callback) {
 try {
 MqttClient client = new MqttClient(BROKER_URL, CLIENT_ID);
 client.connect();
 client.setCallback(callback);
 System.out.println("Set MQTT callback");
 } catch (MqttException e) {
 System.err.println("Failed to set MQTT callback: " + e.getMessage());
 }
 }

}

javapublic class Main {

 public static void main(String[] args) {
 MqttUtil mqttUtil = new MqttUtil();
 mqttUtil.connect();

 // 发布消息 mqttUtil.publish("test/topic", "Hello, MQTT!");

 // 订阅主题 mqttUtil.subscribe("test/topic");

 // 接收消息 mqttUtil.setCallback(new MqttCallback() {
 @Override public void connectionLost(IMqttToken imqttToken) {

 }

 @Override public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
 System.out.println("Received message: " + new String(mqttMessage.getPayload()));
 }

 @Override public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {

 }
 });

 // 等待接收消息 Thread.sleep(1000);

 mqttUtil.unsubscribe("test/topic");

 mqttUtil.disconnect();
 }
}