Consume device messages using an ApsaraMQ for RocketMQ client

Scenario description

Assume that a device reports its geographic location (RID) and a message (CusMsg) to a custom topic in IoT Platform. You need to forward the device message CusMsg to a business server for processing if the RID is cn-shanghai. If the message rate between devices and IoT Platform exceeds 1,000 QPS, we recommend that you use the tag filtering feature of ApsaraMQ for RocketMQ to retrieve the filtered device messages.

The data forwarding and consumption flow is as follows:

1. On the business server, you can use the ApsaraMQ for RocketMQ SDK to register a consumer that receives messages forwarded from IoT Platform.

2. First, you must configure data forwarding from IoT Platform to ApsaraMQ for RocketMQ and set properties and tags for device messages. These properties and tags are used as filter conditions for message consumption. Then, you can connect the device to IoT Platform to report data. The data is then forwarded to ApsaraMQ for RocketMQ.

3. When the consumer that is registered using the ApsaraMQ for RocketMQ SDK retrieves messages, it triggers dynamic server-side filtering. ApsaraMQ for RocketMQ matches messages against the filter expression that is reported by the consumer and delivers only the matching messages. This allows the business server to receive the messages that devices report to IoT Platform.

Before you begin

The steps in this topic are demonstrated using a regular user. If an operation requires administrative permissions, run the command with sudo.

Step 1: Consume messages using the ApsaraMQ for RocketMQ SDK

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

2. Create an instance: Set the region to China (Shanghai) and the Instance Version to 5.0 Series. For more information about specifications, see Product Selection.

   

3. Obtain the instance endpoint.

   

4. Create a topic. Add a topic named iot_to_mq. The configuration is shown in the following figure.

   

5. Create a consumer group. Add a consumer group named GID_iot. The configuration is shown in the following figure.

   

6. Call the SDK to consume messages. Use the RocketMQ Java SDK to register a consumer to consume messages from the topic.

   This example uses the PushConsumer consumer type. For message consumption, you only need to use a consumption listener to process business logic and return a consumption result. The RocketMQ SDK handles message retrieval, consumption status submission, and consumption retries.

   The following code provides an example:

   1. In IntelliJ IDEA, create a Java project.

   2. In the pom.xml file, add the following dependency to import the Java dependency library.

      <dependency>
          <groupId>org.apache.rocketmq</groupId>
          <artifactId>rocketmq-client-java</artifactId>
          <version>5.0.4</version>
      </dependency>

   3. In the Java project, create and run a program for the server to subscribe to normal messages.

      Note

      In a real-world development scenario, replace the values of endpoints, topic, and consumerGroup in the code with the actual values that you obtained and configured in the previous steps.

      import org.apache.rocketmq.client.apis.*;
      import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
      import org.apache.rocketmq.client.apis.consumer.FilterExpression;
      import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
      import org.apache.rocketmq.client.apis.consumer.PushConsumer;
      import org.apache.rocketmq.shaded.org.slf4j.Logger;
      import org.apache.rocketmq.shaded.org.slf4j.LoggerFactory;
      
      import java.io.IOException;
      import java.util.Collections;
      
      public class PushConsumerExample {
          private static final Logger LOGGER = LoggerFactory.getLogger(PushConsumerExample.class);
      
          private PushConsumerExample() {
          }
      
          public static void main(String[] args) throws ClientException, IOException, InterruptedException {
           /**
               * The endpoint of the instance. You can obtain the endpoint from the Endpoints tab on the instance details page in the console.
               * If you access the instance from an ECS instance over the internal network, we recommend that you use the VPC endpoint.
               * If you access the instance from a local computer or an on-premises data center over the internet, you can use the public endpoint. To use the public endpoint, you must enable public network access for the instance.
               */
              String endpoints = "rmq-cn-xxx.{regionId}.rmq.aliyuncs.com:8080";
              // Specify the topic that you want to subscribe to. You must create the topic in the console in advance. Otherwise, an error is returned.
              String topic = "iot_to_mq";
              // Specify the consumer group for the consumer. You must create the group in the console in advance. Otherwise, an error is returned.
              String consumerGroup = "GID_iot";
              final ClientServiceProvider provider = ClientServiceProvider.loadService();
              ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoints);
          /**
               * If you use the public endpoint, you must also set the username and password of the instance for the configuration. You can obtain the username and password from the instance details page in the console.
               * If you access the instance from an Alibaba Cloud ECS instance over the internal network, you do not need to configure these parameters. The server-side intelligently obtains the information based on the VPC.
               */
              //builder.setCredentialProvider(new StaticSessionCredentialsProvider("Instance UserName", "Instance Password"));        
              ClientConfiguration clientConfiguration = builder.build();
              // The filtering rule for message subscription. This indicates that all messages with the specified tag are subscribed.
              String tag = "cn-shanghai";
              FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
              // Initialize PushConsumer. You must bind the consumer group, communication parameters, and subscription relationship.
              PushConsumer pushConsumer = provider.newPushConsumerBuilder()
                      .setClientConfiguration(clientConfiguration)
                      // Set the consumer group.
                      .setConsumerGroup(consumerGroup)
                      // Set the pre-bound subscription relationship.
                      .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                      // Set the message listener.
                      .setMessageListener(messageView -> {
                          // Process the message and return the consumption result.
                          //LOGGER.info("Consume message={}", messageView);
                          System.out.println("Consume Message: " + messageView);
                          return ConsumeResult.SUCCESS;
                      })
                      .build();
              Thread.sleep(Long.MAX_VALUE);
              // If you no longer need to use PushConsumer, you can close the process.
              //pushConsumer.close();
          }
      }                                                 

Step 3: View operational logs for device message forwarding

1. Return to the IoT Platform console, click the target Enterprise instance, and view the device status and data forwarding logs.

   1. In the navigation pane on the left, choose Device Management > Devices. Find the MQdevice device and check its status. The Online status indicates that the device is connected to IoT Platform.

      

   2. In the navigation pane on the left, choose Monitoring > Simple Log Service. Select the MQ_test product and view the data forwarding logs.

      Copy the MQ msgId from the message content. In the next step, you can use this ID to view the device message that was forwarded to ApsaraMQ for RocketMQ and its consumption status.

      

2. Return to the terminal where the ApsaraMQ for RocketMQ resource subscription is running and view the logs for message subscription and consumption.

   

3. Return to the ApsaraMQ for RocketMQ console. On the Instance Details page, view the content of the message that is received by the consumer and the message trace.

   1. Click Message Query. In the Message ID field, enter the MQ msgId value that you copied, and then click Query.

      

   2. In the Actions column for the message, click Details. In the Message Details panel that appears on the right, view the Message Body.

      

   3. Click Close in the Message Details panel to return to the message list. In the Actions column, click Message Trace to view the consumption status of the device message.