通过RocketMQ客户端消费设备消息

场景说明

假设某产品下设备上报到物联网平台的自定义Topic数据为地理位置（RID）和消息（CusMsg），需要将RID=cn-shanghai的设备消息CusMsg流转到业务服务器中使用。如果当前产品下设备与物联网平台之间的上下行消息量大于1,000 QPS，推荐使用RocketMQ通过Tag标签过滤功能，获取筛选的设备消息。

数据流转消费流程如下：

1. 在业务服务器中使用RocketMQ SDK注册一个消费者，用于接收物联网平台转发到RocketMQ中的消息。

2. 先在物联网平台配置设备消息转发到RocketMQ，为设备消息设置属性和标签，用于后续消费者消费时指定过滤条件。然后将设备接入物联网平台并上报数据，数据会流转到RocketMQ。

3. RocketMQ SDK注册的消费者获取消息时会触发服务端的动态过滤计算，RocketMQ根据该消费者上报的过滤条件的表达式进行匹配，并将符合条件的消息投递给该消费者，实现消费者所属业务服务器接收设备上报至物联网平台的消息。

使用前必读

本文操作步骤以普通用户权限为例。如果您在操作过程中涉及到管理员权限才能执行的操作，可尝试使用sudo命令执行。

步骤一：调用RocketMQ SDK消费消息

1. 登录云消息队列 RocketMQ 版控制台，在左侧导航栏单击实例列表。

2. 创建实例：选择地域为华东2（上海），选择实例版本为5.0系列。具体规格含义和取值，请参见产品选型。

   

3. 获取实例接入点。

   

4. 创建Topic：添加Topic为iot_to_mq，配置如下图所示。

   

5. 创建ConsumerGroup：添加消息组为GID_iot，配置如下图所示。

   

6. 调用SDK消费消息：使用RocketMQ的Java SDK注册一个消费者来消费Topic消息。

   本示例使用PushConsumer消费者类型，消费消息仅通过消费监听器处理业务并返回消费结果。消息的获取、消费状态提交以及消费重试都通过RocketMQ的SDK完成。

   示例代码如下：

   1. 在IDEA中创建一个Java工程。

   2. 在pom.xml文件中添加以下依赖引入Java依赖库。

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

   3. 在已创建的Java工程中，创建用户服务器订阅普通消息程序并运行。

      说明

      实际开发场景中，需替换代码中endpoints、topic和consumerGroup的值为上述步骤中获取和配置的真实值。

      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 {
           /**
               * 实例接入点，从控制台实例详情页的接入点页签中获取。
               * 如果是在阿里云ECS内网访问，建议填写VPC接入点。
               * 如果是在本地公网访问，或者是线下IDC环境访问，可以使用公网接入点。使用公网接入点访问，必须开启实例的公网访问功能。
               */
              String endpoints = "rmq-cn-xxx.{regionId}.rmq.aliyuncs.com:8080";
              //指定需要订阅哪个目标Topic，Topic需要提前在控制台创建，如果不创建直接使用会返回报错。
              String topic = "iot_to_mq";
              //为消费者指定所属的消费者分组，Group需要提前在控制台创建，如果不创建直接使用会返回报错。
              String consumerGroup = "GID_iot";
              final ClientServiceProvider provider = ClientServiceProvider.loadService();
              ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoints);
          /**
               * 如果是使用公网接入点访问，configuration还需要设置实例的用户名和密码。用户名和密码在控制台实例详情页获取。
               * 如果是在阿里云ECS内网访问，无需填写该配置，服务端会根据内网VPC信息智能获取。
               */
              //builder.setCredentialProvider(new StaticSessionCredentialsProvider("Instance UserName", "Instance Password"));        
              ClientConfiguration clientConfiguration = builder.build();
              //订阅消息的过滤规则，表示订阅所有Tag的消息。
              String tag = "cn-shanghai";
              FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
              //初始化PushConsumer，需要绑定消费者分组ConsumerGroup、通信参数以及订阅关系。
              PushConsumer pushConsumer = provider.newPushConsumerBuilder()
                      .setClientConfiguration(clientConfiguration)
                      //设置消费者分组。
                      .setConsumerGroup(consumerGroup)
                      //设置预绑定的订阅关系。
                      .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                      //设置消费监听器。
                      .setMessageListener(messageView -> {
                          // 处理消息并返回消费结果。
                          //LOGGER.info("Consume message={}", messageView);
                          System.out.println("Consume Message: " + messageView);
                          return ConsumeResult.SUCCESS;
                      })
                      .build();
              Thread.sleep(Long.MAX_VALUE);
              //如果不需要再使用PushConsumer，可关闭该进程。
              //pushConsumer.close();
          }
      }                                                 

步骤三：查看设备消息转发的运行日志

1. 返回物联网平台控制台，单击目标企业版实例，查看设备状态和消息转发日志。

   1. 左侧导航栏选择设备管理 > 设备，找到目标设备MQdevice，查看设备状态。设备状态显示为在线，则表示设备与物联网平台成功连接。

      

   2. 左侧导航栏选择监控运维 > 日志服务，选择目标产品MQ_test，查看消息转发日志。

      您可复制消息内容中的MQ msgId，执行下一步查看转发到RocketMQ的设备消息和消费消息状态。

      

2. 返回已运行的订阅RocketMQ资源的终端，查看订阅和消费消息的日志。

   

3. 返回云消息队列RocketMQ版控制台，在实例详情页面，查看消费者接收到消息的具体内容和消息轨迹。

   1. 单击消息查询，在Message ID输入框中输入已复制的MQ msgId值，单击查询。

      

   2. 单击消息对应操作列的详情，可在右侧消息详情面板，查看到具体的消息体。

      

   3. 单击消息详情面板的关闭，返回消息列表，单击操作列的消息轨迹，可查看到设备消息消费状态。