在完成資料訂閱通道的配置（建立好訂閱任務和消費組）後，您可以使用DTS提供的SDK來消費訂閱到的資料，本文介紹範例程式碼的使用方法。
如果資料來源是PolarDB-X 1.0或DMS LogicDB，消費訂閱資料的操作步驟，請參見使用SDK消費PolarDB-X 1.0訂閱資料。
本操作為Java語言的SDK用戶端樣本，Python和Go語言的範例程式碼，請參見dts-subscribe-demo。
前提條件
已建立訂閱執行個體，且執行個體的運行狀態為正常。
說明您可以通過訂閱者案概覽，查看建立訂閱執行個體的操作文檔。
已為訂閱執行個體建立消費組。
若使用子帳號（RAM使用者）來消費訂閱到的資料，該帳號需具備AliyunDTSFullAccess許可權，以及訂閱對象的存取權限。授權方法，請參見通過系統策略授權子帳號管理DTS和管理RAM使用者的許可權。
注意事項
在消費訂閱資料時，您需要調用DefaultUserRecord的commit方法以提交位點資訊，否則會導致資料重複消費。
不同的消費之間是相互獨立的。
控制台中的當前位點表示訂閱任務當前訂閱到的位點，而非用戶端所提交的位點。
操作步驟
下載SDK範例程式碼檔案，然後解壓該檔案。
確認SDK代碼的版本。
定位至SDK範例程式碼解壓的目錄。
使用文本編輯工具開啟目錄中的pom.xml檔案。
將資料訂閱SDK的版本（version）修改為最新版本。
說明您可以在dts-new-subscribe-sdk頁面查看最新Maven依賴。
編輯SDK代碼。
使用編碼軟體開啟解壓後的檔案。
根據SDK用戶端的使用模式，開啟對應模式的Java檔案。
說明Java檔案的路徑為
aliyun-dts-subscribe-sdk-java-master/src/test/java/com/aliyun/dts/subscribe/clients/。使用模式
Java檔案
說明
適用情境
ASSIGN模式
DTSConsumerAssignDemo.java
DTS為了保證訊息的全域有序，每個訂閱Topic只有一個partition，且固定分配至partition 0中。當SDK用戶端的使用模式為ASSIGN模式時，建議只啟動一個SDK用戶端。
同一個消費組下僅有一個SDK用戶端消費訂閱資料。
SUBSCRIBE模式
DTSConsumerSubscribeDemo.java
DTS為了保證訊息的全域有序，每個訂閱Topic只有一個partition，且固定分配至partition 0中。當SDK用戶端的使用模式為SUBSCRIBE模式時，您可以在一個消費組下同時啟動多個SDK用戶端，以實現災備。實現原理是當消費組下的正常消費資料的用戶端發生故障後，其他的SDK用戶端將隨機且自動地分配到partition 0，繼續消費。
同一個消費組下同時有多個SDK用戶端消費訂閱資料，即資料災備情境。
設定Java代碼中的參數。
參數
說明
擷取方式
brokerUrl資料訂閱通道的網路地址及連接埠號碼資訊。
說明如果您部署SDK用戶端的伺服器（如ECS執行個體）與資料訂閱執行個體屬於同一專用網路，建議通過VPC網路地址進行資料消費，以減少網路延遲。
鑒於網路穩定性因素，不建議使用公網地址。
在DTS控制台單擊目標訂閱執行個體ID，在基本資料頁面的網路地區，擷取網路地址及連接埠號碼資訊。
topic資料訂閱通道的訂閱Topic。
在DTS控制台單擊目標訂閱執行個體ID，在基本資料頁面的基本資料地區，擷取到訂閱Topic。
sid消費組ID。
在DTS控制台單擊目標訂閱執行個體ID，在數據消費頁面擷取消費組ID/名稱和消費組的帳號資訊。
userName消費組的帳號。
警告如您未使用本文提供的用戶端，請按照
<消費組的帳號>-<消費組ID>的格式設定使用者名稱（例如：dtstest-dtsae******bpv），否則無法正常串連。password該帳號的密碼。
在建立消費組時設定的消費組帳號密碼。
initCheckpoint消費位點，即SDK用戶端消費第一條資料的時間戳記，格式為Unix時間戳記，例如1620962769。
說明消費位點資訊可用於：
當業務程式中斷後，傳入已消費位點繼續消費資料，防止資料丟失。
在訂閱用戶端啟動時，傳入所需的消費位點，調整訂閱位點，實現按需消費資料。
消費位點必須在訂閱執行個體的資料範圍之內，並需轉換為Unix時間戳記。
說明您可以在訂閱工作清單的數據範圍列，查看目標訂閱執行個體的資料範圍。
subscribeModeSDK用戶端的使用模式，無需修改。
ConsumerContext.ConsumerSubscribeMode.ASSIGN：ASSIGN模式。ConsumerContext.ConsumerSubscribeMode.SUBSCRIBE：SUBSCRIBE模式。
無
開啟編碼軟體的專案結構，確保此專案的OpenJDK版本為1.8。
運行該用戶端代碼。
說明代碼首次運行時，編碼軟體需要一定時間自動載入相關外掛程式和依賴項。
SDK用戶端每隔一定時間會統計並顯示消費資料的資訊，包括資料發送和接收的總數、資料總量、每秒請求數接收RPS等參數。
[2025-02-25 18:22:18.160] [INFO ] [subscribe-logMetricsReporter-1-thread-1] [log.metrics:184] - {"outCounts":0.0,"outBytes":0.0,"outRps":0.0,"outBps":0.0,"count":11.0,"inBytes":0.0,"DStoreRecordQueue":0.0,"inCounts":0.0,"inRps":0.0,"inBps":0.0,"__dt":174047893****,"DefaultUserRecordQueue":0.0}參數
說明
outCountsSDK用戶端所消費的資料總數。
outBytesSDK用戶端所消費的資料總量，單位為Byte。
outRpsSDK用戶端消費資料時的每秒請求數。
outBpsSDK用戶端消費資料時每秒傳送的位元數。
countSDK用戶端消費資料資訊（metrics）中的參數總數。
說明不包含
count本身。inBytesDTS伺服器發送的資料總量，單位為Byte。
DStoreRecordQueueDTS伺服器發送資料時，當前資料緩衝隊列的大小。
inCountsDTS伺服器發送資料總數。
inBpsDTS伺服器發送資料時每秒傳送的位元數。
inRpsDTS伺服器發送資料時的每秒請求數。
__dtSDK用戶端接收到資料的目前時間戳，單位為毫秒。
DefaultUserRecordQueue序列化後，當前資料緩衝隊列的大小。
根據您的業務需求，自行編輯代碼以消費訂閱資料。
在消費訂閱資料時，您需要管理消費位點，以確保資料不丟失，且盡量不重複，實現按需消費。
常見問題
無法串連訂閱執行個體，如何處理？
請根據報錯提示進行排查，詳情請參見異常排查。
持久化後的消費位點是什麼格式的資料？
消費位點在持久化處理後，將返回JSON格式的資料。其中，持久化後的消費位點的格式為Unix時間戳記，您可以直接將其傳回SDK進行使用。如下返回資料中，
"timestamp"後的1700709977即為持久化後的消費位點。{"groupID":"dtsglg11d48230***","streamCheckpoint":[{"partition":0,"offset":577989,"topic":"ap_southeast_1_vpc_rm_t4n22s21iysr6****_root_version2","timestamp":170070****,"info":""}]}訂閱任務是否支援多個用戶端並行消費？
不支援。SUBSCRIBE模式允許多個用戶端並行，但只有一個用戶端可以消費到資料。
SDK代碼中封裝的是哪個版本的Kafka用戶端？
2.0.0及以上版本的dts-new-subscribe-sdk封裝的是2.7.0版本的Kafka用戶端（kafka-clients），2.0.0以下版本封裝的是1.0.0版本的Kafka用戶端。
說明若您的應用程式開發流程中採用了依賴包漏洞偵查工具，並發現dts-new-subscribe-sdk封裝的Kafka用戶端（kafka-clients）存在安全性漏洞，則您可以通過替換為
2.1.4-shaded版本來解決該問題。<dependency> <groupId>com.aliyun.dts</groupId> <artifactId>dts-new-subscribe-sdk</artifactId> <version>2.1.4-shaded</version> </dependency>
附錄
管理消費位點
當SDK用戶端初次開機、重啟或者發生內部重試時，您需要查詢並傳入消費位點（即SDK用戶端消費第一條資料的時間戳記，格式為Unix時間戳記）開始或重新消費資料。
若您需要重設用戶端的消費位點，可以根據訂閱的模式（SDK使用模式）參考下表查詢消費位點並進行修改。
情境 | SDK使用模式 | 位點管理方式 |
查詢消費位點 | ASSIGN模式、SUBSCRIBE模式 |
|
初次開機SDK用戶端，需傳入消費位點，以便消費資料。 | ASSIGN模式、SUBSCRIBE模式 | 根據SDK用戶端的使用模式，選擇Java檔案DTSConsumerAssignDemo.java或DTSConsumerSubscribeDemo.java，並配置消費位點（ |
SDK用戶端因內部重試，需重新傳入上一個記錄的消費位點，以繼續消費資料。 | ASSIGN模式 | 按如下順序，尋找上一個記錄的消費位點，找到即可返回位點資訊：
|
SUBSCRIBE模式 | 按如下順序，尋找上一個記錄的消費位點，找到即可返回位點資訊：
| |
已重啟SDK用戶端，需重新傳入上一個記錄的消費位點，以繼續消費資料。 | ASSIGN模式 | 根據consumerContext.java檔案中
|
SUBSCRIBE模式 | 該模式下consumerContext.java檔案中
|
持久化儲存消費位點
如果增量資料擷取模組觸發容災機制（特別是SUBSCRIBE模式），建立的增量資料擷取模組將無法儲存用戶端上次的消費位點資訊，可能會導致用戶端從一個較舊的位點開始消費訂閱資料，從而造成歷史資料的重複消費。例如：增量資料服務切換前，老的增量資料擷取模組位點範圍為2023年11月11日 08:00:00~ 2023年11月12日 08:00:00，用戶端的消費位點為2023年11月12日 08:00:00；增量資料服務切換後，新的增量資料擷取模組位點範圍為2023年11月08日 10:00:00~ 2023年11月12日 08:01:00，那麼用戶端會從新的增量資料擷取模組的起始位點2023年11月08日 10:00:00開始消費，造成重複消費歷史資料。
為了規避這種切換情境對歷史資料的重複消費，建議您在用戶端配置一個在用戶端儲存的消費位點持久化儲存方式。樣本方法如下，您可以根據實際情況進行修改。
建立一個
UserMetaStore()方法，繼承實現AbstractUserMetaStore()方法。例如使用MySQL資料庫儲存位點資訊，Java範例程式碼如下：
public class UserMetaStore extends AbstractUserMetaStore { @Override protected void saveData(String groupID, String toStoreJson) { Connection con = getConnection(); String sql = "insert into dts_checkpoint(group_id, checkpoint) values(?, ?)"; PreparedStatement pres = null; ResultSet rs = null; try { pres = con.prepareStatement(sql); pres.setString(1, groupID); pres.setString(2, toStoreJson); pres.execute(); } catch (Exception e) { e.printStackTrace(); } finally { close(rs, pres, con); } } @Override protected String getData(String groupID) { Connection con = getConnection(); String sql = "select checkpoint from dts_checkpoint where group_id = ?"; PreparedStatement pres = null; ResultSet rs = null; try { pres = con.prepareStatement(sql); pres.setString(1, groupID); ResultSet rs = pres.executeQuery() String checkpoint = rs.getString("checkpoint"); return checkpoint; } catch (Exception e) { e.printStackTrace(); } finally { close(rs, pres, con); } } }在consumerContext.java檔案中的
setUserRegisteredStore(new UserMetaStore())方法，配置外部儲存介質。
異常排查
異常 | 報錯提示 | 原因 | 解決方案 |
無法串連 | |
| 填入正確的 |
| 無法通過broker地址串連真實的IP地址。 | ||
| 使用者名稱和密碼錯誤。 | ||
| consumerContext.java檔案中 | 傳入在訂閱執行個體的資料範圍之內的消費位點。查詢方式，請參見參數說明。 | |
消費訂閱速度變慢 | 無 |
| |