背景信息
云消息队列 MQTT 版支持云端SDK,云上应用可直接通过云端SDK接入云消息队列 MQTT 版服务端进行消息收发。云端SDK使用,请参见云端开发概述。
同时云消息队列 MQTT 版支持和其他云产品进行互通,当前支持的云产品有云消息队列 RocketMQ 版。
本文以公网环境中的Java SDK为例说明如何将云消息队列 RocketMQ 版数据流入至云消息队列 MQTT 版。

网络访问
云消息队列 MQTT 版同时提供了
公网接入点和
VPC 接入点。
- 公网接入点为本地公网环境访问的IP地址,一般用于物联网和移动互联网场景中;
- VPC 接入点为云上私网访问的IP地址,一般用于云端应用接入云消息队列 MQTT 版。
重要
客户端使用接入点连接服务时务必使用域名接入,不得直接使用域名背后的IP地址直接连接,因为IP地址随时会变化。在以下使用情况中出现的问题
云消息队列 MQTT 版产品方概不负责:
- 客户端不使用域名接入而是使用IP地址接入,产品方更新了域名解析导致原有IP地址失效。
- 客户端网络对IP地址设置网络防火墙策略,产品方更新了域名解析后新IP地址被您的防火墙策略拦截。
前提条件
- 安装IDE。您可以使用IntelliJ IDEA或者Eclipse,本文以IntelliJ IDEA为例。
安装Java 8或11。更多信息,请参见安装JDK。
已创建云消息队列 MQTT 版实例、Topic和Group ID,具体操作,请参见创建资源。
已创建云消息队列 RocketMQ 版实例、Topic和Group ID,具体操作,请参见步骤二:创建资源。
1.创建数据流入规则
登录云消息队列 MQTT 版控制台,并在左侧导航栏单击实例列表。
在顶部菜单栏选择目标地域,然后在实例列表中单击实例名称进入实例详情页面。
在左侧导航栏单击规则管理,然后在页面左上角,单击创建规则。
在创建规则页面完成以下操作。
配置基本信息。输入规则ID,选择数据流入的规则类型。

配置规则源。选择已经创建好的云消息队列 RocketMQ 版的实例和Topic。

配置规则目标。选择已经创建好的云消息队列 MQTT 版的Topic。

2.准备测试代码
2.1下载示例代码
下载mqtt-java-demo,并解压该Demo工程包至您指定的文件夹。
在解压的Demo工程中找到lmq-java-demo文件夹,将此文件夹导入IntelliJ IDEA,并确认pom.xml中已包含以下依赖。
<dependencies>
<dependency>
<groupId>org.bouncycastle</groupId>
<artifactId>bcprov-jdk15on</artifactId>
<version>1.70</version>
</dependency>
<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.openservices</groupId>
<artifactId>ons-client</artifactId>
<version>1.8.5.Final</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>
配置访问凭证。
2.2收发消息代码
在RocketMQSendMessageToMQ4IoT.java
类中包含了发送RocketMQ消息和使用MQTT消费的代码,按代码注释说明填写云消息队列 RocketMQ 版和云消息队列 MQTT 版资源的参数。
测试中可以将其中发送P2P消息的相关代码注释掉,示例代码如下。
import com.aliyun.openservices.lmq.example.util.ConnectionOptionWrapper;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.SendResult;
import java.util.Properties;
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 RocketMQSendMessageToMQ4IoT {
public static void main(String[] args) throws Exception {
Properties properties = new Properties();
properties.setProperty(PropertyKeyConst.GROUP_ID, "GID-XXXXX");
properties.put(PropertyKeyConst.AccessKey, System.getenv("MQTT_AK_ENV"));
properties.put(PropertyKeyConst.SecretKey, System.getenv("MQTT_SK_ENV"));
properties.put(PropertyKeyConst.NAMESRV_ADDR, "XXXX");
final String parentTopic = "XXXXX";
Producer producer = ONSFactory.createProducer(properties);
producer.start();
String instanceId = "XXXXX";
String endPoint = "XXXXXX.mqtt.aliyuncs.com";
String accessKey = System.getenv("MQTT_AK_ENV");
String secretKey = System.getenv("MQTT_SK_ENV");
String clientId = "GID_XXXX@@@XXXXX";
final String subTopic = "/testMq4Iot";
final String mq4IotTopic = parentTopic + subTopic;
final int qosLevel = 0;
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>());
mqttClient.setCallback(new MqttCallbackExtended() {
@Override
public void connectComplete(boolean reconnect, String serverURI) {
System.out.println("connect success");
executorService.submit(new Runnable() {
@Override
public void run() {
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 s, MqttMessage mqttMessage) throws Exception {
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++) {
Message msg = new Message(parentTopic, "MQ2MQTT", "hello mq send mqtt msg".getBytes());
msg.putUserProperties(PropertyKeyConst.MqttSecondTopic, subTopic);
SendResult result = producer.send(msg);
System.out.println(result);
}
Thread.sleep(Long.MAX_VALUE);
}
}
3.结果验证
执行RocketMQSendMessageToMQ4IoT.java
类中的Main函数运行代码。可以根据下面操作验证消息的发送和消费情况。
代码验证
如下图所示,RocketMQ消息已经成功发送,MQTT客户端也已经成功消费。

控制台验证