ApsaraMQ for MQTT:Import data from ApsaraMQ for RocketMQ to ApsaraMQ for MQTT

Network access

Process

The following figure shows the process for sending messages from a backend application to an ApsaraMQ for MQTT client.

Step 1: Create an ApsaraMQ for MQTT instance and obtain the endpoint of the instance

1. Log on to the ApsaraMQ for MQTT console. In the left-side navigation pane, click Instances.

2. In the top navigation bar, select the region where the instance that you want to manage resides. Then, in the upper-left corner, click Create Instance.

3. In the panel that appears, use the default value Subscription for the Billing Method parameter. Then, Click OK.

4. On the buy page that appears, select the instance specifications that you want to purchase based on your business requirements, select the terms of service, and then click Buy Now.

   For information about the instance editions provided by ApsaraMQ for MQTT and their feature differences, see Instance editions.

5. On the order payment page, follow the on-screen instructions to complete the payment.

6. After you complete the payment, click Console on the page that appears.

7. Go back to the ApsaraMQ for MQTT console. In the left-side navigation pane, click Instances. In the top navigation bar, select the region in which your instance is deployed.

8. On the Instances page, click the name of the instance or click Details in the Actions column of the instance to go to the Instance Details page.

9. On the Instance Details page, click the Endpoints tab. On the tab, you can view the endpoint information. In the example, the public endpoint is used.

   

Step 2: Create a parent topic

The MQTT protocol supports multi-level topics. You must create a parent topic in the ApsaraMQ for MQTT console. You can specify subtopics in the code without the need to create them in the console. A parent topic and its subtopics are named in the following format: <Name of a parent topic>/<Name of a level-2 topic>/<Name of a level-3 topic>. A parent topic and its subtopics are separated by forward slashes (/). Example: SendMessage/demo/producer. The total length of the names of a parent topic and its subtopics cannot exceed 64 characters. For more information about topics, see Terms.

1. Log on to the ApsaraMQ for MQTT console. In the left-side navigation pane, click Instances.

2. In the top navigation bar, select the region where the instance that you want to manage resides. On the Instances page, click the instance name to go to the Instance Details page.

3. In the left-side navigation pane, click Topics. In the upper-left corner of the Topics page, click Create Topic.
4. In the Create Topic panel, set the Name and Description parameters for the topic and click OK in the lower-left corner.

Step 3: Create a group

For more information about group IDs, see Terms.

1. Log on to the ApsaraMQ for MQTT console. In the left-side navigation pane, click Instances.

2. In the top navigation bar, select the region where the instance that you want to manage resides. On the Instances page, click the instance name to go to the Instance Details page.

3. In the left-side navigation pane, click Groups. In the upper-left corner of the Groups page, click Create Group.
4. In the Create Group panel, set the Group ID parameter and click OK.

Step 4: Create a data inbound rule

The parameter values that you set in the rule must be the same as those that correspond to the resources you create.

1. Log on to the ApsaraMQ for MQTT console. In the left-side navigation pane, click Instances.

2. In the top navigation bar, select the region where the instance that you want to manage resides. On the Instances page, click the instance name to go to the Instance Details page.

3. In the left-side navigation pane, click Rules. In the upper-left corner of the Rules page, click Create Rule.

4. On the Create Rule page, perform the following steps:

   1. In the Configure Basic Information step, configure the following parameters and click Next.

      Parameter	Example	Description
      Rule ID	111111	The ID of the rule. An ID is a globally unique identifier of a rule. A rule ID can contain only letters, digits, hyphens (-), and underscores (_) and must contain at least one letter or digit. A rule ID must be 3 to 64 characters in length. If the value contains more than 64 characters, it is automatically truncated. The ID of a rule cannot be updated after the rule is created.
      Description	migrate from rocketmq	The description of the rule.
      Status	Enable	Specifies whether to enable the current rule. Valid values: Enable Disable
      Rule Type	Inbound data	The type of the rule. Valid values: Data Outbound: The rule is used to export data from ApsaraMQ for MQTT to other Alibaba Cloud services. For more information, see Export data from ApsaraMQ for MQTT to other Alibaba Cloud services. Data Inbound: The rule is used to import data from other Alibaba Cloud services to ApsaraMQ for MQTT. For more information, see Import data from other cloud services to ApsaraMQ for MQTT. Client Status Notification: The rule is used to export the status event data of a ApsaraMQ for MQTT client to other Alibaba Cloud services. For more information, see Export online and offline events of Message Queue for MQTT clients.

   2. In the Configure Rule Source step, specify the data source and click Next.

      Parameter	Example	Description
      Source Service Type	Message Queue for Apache RocketMQ	The cloud service from which the data is forwarded to ApsaraMQ for MQTT. Note Only ApsaraMQ for RocketMQ is supported.
      Message Queue for Apache RocketMQ Instance	MQ_INST_13801563067*****_BbyOD2jQ	The ID of the ApsaraMQ for RocketMQ instance from which the data is forwarded. Note You can select only an instance that resides in the same region as the ApsaraMQ for MQTT instance.
      Topic	TopicA	The ApsaraMQ for RocketMQ topic from which the data is forwarded. In this example, the messages of Topic A are forwarded to a topic of ApsaraMQ for MQTT.

   3. In the Configure Rule Destination step, specify the destination to which the data is forwarded and click Create.

      Parameter	Example	Description
      Topic	TopicB	The topic of ApsaraMQ for MQTT to which the data is forwarded from other Alibaba Cloud services.

   You can view the data inbound rule that you create on the Rules page.

Step 5: Send and receive messages by using the SDK for Java

1. Download a third-party open source SDK for Java. Download link: Eclipse Paho Java Client.

2. Download the demo of the ApsaraMQ for MQTT SDK for Java for reference during code development. Download link: mqtt-java-demo.

3. Decompress the demo project package to a specific folder.

4. In IntelliJ IDEA, import the extracted files to create a project and check whether the pom.xml file contains the following dependencies:

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

   Note

   For information about the ons-client versions of the SDK for Java, see Release notes.

5. In the RocketMQSendMessageToMQ4IoT.java class, specify parameter values based on the instructions provided in the comments. Most parameter values correspond to the ApsaraMQ for MQTT resources that you created from Step 1 to Step 3 and the resources that you created in ApsaraMQ for RocketMQ. Then, execute the main() function to send and receive messages.

   Sample code:

   Note Before you use the sample code to send and receive messages, you must configure environment variables to obtain the credentials that are used to access ApsaraMQ for MQTT. For information about how to configure environment variables, see Configure an access credential.

   The environment variable name of the AccessKey ID that is used to access ApsaraMQ for MQTT is MQTT_AK_ENV, and the environment variable name of the AccessKey secret that is used to access ApsaraMQ for MQTT is MQTT_SK_ENV.

   package com.aliyun.openservices.lmq.example.demo;
   
   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 {
           /**
            * Initialize an ApsaraMQ for RocketMQ client as a sender. In most business scenarios, the sender is deployed on a backend application. 
            */
           Properties properties = new Properties();
           /**
            * The ID of the group that you created in the ApsaraMQ for RocketMQ console. 
            */
           properties.setProperty(PropertyKeyConst.GROUP_ID, "GID-XXXXX");
           /**
            * The AccessKey ID that you created in the Alibaba Cloud Resource Access Management (RAM) console for identity authentication. 
            * The AccessKey pair of an Alibaba Cloud account has permissions on all API operations. To prevent security risks, we recommend that you use a RAM user to call API operations or perform routine O&M. 
            * We strongly recommend that you do not save an AccessKey pair in the project code. Otherwise, the AccessKey pair may be leaked and all resources that are contained in your account may be exposed to potential security risks. 
            * In this example, the AccessKey pair is saved in the environment variables. 
            */
           properties.put(PropertyKeyConst.AccessKey, System.getenv("MQTT_AK_ENV"));
           /**
            * The AccessKey secret that you created in the Alibaba Cloud RAM console for identity authentication. The AccessKey secret is required only for signature authentication. 
            */
           properties.put(PropertyKeyConst.SecretKey, System.getenv("MQTT_SK_ENV"));
           /**
            * The TCP endpoint that is used to access the ApsaraMQ for RocketMQ instance. You can obtain the TCP endpoint on the Instance Details page in the ApsaraMQ for RocketMQ console. 
            */
           properties.put(PropertyKeyConst.NAMESRV_ADDR, "XXXX");
           /**
            * The topic that you created in the ApsaraMQ for RocketMQ console. 
            * When you exchange data between ApsaraMQ for RocketMQ and ApsaraMQ for MQTT, only parent topics can be used by the ApsaraMQ for RocketMQ client. 
            */
           final String parentTopic = "XXXXX";
           Producer producer = ONSFactory.createProducer(properties);
           producer.start();
           //////////////////////////////////////////////////////////////////////////////////////////////////////////////////
           /**
            * Initialize an ApsaraMQ for MQTT client as a receiver. In most business scenarios, the receiver is deployed on a mobile terminal. 
            */
   
           /**
            * The ID of the ApsaraMQ for MQTT instance that you created in the ApsaraMQ for MQTT console. 
            */
           String instanceId = "XXXXX";
           /**
            * The endpoint that you want to use to access the ApsaraMQ for MQTT instance. You can obtain the endpoint on the Instance Details page in the ApsaraMQ for MQTT console. 
            */
           String endPoint = "XXXXXX.mqtt.aliyuncs.com";
           /**
            * The AccessKey ID that you created in the Alibaba Cloud RAM console for identity authentication. 
            * The AccessKey pair of an Alibaba Cloud account has permissions on all API operations. To prevent security risks, we recommend that you use a RAM user to call API operations or perform routine O&M. 
            * We strongly recommend that you do not save an AccessKey pair in the project code. Otherwise, the AccessKey pair may be leaked and all resources that are contained in your account may be exposed to potential security risks. 
            * In this example, the AccessKey pair is saved in the environment variables. 
            */
           String accessKey = System.getenv("MQTT_AK_ENV");
           /**
            * The AccessKey secret that you created in the Alibaba Cloud RAM console for identity authentication. The AccessKey secret is required only for signature authentication. 
            */
           String secretKey = System.getenv("MQTT_SK_ENV");
           /**
            * The globally unique ID that the system assigns to the ApsaraMQ for MQTT client. The client ID must vary based on the TCP connection. If multiple TCP connections use the same client ID, exceptions occur and the connections are unexpectedly closed. 
            * The client ID consists of a group ID and a device ID and is in the GroupID@@@DeviceID format. The group ID is the ID of the group that you create in the ApsaraMQ for MQTT console. The device ID is a custom ID specified by you. The client ID cannot exceed 64 characters in length. 
            */
           String clientId = "GID_XXXX@@@XXXXX";
           /**
            * ApsaraMQ for MQTT allows you to use a subtopic to filter messages. You can specify a string as the name of the subtopic. 
            * The value of the mq4IotTopic parameter can be up to 128 characters in length. 
            */
           final String subTopic = "/testMq4Iot";
           final String mq4IotTopic = parentTopic + subTopic;
           /**
            * The quality of service (QoS) level for message transmission. Valid values: 0, 1, and 2. For more information, see Terms. 
            */
           final int qosLevel = 0;
           ConnectionOptionWrapper connectionOptionWrapper = new ConnectionOptionWrapper(instanceId, accessKey, secretKey, clientId);
           final MemoryPersistence memoryPersistence = new MemoryPersistence();
           /**
            * The protocol and port that are used by the ApsaraMQ for MQTT client. The protocol and port that are used by the ApsaraMQ for MQTT client must match. If Secure Sockets Layer (SSL) encryption is used, the protocol and port must be specified as ssl://endpoint:8883. 
            */
           final MqttClient mqttClient = new MqttClient("tcp://" + endPoint + ":1883", clientId, memoryPersistence);
           /**
            * The timeout period during which the ApsaraMQ for MQTT client waits for a response. The timeout period prevents the ApsaraMQ for MQTT client from waiting for a response for an indefinite period of time. 
            */
           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) {
                   /**
                    * The topic to which the consumer must subscribe at the earliest opportunity after the client connection is established. 
                    */
                   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 {
                   /**
                    * The callback that is invoked to consume the messages that are published. Make sure that the callback does not throw exceptions. If a response is returned for the callback, the messages are consumed. 
                    * The messages must be consumed in the specified period of time. If a message is not consumed within the timeout period specified by the ApsaraMQ for MQTT broker, the broker may attempt to resend the message. Make sure that deduplication is performed to ensure idempotence for message consumption. 
                    */
                   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++) {
               /**
                * Specify the parent topic as the topic and MQ2MQTT as the tag when you use the ApsaraMQ for RocketMQ client to send messages to the ApsaraMQ for MQTT client. 
                */
               Message msg = new Message(parentTopic, "MQ2MQTT", "hello mq send mqtt msg".getBytes());
               /**
                * You can use the MqttSecondTopic parameter to specify a subtopic when you use the ApsaraMQ for RocketMQ client to send messages to the ApsaraMQ for MQTT client. 
                */
               msg.putUserProperties(PropertyKeyConst.MqttSecondTopic, subTopic);
               SendResult result = producer.send(msg);
               System.out.println(result);
               /**
                * Send a point-to-point (P2P) message and specify a subtopic.  
                */
               msg.putUserProperties(PropertyKeyConst.MqttSecondTopic, "/p2p/" + clientId);
               result = producer.send(msg);
               System.out.println(result);
           }
           Thread.sleep(Long.MAX_VALUE);
   
       }
   
   }

Verify the results

After the producer sends a message and the consumer subscribes to the message, you can query the trace of the message in the ApsaraMQ for MQTT console to verify whether the message is sent and received. For more information, see Query message traces.

References

• Use ApsaraMQ for MQTT SDK for Java to implement messaging between ApsaraMQ for MQTT clients

• Export data from ApsaraMQ for MQTT to ApsaraMQ for RocketMQ

• Send client status notification messages from ApsaraMQ for MQTT to ApsaraMQ for RocketMQ