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

Pré-requisitos

Antes de começar, verifique se você tem:

• Created ApsaraMQ for MQTT resources (instância, group ID e tópico pai)

• Um AccessKey pair para autenticação

• JDK instalado

• Uma IDE como IntelliJ IDEA ou Eclipse

Endpoints

Para que um cliente ou service na cloud se comunique com o ApsaraMQ for MQTT, especifique o endpoint da sua instância do ApsaraMQ for MQTT no código do SDK. O formato do endpoint varia conforme o tipo de SDK e o método de acesso à rede.

Formato de endpoint do SDK de cliente

Ao se conectar pelo SDK de cliente, use os seguintes formatos de endpoint:

Formato de endpoint do SDK da cloud

Para conexões via SDK da cloud, adote os seguintes formatos de endpoint:

O endpoint do SDK de cliente também está disponível na aba Endpoints da página Instance Details no console do ApsaraMQ for MQTT. O ID da instância aparece na seção Basic Information.

Etapa 1: Adicionar dependências do Maven

Clone ou baixe o projeto de demonstração e verifique se o arquivo pom.xml contém as seguintes dependências:

<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>

Para consultar o código-fonte e a documentação do Eclipse Paho Java Client, acesse Eclipse Paho Java Client.

Etapa 2: Definir variáveis de ambiente

Armazene seu par de AccessKey em variáveis de ambiente para evitar a exposição acidental de credenciais no código-fonte.

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

Etapa 3: Definir parâmetros de conexão

Configure os parâmetros abaixo com base nos recursos criados no console do 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;

Etapa 4: Criar o cliente MQTT e registrar callbacks

Registre os callbacks antes de estabelecer a conexão. Se você registrá-los após chamar connect(), poderá perder mensagens, especialmente ao retomar uma sessão persistente.

Crie o cliente 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>());

Configure os callbacks de eventos para tratar eventos de conexão e mensagens recebidas:

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]);
    }
});

Comportamentos principais dos callbacks:

Etapa 5: Conectar e publicar mensagens

Conecte-se ao broker e publique mensagens. Este exemplo envia tanto mensagens comuns de publicação/assinatura quanto mensagens ponto a ponto (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);

Mensagens P2P: Quando o cliente de destino for conhecido, envie a mensagem para o tópico {parentTopic}/p2p/{targetClientId}. O cliente destinatário recebe a mensagem sem precisar assinar o tópico, o que simplifica cenários de comunicação com um único receptor.

Código completo

A listagem a seguir reúne todas as etapas anteriores em uma única classe executável. Ela também está disponível no projeto de demonstração no 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);
    }
}

Verificar os resultados

Após executar a aplicação, consulte os rastreamentos de mensagens no console do ApsaraMQ for MQTT para confirmar o envio e o recebimento das mensagens. Para instruções detalhadas, consulte Consultar rastreamentos de mensagens.

Próximos passos

• Send and receive messages between MQTT clients and backend service applications

• Termos