幫助中心
首頁 IoT Platform 實踐教程 最佳實務 訊息通訊 通過RocketMQ用戶端消費裝置訊息

通過RocketMQ用戶端消費裝置訊息

更新時間：
Copy as MD
產品

在物聯網應用情境中，如果需要將裝置上報至物聯網平台的資料轉寄至您商務服務器的應用中進行消費（例如過濾、分析、儲存等），可以使用物聯網平台提供的雲產品流轉服務，例如通過雲訊息佇列RocketMQ版（以下簡稱RocketMQ）實現裝置資料流轉並消費。RocketMQ提供多語言用戶端SDK實現訊息收發，本文介紹如何使用RocketMQ的Java SDK來接收物聯網平台的裝置訊息。

情境說明

假設某產品下裝置上報到物聯網平台的自訂Topic資料為地理位置（RID）和訊息（CusMsg），需要將RID=cn-shanghai的裝置訊息CusMsg流轉到商務服務器中使用。如果當前產品下裝置與物聯網平台之間的上下行訊息量大於1,000 QPS，推薦使用RocketMQ通過Tag標籤過濾功能，擷取篩選的裝置訊息。

  1. 在商務服務器中使用RocketMQ SDK註冊一個消費者，用於接收物聯網平台轉寄到RocketMQ中的訊息。

  2. 先在物聯網平台配置裝置訊息轉寄到RocketMQ，為裝置訊息設定屬性和標籤，用於後續消費者消費時指定過濾條件。然後將裝置接入物聯網平台並上報資料，資料會流轉到RocketMQ。

  3. RocketMQ SDK註冊的消費者擷取訊息時會觸發服務端的動態過濾計算，RocketMQ根據該消費者上報的過濾條件的運算式進行匹配，並將合格訊息投遞給該消費者，實現消費者所屬商務服務器接收裝置上報至物聯網平台的訊息。

前提條件

  • 登入阿里雲帳號。

  • 已開通物聯網平台服務。

  • 已購買企業版執行個體：本樣本購買華東2（上海）地區的標準型企業版執行個體。

  • 已開通雲訊息佇列RocketMQ版服務。

    如未開通，請登入雲訊息佇列RocketMQ版產品頁面，開通服務。

  • 已準備開發環境。

    • 本樣本以Java SDK為例介紹使用RocketMQ實現訊息消費。

    • 本樣本購買Alibaba Cloud Linux作業系統的ECS執行個體（預設已具備C語言編譯和運行能力），使用物聯網平台提供的裝置C Link SDK Demo，類比實際裝置接入物聯網平台並上報資料。購買ECS執行個體的方法，請參見建立執行個體。

背景資訊

物聯網平台提供服務端訂閱和雲產品流轉服務，均可實現裝置資料流轉。您可對比物聯網平台支援的流轉方案及應用情境，選擇您業務需要的合適方案。具體內容，請參見資料流轉方案對比。

本文方案優勢：

  • 裝置使用MQTT協議接入物聯網平台，資料轉送鏈路支援TLS加密，保障資料不被篡改。MQTT協議說明，請參見MQTT協議規範。

  • 通過RocketMQ削峰填穀，緩衝訊息，減輕伺服器同時接收大量裝置訊息的壓力。RocketMQ產品優勢和應用情境，請參見雲訊息佇列RocketMQ版產品介紹。

使用前必讀

本文操作步驟以普通使用者權限為例。如果您在操作過程中涉及到管理員權限才能執行的操作，可嘗試使用sudo命令執行。

步驟一：調用RocketMQ SDK消費訊息

說明

  • RocketMQ執行個體所在地區必須與物聯網平台執行個體所在地區保持一致。本樣本以華東2（上海）地區為例進行介紹。

  • 物聯網平台支援將裝置資料流轉到RocketMQ 4.x和5.x版本執行個體的Topic中。本樣本以資料流轉到RocketMQ 5.x版本為例進行介紹。

  1. 登入雲訊息佇列 RocketMQ 版控制台，在左側導覽列單擊執行個體列表。

  2. 建立執行個體：選擇地區為華東2（上海），選擇執行個體版本為5.0系列。具體規格含義和取值，請參見產品選型。

    選擇主系列為專業版，子系列為叢集高可用版（生產環境推薦），訊息收發計算規格為rmq.s2.2xlarge。配置VPC ID和VSwitch ID，將公網訪問設定為開啟，公網計費類型為固定頻寬計費，公網頻寬規格設為1 Mbps。

  3. 擷取執行個體存取點。

    在存取點資訊頁面，用戶端協議為 TCP 時，可擷取兩種網路類型的存取點：VPC專用網路存取點（格式如 rmq-c<執行個體ID>.cn-shanghai.rmq.aliyuncs.com:8080）和公網存取點。公網存取點支援通過編輯白名單配置網路存取控制。

  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. 登入物聯網平台控制台。

    2. 在頂部功能表列選擇地區：華東2（上海）。

    3. 在執行個體概覽頁簽的全部環境下，找到對應的執行個體，單擊執行個體卡片。

    4. 在左側導覽列選擇裝置管理 > 產品，單擊建立產品，配置參數，單擊確認。具體說明，請參見建立產品。

      本樣本中，產品名稱為MQ_test，節點類型為直連裝置，其他參數使用預設值。

    5. 單擊查看產品詳情，在產品詳情頁面，單擊Topic類列表 > 自訂Topic，然後單擊自訂Topic類，定義一個具有發布許可權的Topic類，用於裝置上報資料。具體說明，請參見添加自訂Topic類。

      本樣本中，定義的Topic類：/${YourProductKey}/${YourDeviceName}/user/data。

    6. 在左側導覽列選擇裝置管理 > 裝置，單擊添加裝置，為產品MQ_test建立裝置。具體說明，請參見單個建立裝置。

      本樣本中，建立了一個名稱為MQdevice的裝置。建立成功後，複製裝置認證並儲存。

  2. 配置資料轉寄到RocketMQ。

    1. 在左側導覽列選擇訊息轉寄 > 雲產品流轉。

    2. 在雲產品流轉頁面，單擊右上方體驗新版，進入新版功能頁面。

      說明

      如果您已執行過此操作，再次進入雲產品流轉頁面，會直接進入新版功能頁面。

    3. 添加裝置MQdevice的自訂Topic為待流轉的資料來源。

      1. 單擊資料來源頁簽，然後單擊建立資料來源，輸入資料來源名稱（例如：MQdata）和描述，單擊確定。

      2. 在資料來源詳情頁面，單擊添加Topic，在彈出對話方塊中，選擇需要處理的訊息Topic：/${YourProductKey}/${YourDeviceName}/user/data，然後單擊確定。

    4. 配置資料轉寄的資料目的。

      1. 單擊資料目的頁簽，然後單擊建立資料目的。

      2. 在建立資料目的對話方塊，配置資料目的到已建立RocketMQ執行個體下的目標Topic：iot_to_mq，單擊確定。具體說明，請參見資料轉寄到雲訊息佇列RocketMQ版。

      本樣本中，資料目的名稱設定為ToRocketMQ，選擇操作選擇發送資料到訊息佇列（RocketMQ）中，地區選擇華東2（上海），執行個體選擇IoT-forward-data-test，Topic選擇iot_to_mq，授權選擇AliyunIOTAccessingMQRole。

    5. 步驟一：建立解析器：建立解析器：DataParser。

    6. 在解析器詳情頁面，關聯資料來源和資料目的。

      1. 在設定精靈的資料來源下，單擊關聯資料來源。

      2. 在彈出的對話方塊中，單擊資料來源下拉式清單，選擇已建立的資料來源MQdata，單擊確定。

      3. 單擊設定精靈的資料目的，然後單擊資料目的列表右上方的關聯資料目的。

      4. 在彈出的對話方塊中，單擊資料目的下拉式清單，選擇已建立的資料目的ToRocketMQ，單擊確定。

        在資料目的列表，查看並儲存資料目的ID，例如為1000。

    7. 配置解析器指令碼。

      1. 單擊設定精靈的解析指令碼，然後單擊編輯草稿。

      2. 在指令碼輸入框，輸入解析指令碼。

        本樣本使用函數writeMq(destinationId, payload, tag)流轉資料。函數參數說明，請參見函數列表。指令碼代碼如下：

        // 裝置上報資料內容，json格式
var data = payload("json");
var tag= data.RID;
var msg= data.CusMsg;
// 流轉MQ
writeMq(1000, msg, tag);

      3. 單擊調試，根據頁面提示，選擇產品和裝置，輸入Topic和Payload資料，驗證指令碼可執行。

        產品選擇MQ_test，裝置選擇MQdevice，Topic輸入/${ProductKey}/MQdevice/user/data，Payload資料輸入{"CusMsg": "Custom message forwarding test", "RID":"cn-shanghai"}。

        運行結果如下，表示指令碼執行成功。

        action:
  transmit to MQ[destinationId=1000], data:Custom messag
variables:
  msg : Custom message forwarding test
  data : {"CusMsg":"Custom message forwarding test","RID":"cn-shanghai"}
  tag : cn-shanghai

      4. 單擊發布。

    8. 回到雲產品流轉頁面的解析器頁簽，單擊解析器DataParser對應操作列的啟動，啟動解析器。

  3. 裝置接入物聯網平台並上報資料。

    1. 擷取裝置端C Link SDK。

      1. 擷取C Link SDK，然後重新命名為LinkSDK.zip。

      2. 登入ECS執行個體。登入方式，請參見選擇ECS遠端連線方式。

      3. 上傳LinkSDK.zip檔案到ECS執行個體。

      4. 執行以下命令，安裝unzip。

        yum update
yum install zip

      5. 進入LinkSDK.zip檔案所在目錄後，執行以下命令，解壓檔案。

        unzip LinkSDK.zip

      LinkSDK檔案夾的內容說明，請參見檔案說明表。本樣本使用LinkSDK檔案下非雲網關裝置接入的Demo檔案：./mqtt_basic_demo.c。

    2. 配置裝置範例程式並運行。

      本樣本僅說明修改的代碼內容。裝置程式開發的詳細說明，請參見MQTT接入使用。

      1. 開啟/LinkSDK/demos路徑下的mqtt_basic_demo.c檔案，配置裝置接入認證的參數。

        /* TODO: 替換為自己裝置的裝置認證 */
char *product_key       = "k0******";
char *device_name       = "MQdevice";
char *device_secret     = "8c684ef*************";
......
char  *mqtt_host = "iot-cn-******.mqtt.iothub.aliyuncs.com";

        參數

        樣本

        說明

        product_key

        k0******

        裝置認證資訊。即完成添加裝置後，您儲存至本地的裝置認證。

        您也可以在物聯網平台的裝置詳情頁查看裝置的認證資訊。

        device_name

        MQdevice

        device_secret

        8c684ef*************

        mqtt_host

        iot-cn-******.mqtt.iothub.aliyuncs.com

        裝置的MQTT接入網域名稱。在執行個體詳情頁面單擊右上方的查看開發配置，在開發配置面板查看接入網域名稱。

        執行個體的更多詳細說明，請參見查看和配置執行個體終端節點資訊（Endpoint）。

      2. 取消以下代碼的注釋符號，調用aiot_mqtt_pub，向指定的Topic：/k0******/MQdevice/user/data發送訊息。

        /* MQTT 發布訊息功能樣本, 請根據自己的業務需求進行使用 */
    {
        char *pub_topic = "/k0******/MQdevice/user/data";
        char *pub_payload = "{\"CusMsg\": \"Custom message forwarding test\",\"RID\": \"cn-shanghai\"}";
        res = aiot_mqtt_pub(mqtt_handle, pub_topic, (uint8_t *)pub_payload, (uint32_t)strlen(pub_payload), 0);
        if (res < 0) {
            printf("aiot_mqtt_sub failed, res: -0x%04X\n", -res);
            return -1;
        }
    }

      3. 儲存mqtt_basic_demo.c檔案後，在SDK根目錄/LinkSDK下，執行以下命令，完成裝置範例程式的編譯。

        make clean
make

        產生的範例程式data-basic-demo存放在./output目錄下。

      4. 執行以下命令，運行範例程式。

        ./output/data-basic-demo

        返回如下日誌，裝置接入並發布訊息成功：

        [1698060995.066][LK-0313] MQTT user calls aiot_mqtt_connect api, connect
[1698060995.066][LK-032A] mqtt host: iot-******.mqtt.iothub.aliyuncs.com
[1698060995.066][LK-0317] user name: MQdevice&k09******
establish tcp connection with server(host='iot-0******.mqtt.iothub.aliyuncs.com', port=[443])
success to establish tcp, fd=3
local port: 33626
[1698060995.066][LK-1000] establish mbedtls connection with server(host='iot-0******.mqtt.iothub.aliyuncs.com', port=[443])
[1698060995.122][LK-1000] success to establish mbedtls connection, (cost 45376 bytes in total, max used 48344 bytes)
[1698060995.155][LK-0313] MQTT connect success in 93 ms
AIOT_MQTTEVT_CONNECT
[1698060995.155][LK-0309] pub: /k09j9******/MQdevice/user/data
[LK-030A] > 7B 22 43 75 73 4D 73 67  22 3A 20 22 43 75 73 74 | {"CusMsg": "Cust
[LK-030A] > 6F 6D 20 6D 65 73 73 61  67 65 20 66 6F 72 77 61 | om message forwa
[LK-030A] > 72 64 69 6E 67 20 74 65  73 74 22 2C 22 52 49 44 | rding test","RID
[LK-030A] > 22 3A 20 22 63 6E 2D 73  68 61 6E 67 68 61 69 22 | ": "cn-shanghai"
[LK-030A] > 7D                                               | }               
heartbeat response

步驟三：查看裝置訊息轉寄的作業記錄

  1. 返回物聯網平台控制台，單擊目標企業版執行個體，查看裝置狀態和訊息轉寄日誌。

    1. 左側導覽列選擇裝置管理 > 裝置，找到目標裝置MQdevice，查看裝置狀態。裝置狀態顯示為線上，則表示裝置與物聯網平台成功串連。

    2. 左側導覽列選擇監控營運 > Log Service，選擇目標產品MQ_test，查看訊息轉寄日誌。

      您可複製訊息內容中的MQ msgId，執行下一步查看轉寄到RocketMQ的裝置訊息和消費訊息狀態。

      日誌列表中，業務類型為雲產品流轉的記錄表示資料已成功轉寄至MQ。單擊訊息內容列的查看連結，可在彈出的詳情中查看完整JSON，其中包含 MQ msgId 欄位值。

  2. 返回已啟動並執行訂閱RocketMQ資源的終端，查看訂閱和消費訊息的日誌。

    Consume Message: MessageViewImpl{messageId=xxx, topic=iot_to_mq, bornHost=iotx-data-xxx, bornTimestamp=xxx, endpoints=dns:rmq-cn-xxx.cn-shanghai.rmq.aliyuncs.com:8081, deliveryAttempt=1, tag=cn-shanghai, keys=[0], messageGroup=null, deliveryTimestamp=null, properties={}}

  3. 返回雲訊息佇列RocketMQ版控制台，在執行個體詳情頁面，查看消費者接收到訊息的具體內容和訊息軌跡。

    1. 單擊訊息查詢，在Message ID輸入框中輸入已複製的MQ msgId值，單擊查詢。

      查詢時 Topic 選擇 iot_to_mq，查詢結果表格中顯示一條訊息記錄，確認裝置訊息已成功轉寄至訊息佇列 RocketMQ。

    2. 單擊訊息對應操作列的詳情，可在右側訊息詳情面板，查看到具體的訊息體。

      其中，訊息標籤(Tags) 為 cn-shanghai，訊息體 內容為 Custom message forwarding test，表明 IoT 訊息已成功轉寄至雲訊息佇列 RocketMQ 版。

    3. 單擊訊息詳情面板的關閉，返回訊息列表，單擊操作列的訊息軌跡，可查看到裝置訊息消費狀態。

      訊息軌跡顯示訊息投遞鏈路：生產者發送成功，訊息類型為普通訊息，Topic 為 iot_to_mq，Tag 為 cn-shanghai；消費組 GID_iot 共投遞 1 次，消費成功。

上一篇：基於Topic訊息路由的M2M裝置間通訊下一篇：服務端訂閱（MNS）