使用Kafka用戶端消費訂閱資料

注意事項

• 使用本文提供的Demo消費資料時，如果採用auto commit（自動認可），可能會因為資料還沒被消費完就執行了提交操作，從而丟失部分資料，建議採用手動提交的方式以避免該問題。

  說明

  如果發生故障沒有提交成功，重啟用戶端後會從上一個記錄的位點進行資料消費，期間會有部分重複資料，您需要手動過濾。

• 資料以Avro序列化儲存，詳細格式請參見Record.avsc文檔。

  警告

  如果您使用的不是本文提供的Kafka用戶端，在進行還原序列化解析時，可能出現解析的資料有誤，您需要自行驗證資料的正確性。

• 關於offsetForTimes介面，DTS的搜尋單位為秒，原生Kafka的搜尋單位為毫秒。

• 由於資料訂閱服務端會因容災等原因導致網路閃斷，若您未使用本文提供的Kafka用戶端，您使用的Kafka用戶端需具備網路重試能力。

• 若您使用原生的Kafka用戶端消費訂閱資料，則可能會在DTS發生增量資料擷取模組切換行為，從而使subscribe模式下訂閱用戶端儲存在服務端的消費位點被清除，您需要手動調整訂閱的消費位點以實現按需消費資料。

Kafka用戶端運行流程說明

請下載Kafka用戶端Demo代碼。更多關於代碼使用的詳細介紹，請參見Demo中的Readme文檔。

操作步驟

本文以IntelliJ IDEA軟體（Community Edition 2018.1.4 Windows版本）為例，介紹如何運行該用戶端消費訂閱通道中的資料。

1. 建立新版資料訂閱通道，詳情請參見建立RDS MySQL資料訂閱通道、建立PolarDB MySQL版資料訂閱通道或建立Oracle資料訂閱通道。

2. 建立一個或多個消費組，詳情請參見新增消費組。

3. 下載Kafka用戶端Demo代碼，然後解壓該檔案。

   說明

   單擊，然後選擇Download ZIP下載檔案。

4. 開啟IntelliJ IDEA軟體，然後單擊Open。

   

5. 在彈出的對話方塊中，定位至Kafka用戶端Demo代碼下載的目錄，參照下圖依次展開檔案夾，找到專案物件模型檔案：pom.xml。

   

6. 在彈出對話方塊中，選擇Open as Project。

7. 在IntelliJ IDEA軟體介面，依次展開檔案夾，找到並雙擊開啟Kafka用戶端Demo檔案：NotifyDemoDB.java。

8. 設定NotifyDemoDB.java檔案中的各參數對應的值。

   

   參數	說明	擷取方式

   參數	說明	擷取方式
   USER_NAME	消費組的帳號。 警告 如您未使用本文提供的用戶端，請按照<消費組的帳號>-<消費組ID>的格式設定使用者名稱（例如：dtstest-dtsae******bpv），否則無法正常串連。	在DTS控制台單擊目標訂閱執行個體ID，然後單擊資料消費，您可以擷取到消費組ID和消費組的帳號資訊。 說明 消費組帳號的密碼已在您建立消費組時指定。
   PASSWORD_NAME	該帳號的密碼。
   SID_NAME	消費組ID。
   GROUP_NAME	消費組名稱，需保持和消費組ID相同（即本參數也填入消費組ID）。
   KAFKA_TOPIC	資料訂閱通道的訂閱Topic。	在DTS控制台單擊目標訂閱執行個體ID，在任務管理頁面，您可以擷取到訂閱Topic、網路地址資訊。
   KAFKA_BROKER_URL_NAME	資料訂閱通道的網路地址資訊。 說明 如果您部署Kafka用戶端所屬的ECS執行個體與資料訂閱通道屬於傳統網路或同一專用網路，建議通過內網地址進行資料訂閱，網路延遲最小。 不建議使用公網地址。
   INITIAL_CHECKPOINT_NAME	消費的資料時間點，格式為Unix時間戳記，例如1592269238。 說明 您需要自行儲存時間點資訊，以便： 當業務程式中斷後，傳入已消費的資料時間點繼續消費資料，防止資料丟失。 在訂閱用戶端啟動時，傳入所需的消費位點，調整訂閱位點，實現按需消費資料。 若SUBSCRIBE_MODE_NAME為subscribe時，傳入的INITIAL_CHECKPOINT_NAME僅在訂閱用戶端初次開機時生效。	消費的資料時間點必須在訂閱執行個體的資料範圍（如圖示）之內，並需轉化為Unix時間戳記。 說明 Unix時間戳記轉換工具可用搜尋引擎擷取。
   USE_CONFIG_CHECKPOINT_NAME	預設取值為true，即強制使用指定的資料時間點來消費資料，避免丟失已接收到的但未處理的資料。	無
   SUBSCRIBE_MODE_NAME	一個消費組下支援同時啟動兩個及以上Kafka用戶端，如需實現該功能，請將所有用戶端該參數的值設定為subscribe。 預設值為assign，即不使用該功能，建議只部署一個用戶端。	無

9. 在IntelliJ IDEA軟體介面的頂部，選擇Run > Run運行該用戶端。

   說明

   首次運行時，軟體需要一定時間自動載入相關依賴包並完成安裝。

執行結果

運行結果如下圖所示，該用戶端可正常訂閱到源庫的資料變更資訊。

您也可以去除NotifyDemoDB.java檔案中的列印日誌詳情的注釋（即刪除第25行//log.info(ret);中的//），然後再次運行該用戶端即可查看詳細的資料變更資訊。

常見問題

• Q：為什麼需要自行記錄用戶端的消費位點？

  A：由於DTS記錄的消費位點是接收到Kafka消費用戶端執行commit操作的時間點，可能與當前實際消費到的時間點存在一定的時間差。當業務程式或Kafka消費用戶端異常中斷後，您可以傳入自行記錄的消費位點以繼續消費，避免消費到重複的資料或缺失部分資料。

MySQL欄位類型與dataTypeNumber數值的對應關係

詳情請參見SQL Type field。

Oracle欄位類型與dataTypeNumber數值的對應關係

PostgreSQL欄位類型與dataTypeNumber數值的對應關係