Java客户端订阅BifroMQ服务端MQTT协议消息-摩杜云开发者社区

1 安装BifroMQ

docker run -d --name bifromq -p 1883:1883 bifromq/bifromq:latest

来看如何连接到BifroMQ呢？

2 MQTT Client安装

使用已有的第三方 MQTT 客户端工具，快速验证是否可与 BifroMQ 提供的 MQTT 服务交流发送或接收消息。

免费第三方 MQTT 客户端工具较多：MQTT.fx、MQTTBox、MQTTLens、MQTTX等。这以 MQTTX 为例。

MQTTX安装

访问 https://mqttx.app/ 下载 MQTTX 并安装。

3 验证连接

3.1 新建配置

打开 MQTTX，点击“New Connection”或者左侧边栏上的“+”，创建一个新的连接配置。

3.2 各项必填配置

Name：本配置的名字，随意写
ClientID：客户端ID，自行设置或使用右边的按钮随机生成。支持“a-z”,“0-9”,“_”,“-”字符且不能大于128bytes，UTF8编码，不可重名。
Host：连接地址。前面的协议头可以自行选择，mqtt://、mqtts://、wss:// 分别对应 MQTT 连接中的 TCP、TLS/SSL、WSS 三种方式。后面域名请填写可以实际访问到 BifroMQ MQTT 服务的地址。
Port：连接端口。根据选择的协议头确定对应的端口。端口默认： TCP:1883、TLS/SSL:1884、WSS:443。
Username & Password：您拥有的用户名和密码。
MQTT Version：选择3.1.1

正确填写连接配置后，点击右上角 Connect 连接服务器。一切正常会提示Connect Successful，并转入消息收发界面。

订阅主题

点击左上的“New Subscription”按钮，并在弹出的对话框中填入主题Topic。请注意，须填写你有订阅权限的主题，这里我们使用一个测试主题，对应此设备为$iot/device1/user/fortest，选择默认QoS 0，点击右下“Confirm”进行订阅。

发布消息

在消息收发界面右下填入主题Topic，这里我们依然填写刚才的测试主题，自己给自己发送一条消息。选择默认的QoS 0，填写消息内容后点击发送按钮完成发送。

收到消息

发送成功后，可以在右方看到已发送的消息。MQTT服务工作正常的话，已经订阅了该主题的我们可以收到该消息，也会展示在对话界面左边。

4 Java 客户端

4.1 添加依赖

根据要使用的MQTT协议版本，在Maven工程中添加如下依赖：

MQTT 3.1和3.1.1版本

Explain<dependencies>
  <dependency>
      <groupId>org.eclipse.paho</groupId>
      <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
      <version>1.2.0</version>
    </dependency>
</dependencies>

MQTT 5.0版本

Explain<dependency>
   <groupId>org.eclipse.paho</groupId>
   <artifactId>org.eclipse.paho.mqttv5.client</artifactId>
   <version>1.2.5</version>
</dependency>∫

发布消息

以下示例代码上报物模型属性LightSwitch。

Explain// Paho MQTT发布消息。
String topic = "/sys/" + productKey + "/" + deviceName + "/thing/event/property/post";
String content = "{\"id\":\"1\",\"version\":\"1.0\",\"params\":{\"LightSwitch\":1}}";
MqttMessage message = new MqttMessage(content.getBytes());
message.setQos(0);
sampleClient.publish(topic, message);

订阅Topic，获取云端下发消息

订阅的是上报属性值后，物联网平台返回应答消息的Topic。

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

public class MqttClientExample {

    private static final String BROKER_URI = "tcp://localhost:1883";
    private static final String CLIENT_ID = "mqtt-java-client-example";
    private static final String TOPIC = "$iot/device1/user/fortest";

    public static void main(String[] args) throws MqttException {
        IMqttClient mqttClient = new MqttClient(BROKER_URI, CLIENT_ID);

        MqttConnectOptions options = new MqttConnectOptions();
        options.setAutomaticReconnect(true);
        options.setCleanSession(true);
        options.setConnectionTimeout(10);
        mqttClient.connect(options);
        System.out.println("Connected to MQTT broker");

        // 订阅主题
        mqttClient.subscribe(TOPIC, 0, new MqttPostPropertyMessageListener());
    }

    static class MqttPostPropertyMessageListener implements IMqttMessageListener {

        @Override
        public void messageArrived(String topic, MqttMessage payload) throws Exception {
            System.out.println("reply topic  : " + topic);
            System.out.println("reply payload: " + payload.toString());
        }

    }
}