ApsaraMQ for MQTT:Envio e recebimento de mensagens entre clientes MQTT e aplicações de backend

Tipos de endpoint

Um Public-dedicated endpoint fornece uma conexão de Internet exclusiva para a instância. Durante picos de tráfego, evita interferências de outras instâncias no mesmo cluster e reduz riscos de falhas de conexão e latência de mensagens. Recomendamos endpoints dedicados de Internet para instâncias de produção.

O VPC-dedicated Endpoint estabelece uma conexão de rede interna exclusiva para a instância, com maior isolamento e segurança. Essa opção é ideal para cenários com requisitos rigorosos de segurança no acesso a dados.

Também é possível adicionar o Anti-DDoS da Alibaba Cloud ao endereço IP público de um endpoint dedicado para reforçar a proteção do acesso pela Internet.

Suporte por versão

Endpoints do SDK do cliente

Ao se conectar com o SDK do cliente, configure o endpoint nos seguintes formatos:

Encontre o endpoint do SDK do cliente na aba Endpoints da página Instance Details no console do ApsaraMQ for MQTT.

Endpoints do SDK da cloud

Ao se conectar com o SDK da cloud, configure o endpoint nos seguintes formatos:

Localize o ID da instância na seção Basic Information da página Instance Details no console do ApsaraMQ for MQTT. O VPC-dedicated Endpoint fica disponível somente após a criação de um endpoint de rede VPC pelo botão Create VPC Endpoint na aba Endpoints.

<dependencies>
    <dependency>
        <groupId>commons-codec</groupId>
        <artifactId>commons-codec</artifactId>
        <version>1.10</version>
    </dependency>
    <dependency>
        <groupId>org.eclipse.paho</groupId>
        <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
        <version>1.2.2</version>
    </dependency>
    <dependency>
        <groupId>org.apache.httpcomponents</groupId>
        <artifactId>httpclient</artifactId>
        <version>4.5.2</version>
    </dependency>
    <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>fastjson</artifactId>
        <version>1.2.83</version>
    </dependency>
    <dependency>
        <groupId>com.aliyun</groupId>
        <artifactId>aliyun-java-sdk-onsmqtt</artifactId>
        <version>1.0.3</version>
    </dependency>
    <dependency>
        <groupId>com.aliyun</groupId>
        <artifactId>aliyun-java-sdk-core</artifactId>
        <version>4.5.0</version>
    </dependency>
</dependencies>

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

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

// Set the response timeout (ms) to avoid indefinite waits
mqttClient.setTimeToWait(5000);

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 {
        // Process messages within the broker's timeout to avoid redelivery.
        // Implement deduplication to ensure idempotent processing.
        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());

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

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

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 {
        // ApsaraMQ for MQTT instance ID
        String instanceId = "XXXXX";

        // Client SDK endpoint. Use the domain name, not the IP address.
        String endPoint = "XXXXX.mqtt.aliyuncs.com";

        // Load credentials from environment variables
        String accessKey = System.getenv("MQTT_AK_ENV");
        String secretKey = System.getenv("MQTT_SK_ENV");

        // Client ID format: {GroupID}@@@{DeviceID}
        // Must be unique per TCP connection. Max 64 characters.
        String clientId = "GID_XXXXX@@@XXXXX";

        // Parent topic created in the ApsaraMQ for MQTT console.
        // An invalid or unauthorized topic causes the broker to close the connection.
        final String parentTopic = "XXXXX";

        // Subtopic for message filtering. Max 128 characters.
        final String mq4IotTopic = parentTopic + "/" + "testMq4Iot";

        // QoS level: 0, 1, or 2
        final int qosLevel = 0;

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

        // Connect over TCP on port 1883. For SSL, use ssl://endpoint:8883.
        final MqttClient mqttClient =
            new MqttClient("tcp://" + endPoint + ":1883", clientId, memoryPersistence);

        // Set the response timeout (ms) to avoid indefinite waits
        mqttClient.setTimeToWait(5000);

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

        // Set callbacks before connecting to avoid missing messages
        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 {
                // Process messages within the broker's timeout to avoid redelivery.
                // Implement deduplication to ensure idempotent processing.
                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++) {
            // Publish to a subtopic (standard pub/sub)
            MqttMessage message = new MqttMessage("hello mq4Iot pub sub msg".getBytes());
            message.setQos(qosLevel);
            mqttClient.publish(mq4IotTopic, message);

            // Publish a point-to-point (P2P) message directly to a specific client.
            // The target client does not need to subscribe to this topic.
            // Topic format: {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);
    }
}

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

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

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 {
        // Cloud SDK endpoint. Use the domain name, not the IP address.
        String domain = "post-cn-jaj3h8i****.mqtt.aliyuncs.com";

        // Cloud SDK port. Must be 5672.
        int port = 5672;

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

        // Load credentials from environment variables
        String accessKey = System.getenv("MQTT_AK_ENV");
        String secretKey = System.getenv("MQTT_SK_ENV");

        // Parent topic to subscribe to.
        // The cloud SDK subscribes to parent topics only (no subtopics).
        // An invalid or unauthorized topic causes the broker to close the connection.
        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();

        // Subscribe and process incoming messages
        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));
            }
        });
    }
}

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

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