ApsaraMQ for MQTT：Kirim dan terima pesan antara klien MQTT dan aplikasi backend

Endpoint Client SDK

Anda dapat menemukan endpoint Client SDK di tab Endpoints pada halaman Instance Details di Konsol ApsaraMQ for MQTT.

Endpoint Cloud SDK

Anda dapat menemukan ID instans di bagian Basic Information pada halaman Instance Details di Konsol ApsaraMQ for MQTT.

Hubungkan ke broker

Buat klien MQTT dan bangun koneksi. Gunakan tcp://endpoint:1883 untuk teks biasa atau ssl://endpoint:8883 untuk SSL/TLS.

String instanceId = "XXXXX";
String endPoint = "XXXXX.mqtt.aliyuncs.com";
String accessKey = System.getenv("MQTT_AK_ENV");
String secretKey = System.getenv("MQTT_SK_ENV");
String clientId = "GID_XXXXX@@@XXXXX";

ConnectionOptionWrapper connectionOptionWrapper =
    new ConnectionOptionWrapper(instanceId, accessKey, secretKey, clientId);
final MemoryPersistence memoryPersistence = new MemoryPersistence();

final MqttClient mqttClient =
    new MqttClient("tcp://" + endPoint + ":1883", clientId, memoryPersistence);

// Atur timeout respons (ms) untuk menghindari penantian tanpa batas
mqttClient.setTimeToWait(5000);

ID klien menggunakan format {GroupID}@@@{DeviceID} dan harus unik secara global per koneksi TCP. Penggunaan ulang ID klien di beberapa koneksi menyebabkan broker memutus sesi yang ada.

Siapkan callback

Atur callback sebelum melakukan koneksi untuk menghindari kehilangan pesan selama pemulihan sesi.

mqttClient.setCallback(new MqttCallbackExtended() {
    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        System.out.println("connect success");
    }

    @Override
    public void connectionLost(Throwable throwable) {
        throwable.printStackTrace();
    }

    @Override
    public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
        // Proses pesan dalam batas waktu broker untuk menghindari pengiriman ulang.
        // Terapkan deduplikasi untuk memastikan pemrosesan idempoten.
        System.out.println(
            "receive msg from topic " + s + " , body is "
            + new String(mqttMessage.getPayload()));
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        System.out.println("send msg succeed topic is : "
            + iMqttDeliveryToken.getTopics()[0]);
    }
});

mqttClient.connect(connectionOptionWrapper.getMqttConnectOptions());

Publikasikan pesan

Publikasikan pesan ke subtopik. Subtopik ditambahkan ke topik induk dan dapat mencapai maksimal 128 karakter.

final String parentTopic = "XXXXX";
final String mq4IotTopic = parentTopic + "/" + "testMq4Iot";
final int qosLevel = 0;

MqttMessage message = new MqttMessage("hello mq4Iot pub sub msg".getBytes());
message.setQos(qosLevel);
mqttClient.publish(mq4IotTopic, message);

Level QoS: 0 (at most once), 1 (at least once), atau 2 (exactly once).

Kirim pesan point-to-point (P2P)

Pesan P2P mengirimkan pesan langsung ke klien tertentu. Klien target tidak perlu berlangganan ke topik tersebut. Publikasikan ke {parentTopic}/p2p/{targetClientId}.

String receiverId = "GID_test@@@device_002";
final String p2pSendTopic = parentTopic + "/p2p/" + receiverId;
MqttMessage p2pMessage = new MqttMessage("hello mq4Iot p2p msg".getBytes());
p2pMessage.setQos(qosLevel);
mqttClient.publish(p2pSendTopic, p2pMessage);

Kode produsen lengkap

Kode berikut menggabungkan semua langkah di atas. Buka MQ4IoTProducerDemo.java dan perbarui parameternya.

package com.aliyun.openservices.lmq.example.demo;

import com.aliyun.openservices.lmq.example.util.ConnectionOptionWrapper;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;


public class MQ4IoTProducerDemo {
    public static void main(String[] args) throws Exception {
        // ID instans ApsaraMQ for MQTT
        String instanceId = "XXXXX";

        // Endpoint Client SDK. Gunakan nama domain, bukan alamat IP.
        String endPoint = "XXXXX.mqtt.aliyuncs.com";

        // Muat kredensial dari variabel lingkungan
        String accessKey = System.getenv("MQTT_AK_ENV");
        String secretKey = System.getenv("MQTT_SK_ENV");

        // Format ID klien: {GroupID}@@@{DeviceID}
        // Harus unik per koneksi TCP. Maksimal 64 karakter.
        String clientId = "GID_XXXXX@@@XXXXX";

        // Topik induk yang dibuat di konsol ApsaraMQ for MQTT.
        // Topik yang tidak valid atau tidak sah menyebabkan broker menutup koneksi.
        final String parentTopic = "XXXXX";

        // Subtopik untuk penyaringan pesan. Maksimal 128 karakter.
        final String mq4IotTopic = parentTopic + "/" + "testMq4Iot";

        // Level QoS: 0, 1, atau 2
        final int qosLevel = 0;

        ConnectionOptionWrapper connectionOptionWrapper =
            new ConnectionOptionWrapper(instanceId, accessKey, secretKey, clientId);
        final MemoryPersistence memoryPersistence = new MemoryPersistence();

        // Hubungkan melalui TCP pada port 1883. Untuk SSL, gunakan ssl://endpoint:8883.
        final MqttClient mqttClient =
            new MqttClient("tcp://" + endPoint + ":1883", clientId, memoryPersistence);

        // Atur timeout respons (ms) untuk menghindari penantian tanpa batas
        mqttClient.setTimeToWait(5000);

        final ExecutorService executorService = new ThreadPoolExecutor(
            1, 1, 0, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());

        // Atur callback sebelum melakukan koneksi untuk menghindari kehilangan pesan
        mqttClient.setCallback(new MqttCallbackExtended() {
            @Override
            public void connectComplete(boolean reconnect, String serverURI) {
                System.out.println("connect success");
            }

            @Override
            public void connectionLost(Throwable throwable) {
                throwable.printStackTrace();
            }

            @Override
            public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
                // Proses pesan dalam batas waktu broker untuk menghindari pengiriman ulang.
                // Terapkan deduplikasi untuk memastikan pemrosesan idempoten.
                System.out.println(
                    "receive msg from topic " + s + " , body is "
                    + new String(mqttMessage.getPayload()));
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
                System.out.println("send msg succeed topic is : "
                    + iMqttDeliveryToken.getTopics()[0]);
            }
        });

        mqttClient.connect(connectionOptionWrapper.getMqttConnectOptions());

        for (int i = 0; i < 10; i++) {
            // Publikasikan ke subtopik (pub/sub standar)
            MqttMessage message = new MqttMessage("hello mq4Iot pub sub msg".getBytes());
            message.setQos(qosLevel);
            mqttClient.publish(mq4IotTopic, message);

            // Publikasikan pesan point-to-point (P2P) langsung ke klien tertentu.
            // Klien target tidak perlu berlangganan ke topik ini.
            // Format topik: {parentTopic}/p2p/{targetClientId}
            String receiverId = "xxx";
            final String p2pSendTopic = parentTopic + "/p2p/" + receiverId;
            message = new MqttMessage("hello mq4Iot p2p msg".getBytes());
            message.setQos(qosLevel);
            mqttClient.publish(p2pSendTopic, message);
        }
        Thread.sleep(Long.MAX_VALUE);
    }
}

Verifikasi produsen

Jalankan MQ4IoTProducerDemo.java. Jika koneksi berhasil, Anda akan melihat output seperti berikut:

connect success
send msg succeed topic is : XXXXX/testMq4Iot
send msg succeed topic is : XXXXX/p2p/xxx
...

Siapkan proyek

1. Unduh Cloud SDK. Lihat Catatan rilis untuk versi terbaru.

2. Unduh proyek demo: mqtt-server-sdk-demo.

3. Ekstrak paket dan impor ke IntelliJ IDEA.

4. Verifikasi bahwa file pom.xml Anda menyertakan dependensi berikut:

      <dependencies>
          <dependency>
              <groupId>com.alibaba.mqtt</groupId>
              <artifactId>server-sdk</artifactId>
              <version>1.0.0.Final</version>
          </dependency>
          <dependency>
              <groupId>com.alibaba</groupId>
              <artifactId>fastjson</artifactId>
              <version>1.2.83</version>
          </dependency>
      </dependencies>

Konfigurasikan dan jalankan konsumen

Buka MQTTConsumerDemo.java dan perbarui parameter berikut:

Kredensial dimuat dari variabel lingkungan yang sama (MQTT_AK_ENV dan MQTT_SK_ENV) yang dikonfigurasi di Langkah 2.

String domain = "post-cn-jaj3h8i****.mqtt.aliyuncs.com";
int port = 5672;
String instanceId = "post-cn-jaj3h8i****";
String accessKey = System.getenv("MQTT_AK_ENV");
String secretKey = System.getenv("MQTT_SK_ENV");
String firstTopic = "firstTopic";

ChannelConfig channelConfig = new ChannelConfig();
channelConfig.setDomain(domain);
channelConfig.setPort(port);
channelConfig.setInstanceId(instanceId);
channelConfig.setAccessKey(accessKey);
channelConfig.setSecretKey(secretKey);

ServerConsumer serverConsumer = new ServerConsumer(channelConfig, new ConsumerConfig());
serverConsumer.start();

serverConsumer.subscribeTopic(firstTopic, new MessageListener() {
    @Override
    public void process(String msgId, MessageProperties messageProperties,
                        byte[] payload) {
        System.out.println("Receive:" + msgId + ","
            + JSONObject.toJSONString(messageProperties) + ","
            + new String(payload));
    }
});

package com.aliyun.openservices.lmq.example;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.mqtt.server.ServerConsumer;
import com.alibaba.mqtt.server.callback.MessageListener;
import com.alibaba.mqtt.server.config.ChannelConfig;
import com.alibaba.mqtt.server.config.ConsumerConfig;
import com.alibaba.mqtt.server.model.MessageProperties;

public class MQTTConsumerDemo {
    public static void main(String[] args) throws Exception {
        // Endpoint Cloud SDK. Gunakan nama domain, bukan alamat IP.
        String domain = "post-cn-jaj3h8i****.mqtt.aliyuncs.com";

        // Port Cloud SDK. Harus 5672.
        int port = 5672;

        // ID instans ApsaraMQ for MQTT
        String instanceId = "post-cn-jaj3h8i****";

        // Muat kredensial dari variabel lingkungan
        String accessKey = System.getenv("MQTT_AK_ENV");
        String secretKey = System.getenv("MQTT_SK_ENV");

        // Topik induk yang akan di-subscribe.
        // Cloud SDK hanya berlangganan ke topik induk (tidak ke subtopik).
        // Topik yang tidak valid atau tidak sah menyebabkan broker menutup koneksi.
        String firstTopic = "firstTopic";

        ChannelConfig channelConfig = new ChannelConfig();
        channelConfig.setDomain(domain);
        channelConfig.setPort(port);
        channelConfig.setInstanceId(instanceId);
        channelConfig.setAccessKey(accessKey);
        channelConfig.setSecretKey(secretKey);

        ServerConsumer serverConsumer = new ServerConsumer(channelConfig, new ConsumerConfig());
        serverConsumer.start();

        // Berlangganan dan proses pesan masuk
        serverConsumer.subscribeTopic(firstTopic, new MessageListener() {
            @Override
            public void process(String msgId, MessageProperties messageProperties,
                                byte[] payload) {
                System.out.println("Receive:" + msgId + ","
                    + JSONObject.toJSONString(messageProperties) + ","
                    + new String(payload));
            }
        });
    }
}

Bagaimana cara mengaktifkan koneksi ulang otomatis?

Atur setAutomaticReconnect(true) pada MqttConnectOptions sebelum melakukan koneksi:

MqttConnectOptions options = connectionOptionWrapper.getMqttConnectOptions();
options.setAutomaticReconnect(true);
mqttClient.connect(options);

Bagaimana cara menghubungkan melalui SSL/TLS?

Ganti tcp:// dengan ssl:// dan gunakan port 8883:

final MqttClient mqttClient =
    new MqttClient("ssl://" + endPoint + ":8883", clientId, memoryPersistence);

Dapatkah saya mengirim pesan dari aplikasi backend ke klien MQTT?

Ya. Gunakan MQTTProducerDemo.java dalam proyek demo Cloud SDK untuk arah pengiriman pesan sebaliknya (backend ke klien).