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 客户端库,截止目前最新版本安装如下:

  1. <dependency>
  2. <groupId>org.eclipse.paho</groupId>
  3. <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
  4. <version>1.2.2</version>
  5. </dependency>

Paho Java 使用示例

Java 体系中 Paho Java 是比较稳定、广泛应用的 MQTT 客户端库,本示例包含 Java 语言的 Paho Java 连接 EMQ X Broker,并进行消息收发完整代码:

App.java

  1. package io.emqx;
  2. import org.eclipse.paho.client.mqttv3.MqttClient;
  3. import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
  4. import org.eclipse.paho.client.mqttv3.MqttException;
  5. import org.eclipse.paho.client.mqttv3.MqttMessage;
  6. import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
  7. public class App {
  8. public static void main(String[] args) {
  9. String subTopic = "testtopic/#";
  10. String pubTopic = "testtopic/1";
  11. String content = "Hello World";
  12. int qos = 2;
  13. String broker = "tcp://broker.emqx.io:1883";
  14. String clientId = "emqx_test";
  15. MemoryPersistence persistence = new MemoryPersistence();
  16. try {
  17. MqttClient client = new MqttClient(broker, clientId, persistence);
  18. // MQTT 连接选项
  19. MqttConnectOptions connOpts = new MqttConnectOptions();
  20. connOpts.setUserName("emqx_test");
  21. connOpts.setPassword("emqx_test_password".toCharArray());
  22. // 保留会话
  23. connOpts.setCleanSession(true);
  24. // 设置回调
  25. client.setCallback(new PushCallback());
  26. // 建立连接
  27. System.out.println("Connecting to broker: " + broker);
  28. client.connect(connOpts);
  29. System.out.println("Connected");
  30. System.out.println("Publishing message: " + content);
  31. // 订阅
  32. client.subscribe(subTopic);
  33. // 消息发布所需参数
  34. MqttMessage message = new MqttMessage(content.getBytes());
  35. message.setQos(qos);
  36. client.publish(pubTopic, message);
  37. System.out.println("Message published");
  38. client.disconnect();
  39. System.out.println("Disconnected");
  40. client.close();
  41. System.exit(0);
  42. } catch (MqttException me) {
  43. System.out.println("reason " + me.getReasonCode());
  44. System.out.println("msg " + me.getMessage());
  45. System.out.println("loc " + me.getLocalizedMessage());
  46. System.out.println("cause " + me.getCause());
  47. System.out.println("excep " + me);
  48. me.printStackTrace();
  49. }
  50. }
  51. }

回调消息处理类 OnMessageCallback.java

  1. package io.emqx;
  2. import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
  3. import org.eclipse.paho.client.mqttv3.MqttCallback;
  4. import org.eclipse.paho.client.mqttv3.MqttMessage;
  5. public class OnMessageCallback implements MqttCallback {
  6. public void connectionLost(Throwable cause) {
  7. // 连接丢失后,一般在这里面进行重连
  8. System.out.println("连接断开,可以做重连");
  9. }
  10. public void messageArrived(String topic, MqttMessage message) throws Exception {
  11. // subscribe后得到的消息会执行到这里面
  12. System.out.println("接收消息主题:" + topic);
  13. System.out.println("接收消息Qos:" + message.getQos());
  14. System.out.println("接收消息内容:" + new String(message.getPayload()));
  15. }
  16. public void deliveryComplete(IMqttDeliveryToken token) {
  17. System.out.println("deliveryComplete---------" + token.isComplete());
  18. }
  19. }

Paho Java MQTT 5.0 支持

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