工作原理

云產品流轉可將設備消息轉發到RocketMQ SDK客戶端消費。上圖工作流程：

1. 業務服務器：使用RocketMQ SDK注冊一個消費者，用于接收物聯網平臺轉發到RocketMQ中的消息。

2. 物聯網平臺：配置云產品流轉將設備消息轉發到RocketMQ。

   1. 數據源：支持的Topic類型消息，請參見數據格式（非云網關產品和設備）、自定義Topic（MQTT云網關）、消息轉發Topic（GB/T 32960云網關）、消息轉發Topic（JT/T 808云網關）、消息轉發Topic（SL 651云網關）。

   2. 數據目的：RocketMQ中接收設備消息的Topic，需要先在云消息隊列RocketMQ版控制臺創建。具體操作，請參見創建Topic。

   3. 解析器腳本：配置通過數據流轉函數writeMq(destinationId, payload, tag)將設備數據轉發到RocketMQ的Topic中。

      函數說明，請參見流轉數據到數據目的函數。

      您可為設備消息設置屬性和標簽，用于后續消費者消費時指定過濾條件。

3. 云消息隊列RocketMQ版：RocketMQ SDK注冊的消費者獲取消息時會觸發服務端的動態過濾計算，RocketMQ根據該消費者上報的過濾條件的表達式進行匹配，并將符合條件的消息投遞給該消費者，實現消費者所屬業務服務器接收設備上報至物聯網平臺的消息。

應用場景

業務服務器接收設備消息：云產品流轉可以靈活地轉發設備消息到消息隊列RocketMQ中消費。

• 對設備數據進行復雜或精細化處理的海量設備場景。

• 設備消息量大于1,000 QPS的場景。

使用限制

• 物聯網平臺實例及所在地域支持將數據轉發到消息隊列RocketMQ。支持的地域詳細信息，請參見各地域功能說明。

• 目前，RocketMQ實例所在地域必須與當前物聯網平臺實例所在地域保持一致。存量的跨地域實例下數據流轉配置不受影響，仍可繼續正常流轉數據。

• 僅支持將數據流轉到RocketMQ 4.x或5.x版本實例的Topic中。如果是RocketMQ 5.x版本實例的Topic，該Topic的消息類型必須為普通消息。

• 目前，新版和舊版云產品流轉功能均支持將數據流轉到消息隊列RocketMQ。舊版云產品流轉使用示例，請參見數據轉發到消息隊列RocketM（舊版）。

背景信息

當使用RocketMQ 5.0時，轉發的數據目的配置完成后，會自動完成以下配置，實現設備數據通過物聯網平臺的規則引擎轉發到消息隊列RocketMQ。

• 物聯網平臺占用RocketMQ實例所在虛擬交換機的2個IP地址。

• 在RocketMQ實例所在的VPC網絡下創建托管安全組，安全組名稱默認以sg-nsm-開頭。

步驟一：創建數據目的

1. 登錄物聯網平臺控制臺。

2. 在實例概覽頁簽的全部環境下，找到對應的實例，單擊實例卡片。

3. 在左側導航欄，選擇消息轉發 > 云產品流轉。

4. 在云產品流轉頁面，單擊右上角體驗新版，進入新版功能頁面。

   說明

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

5. 單擊數據目的頁簽，然后單擊創建數據目的。

6. 在創建數據目的對話框，輸入數據目的名稱，例如DataPurpose，按照以下參數說明，完成配置，然后單擊確定。

   

   參數	說明
   選擇操作	選擇發送數據到消息隊列（RocketMQ）中。
   地域	選擇RocketMQ所在地域。
   實例	選擇RocketMQ實例。 您可以單擊創建實例，跳轉到消息隊列控制臺，創建RocketMQ實例，請參見消息隊列文檔。
   Topic	選擇用于接收物聯網平臺數據的RocketMQ Topic。 您可以單擊創建Topic，跳轉到消息隊列控制臺，創建RocketMQ Topic。
   授權	授權物聯網平臺將數據寫入RocketMQ。 如您還未創建相關角色，單擊創建RAM角色，跳轉到RAM控制臺，創建角色和授權策略，請參見創建RAM角色。

步驟二：配置并啟動解析器

1. 創建解析器，例如DataParser。具體操作，請參見步驟一：創建解析器。

2. 在解析器詳情頁面，關聯數據源。

   1. 在配置向導的數據源下，單擊關聯數據源。

   2. 在彈出的對話框中，單擊數據源下拉列表，選擇已創建的數據源DataSource，單擊確定。

3. 在解析器詳情頁面，關聯數據目的。

   1. 單擊配置向導的數據目的，然后單擊數據目的列表右上方的關聯數據目的。

   2. 在彈出的對話框中，單擊數據目的下拉列表，選擇已創建的數據目的DataPurpose，單擊確定。

   3. 在數據目的列表，查看并保存數據目的ID，例如為1000。

      后續解析腳本中，需使用此處的數據目的ID。

4. 在解析器詳情頁面，單擊解析器。

5. 在腳本輸入框，輸入解析腳本。

   函數參數說明，請參見函數列表。

   //通過payload函數，獲取設備上報的消息內容，并按照JSON格式轉換。
   var data = payload("json");
   //直接流轉物模型上報數據。
   writeMq(1000, data, "調試");

6. 單擊調試，根據頁面提示，選擇產品和設備，輸入Topic和Payload數據，驗證腳本可執行。

   參數示例如下：

   運行結果如下，表示腳本執行成功。

   

7. 單擊發布。

8. 回到云產品流轉頁面的解析器頁簽，單擊解析器DataParser對應的啟動按鈕，啟動解析器。

9. 在RocketMQ控制臺，查看是否成功接收到消息。

步驟三：查看設備消息轉發的運行日志

1. 在物聯網平臺控制臺，單擊目標企業版實例，查看設備狀態和消息轉發日志。具體操作，請參見云端運行日志。

2. 在業務服務器中已運行的訂閱RocketMQ資源的終端，查看訂閱和消費消息的日志。

3. 在云消息隊列RocketMQ版控制臺，在實例詳情頁面，查看消費者接收到消息的具體內容和消息軌跡。具體操作，請參見查詢消息軌跡。

操作樣例

通過RocketMQ客戶端消費設備消息

常見問題

• 云消息隊列 RocketMQ 版5.0系列相比4.0系列實例有什么優勢？

• 使用云消息隊列RocketMQ 5.x系列的常見問題

• 使用云消息隊列RocketMQ 4.x系列的常見問題