ApsaraMQ for MQTT:Send and receive messages between MQTT clients by using the Java SDK

Prérequis

Avant de commencer, assurez-vous d'avoir :

• Créé des ressources ApsaraMQ for MQTT (instance, ID de groupe et topic parent)

• Une paire AccessKey pour l'authentification

• Le JDK installé

• Un IDE tel qu'IntelliJ IDEA ou Eclipse

Endpoints

Lorsqu'un client ou un service cloud communique avec ApsaraMQ for MQTT, spécifiez l'endpoint de votre instance ApsaraMQ for MQTT dans le code du SDK. Le format de l'endpoint dépend du type de SDK et de votre méthode d'accès réseau.

Format d'endpoint du SDK client

Lorsque vous vous connectez via le SDK client, utilisez les formats d'endpoint suivants :

Format d'endpoint du SDK cloud

Lorsque vous vous connectez via le SDK cloud, utilisez les formats d'endpoint suivants :

Vous pouvez également trouver l'endpoint du SDK client dans l'onglet Endpoints de la page Instance Details dans la console ApsaraMQ for MQTT. L'ID d'instance s'affiche dans la section Basic Information.

Étape 1 : Ajouter les dépendances Maven

Clonez ou téléchargez le projet de démonstration, puis vérifiez que le fichier pom.xml inclut les dépendances suivantes :

<dependencies>
    <!-- Eclipse Paho: MQTT 3.1.1 client library -->
    <dependency>
        <groupId>org.eclipse.paho</groupId>
        <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
        <version>1.2.2</version>
    </dependency>
    <!-- Commons Codec: HMAC signature calculation for authentication -->
    <dependency>
        <groupId>commons-codec</groupId>
        <artifactId>commons-codec</artifactId>
        <version>1.10</version>
    </dependency>
    <!-- Apache HttpClient: HTTP requests for token-based auth -->
    <dependency>
        <groupId>org.apache.httpcomponents</groupId>
        <artifactId>httpclient</artifactId>
        <version>4.5.2</version>
    </dependency>
    <!-- Fastjson: JSON parsing for API responses -->
    <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>fastjson</artifactId>
        <version>1.2.83</version>
    </dependency>
    <!-- Alibaba Cloud SDK for MQTT: server-side API calls -->
    <dependency>
        <groupId>com.aliyun</groupId>
        <artifactId>aliyun-java-sdk-onsmqtt</artifactId>
        <version>1.0.3</version>
    </dependency>
    <!-- Alibaba Cloud Core SDK: shared SDK infrastructure -->
    <dependency>
        <groupId>com.aliyun</groupId>
        <artifactId>aliyun-java-sdk-core</artifactId>
        <version>4.5.0</version>
    </dependency>
</dependencies>

Pour le code source et la documentation du client Eclipse Paho Java, consultez Eclipse Paho Java Client.

Étape 2 : Définir les variables d'environnement

Stockez votre paire AccessKey dans des variables d'environnement afin d'éviter toute exposition accidentelle des identifiants dans le code source.

export MQTT_AK_ENV=<your-access-key-id>
export MQTT_SK_ENV=<your-access-key-secret>

Étape 3 : Définir les paramètres de connexion

Définissez les paramètres suivants en fonction des ressources que vous avez créées dans la console ApsaraMQ for MQTT :

// Instance ID from the console's Basic Information section
String instanceId = "<your-instance-id>";

// Endpoint from the console's Endpoints tab
String endPoint = "<your-instance-id>.mqtt.aliyuncs.com";

// Credentials from environment variables
String accessKey = System.getenv("MQTT_AK_ENV");
// The AccessKey secret is required for signature authentication mode
String secretKey = System.getenv("MQTT_SK_ENV");

// Client ID format: GroupID@@@DeviceID (max 64 characters)
// Each TCP connection must use a unique client ID.
// Duplicate client IDs cause connection conflicts and unexpected disconnections.
String clientId = "<your-group-id>@@@<your-device-id>";

// Parent topic created in the console
final String parentTopic = "<your-parent-topic>";

// Subtopic for message filtering (max 128 characters total)
final String mq4IotTopic = parentTopic + "/" + "testMq4Iot";

// QoS level: 0 (at most once), 1 (at least once), or 2 (exactly once)
final int qosLevel = 0;

Étape 4 : Créer le client MQTT et enregistrer les callbacks

Enregistrez les callbacks avant la connexion. Si vous enregistrez les callbacks après l'appel à connect(), vous risquez de manquer des messages, notamment lors de la reprise d'une session persistante.

Créez le client MQTT :

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

final MemoryPersistence memoryPersistence = new MemoryPersistence();

// Protocol and port must match:
//   TCP  -> tcp://endpoint:1883
//   SSL  -> ssl://endpoint:8883
final MqttClient mqttClient = new MqttClient(
    "tcp://" + endPoint + ":1883", clientId, memoryPersistence);

// Timeout for waiting on broker responses (milliseconds)
mqttClient.setTimeToWait(5000);

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

Configurez les callbacks d'événements pour gérer les événements de connexion et les messages entrants :

mqttClient.setCallback(new MqttCallbackExtended() {
    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        // Subscribe immediately after each connection (including reconnections)
        System.out.println("connect success");
        executorService.submit(() -> {
            try {
                final String[] topicFilter = {mq4IotTopic};
                final int[] qos = {qosLevel};
                mqttClient.subscribe(topicFilter, qos);
            } catch (MqttException e) {
                e.printStackTrace();
            }
        });
    }

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

    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
        // Process incoming messages here.
        // Do not throw exceptions from this callback. The broker treats a
        // normally returned callback as message acknowledgment.
        // Implement deduplication to ensure idempotent consumption.
        System.out.println(
            "receive msg from topic " + topic + " , body is "
            + new String(mqttMessage.getPayload()));
    }

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

Comportements clés des callbacks :

Étape 5 : Se connecter et publier des messages

Connectez-vous au broker et publiez des messages. Cet exemple envoie à la fois des messages pub/sub normaux et des messages point à point (P2P) :

mqttClient.connect(connectionOptionWrapper.getMqttConnectOptions());

for (int i = 0; i < 10; i++) {
    // Publish a normal message to the subscribed topic
    MqttMessage message = new MqttMessage("hello mq4Iot pub sub msg".getBytes());
    message.setQos(qosLevel);
    mqttClient.publish(mq4IotTopic, message);

    // Publish a P2P message directly to a specific client
    // P2P topic format: {parentTopic}/p2p/{targetClientId}
    // The target client does not need to subscribe to this topic.
    final String p2pSendTopic = parentTopic + "/p2p/" + clientId;
    message = new MqttMessage("hello mq4Iot p2p msg".getBytes());
    message.setQos(qosLevel);
    mqttClient.publish(p2pSendTopic, message);
}

// Keep the client running to continue receiving messages
Thread.sleep(Long.MAX_VALUE);

Messagerie P2P : Lorsque vous connaissez le client cible exact, envoyez un message au topic {parentTopic}/p2p/{targetClientId}. Le client cible reçoit le message sans avoir besoin de s'y abonner. Cela simplifie les scénarios où un message est destiné à un seul destinataire.

Code complet

Le listing suivant combine toutes les étapes ci-dessus en une seule classe exécutable. Il est également disponible dans le projet de démonstration sur GitHub.

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 MQ4IoTSendMessageToMQ4IoTUseSignatureMode {
    public static void main(String[] args) throws Exception {
        // --- Connection parameters ---
        String instanceId = "<your-instance-id>";
        String endPoint = "<your-instance-id>.mqtt.aliyuncs.com";
        String accessKey = System.getenv("MQTT_AK_ENV");
        String secretKey = System.getenv("MQTT_SK_ENV");
        String clientId = "<your-group-id>@@@<your-device-id>";
        final String parentTopic = "<your-parent-topic>";
        final String mq4IotTopic = parentTopic + "/" + "testMq4Iot";
        final int qosLevel = 0;

        // --- Create client ---
        ConnectionOptionWrapper connectionOptionWrapper =
            new ConnectionOptionWrapper(instanceId, accessKey, secretKey, clientId);
        final MemoryPersistence memoryPersistence = new MemoryPersistence();
        final MqttClient mqttClient = new MqttClient(
            "tcp://" + endPoint + ":1883", clientId, memoryPersistence);
        mqttClient.setTimeToWait(5000);

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

        // --- Register callbacks before connecting ---
        mqttClient.setCallback(new MqttCallbackExtended() {
            @Override
            public void connectComplete(boolean reconnect, String serverURI) {
                System.out.println("connect success");
                executorService.submit(() -> {
                    try {
                        final String[] topicFilter = {mq4IotTopic};
                        final int[] qos = {qosLevel};
                        mqttClient.subscribe(topicFilter, qos);
                    } catch (MqttException e) {
                        e.printStackTrace();
                    }
                });
            }

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

            @Override
            public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
                System.out.println(
                    "receive msg from topic " + topic + " , body is "
                    + new String(mqttMessage.getPayload()));
            }

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

        // --- Connect and publish ---
        mqttClient.connect(connectionOptionWrapper.getMqttConnectOptions());
        for (int i = 0; i < 10; i++) {
            MqttMessage message = new MqttMessage("hello mq4Iot pub sub msg".getBytes());
            message.setQos(qosLevel);
            mqttClient.publish(mq4IotTopic, message);

            final String p2pSendTopic = parentTopic + "/p2p/" + clientId;
            message = new MqttMessage("hello mq4Iot p2p msg".getBytes());
            message.setQos(qosLevel);
            mqttClient.publish(p2pSendTopic, message);
        }
        Thread.sleep(Long.MAX_VALUE);
    }
}

Vérifier les résultats

Après avoir exécuté l'application, interrogez les traces de messages dans la console ApsaraMQ for MQTT pour confirmer que les messages ont été envoyés et reçus. Pour les étapes détaillées, consultez Interroger les traces de messages.

Étapes suivantes

• Envoyer et recevoir des messages entre clients MQTT et applications de service backend

• Termes