MQTT Java 客户端

MQTT Java 客户端库

Eclipse Paho Java Client 是用 Java 编写的 MQTT 客户端库（MQTT Java Client），可用于 JVM 或其他 Java 兼容平台（例如Android）。

Eclipse Paho Java Client 提供了MqttAsyncClient 和 MqttClient 异步和同步 API。

通过 Maven 安装 Paho Java

通过包管理工具 Maven 可以方便地安装 Paho Java 客户端库，截止目前最新版本安装如下：

 
    
    
      org.eclipse.paho 
     
    
      org.eclipse.paho.client.mqttv3 
     
    
      1.2.2 
     
    

Paho Java 使用示例

 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; public class App { public static void main(String[] args) { String subTopic = "testtopic/#"; String pubTopic = "testtopic/1"; String content = "Hello World"; int qos = 2; String broker = "tcp://127.0.0.1:1883"; String clientId = "mqtt_test"; MemoryPersistence persistence = new MemoryPersistence(); try { MqttClient client = new MqttClient(broker, clientId, persistence); // MQTT 连接选项 MqttConnectOptions connOpts = new MqttConnectOptions(); // connOpts.setUserName("mqtt_test"); // connOpts.setPassword("mqtt_test_password".toCharArray()); // 保留会话 connOpts.setCleanSession(true); // 设置回调 //client.setCallback(new PushCallback()); client.setCallback(new OnMessageCallback()); // 建立连接 System.out.println("Connecting to broker: " + broker); client.connect(connOpts); System.out.println("Connected"); System.out.println("Publishing message: " + content); // 订阅 client.subscribe(subTopic); // 消息发布所需参数 MqttMessage message = new MqttMessage(content.getBytes()); message.setQos(qos); client.publish(pubTopic, message); System.out.println("Message published"); /* client.disconnect(); System.out.println("Disconnected"); client.close(); System.exit(0);*/ } catch (MqttException me) { System.out.println("reason " + me.getReasonCode()); System.out.println("msg " + me.getMessage()); System.out.println("loc " + me.getLocalizedMessage()); System.out.println("cause " + me.getCause()); System.out.println("excep " + me); me.printStackTrace(); } } } 

回调消息处理类 OnMessageCallback.java

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken; import org.eclipse.paho.client.mqttv3.MqttCallback; import org.eclipse.paho.client.mqttv3.MqttMessage; public class OnMessageCallback implements MqttCallback { public void connectionLost(Throwable cause) { // 连接丢失后，一般在这里面进行重连 System.out.println("连接断开，可以做重连"); } public void messageArrived(String topic, MqttMessage message) throws Exception { // subscribe后得到的消息会执行到这里面 System.out.println("接收消息主题:" + topic); System.out.println("接收消息Qos:" + message.getQos()); System.out.println("接收消息内容:" + new String(message.getPayload())); } public void deliveryComplete(IMqttDeliveryToken token) { System.out.println("deliveryComplete---------" + token.isComplete()); } } 

Paho Java MQTT 5.0 支持

目前 Paho Java 还在适配 MQTT 5.0，尚未全面支持。