Java开发MQTT实例代码分享

<project...>
<repositories>
    <repository>
        <id>Eclipse Paho Repo</id>
        <url>%REPOURL%</url>
    </repository>
</repositories>
...
<dependencies>
    <dependency>
        <groupId>org.eclipse.paho</groupId>
        <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
        <version>%VERSION%</version>
    </dependency>
</dependencies>
</project>

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

publicclassMqttPublisher {

publicstaticvoidmain(String[] args) {
// 连接参数
Stringbroker="tcp://broker.hivemq.com:1883"; // 公共MQTT代理
StringclientId="JavaPublisher";
Stringtopic="java/mqtt/test";
Stringcontent="Hello from Java MQTT Publisher!";
intqos=2; // 服务质量等级 (0,1,2)

try {
// 1. 创建客户端
MqttClientclient=newMqttClient(broker, clientId, newMemoryPersistence());

// 2. 配置连接选项
MqttConnectOptionsoptions=newMqttConnectOptions();
            options.setCleanSession(true);
            options.setConnectionTimeout(10);
            options.setKeepAliveInterval(20);
// 如需认证可添加:
// options.setUserName("username");
// options.setPassword("password".toCharArray());

// 3. 连接代理
            System.out.println("Connecting to broker: " + broker);
            client.connect(options);
            System.out.println("Connected!");

// 4. 创建消息并设置QoS
MqttMessagemessage=newMqttMessage(content.getBytes());
            message.setQos(qos);
            message.setRetained(true); // 设置保留消息

// 5. 发布消息
            client.publish(topic, message);
            System.out.println("Message published: " + content);

// 6. 断开连接
            client.disconnect();
            System.out.println("Disconnected");

// 7. 关闭客户端
            client.close();

        } catch (MqttException e) {
            e.printStackTrace();
        }
    }
}

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

publicclassMqttSubscriberimplementsMqttCallback {

publicstaticvoidmain(String[] args) {
String broker = "tcp://broker.hivemq.com:1883";
String clientId = "JavaSubscriber";
String topic = "java/mqtt/test";
        int qos = 2;

try {
// 1. 创建客户端
MqttClient client = newMqttClient(broker, clientId, newMemoryPersistence());

// 2. 配置回调
            client.setCallback(newMqttSubscriber());

// 3. 连接选项
MqttConnectOptions options = newMqttConnectOptions();
            options.setCleanSession(true);

// 4. 连接代理
System.out.println("Connecting to broker: " + broker);
            client.connect(options);
System.out.println("Connected!");

// 5. 订阅主题
            client.subscribe(topic, qos);
System.out.println("Subscribed to topic: " + topic);
System.out.println("Waiting for messages...");

// 保持程序运行以接收消息
Thread.sleep(Long.MAX_VALUE);

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

// 连接丢失回调
@Override
publicvoidconnectionLost(Throwable cause) {
System.out.println("Connection lost! " + cause.getMessage());
    }

// 消息到达回调
@Override
publicvoidmessageArrived(String topic, MqttMessage message) {
System.out.println("\nReceived message:" +
"\n\tTopic: " + topic +
"\n\tQoS: " + message.getQos() +
"\n\tContent: " + newString(message.getPayload()));
    }

// 消息传递完成回调
@Override
publicvoiddeliveryComplete(IMqttDeliveryToken token) {
// 发布者使用的方法，订阅者可忽略
    }
}

options.setUserName("admin");
options.setPassword("password123".toCharArray());