Data Transmission Service:SDK サンプルコードによるサブスクリプションデータの消費

手順

次の手順では、IntelliJ IDEA (Community Edition 2020.1 for Windows) で SDK サンプルコードを実行してデータサブスクリプションデータを消費する方法を説明します。

1. データサブスクリプションインスタンスを作成します。詳細については、「ApsaraDB RDS for MySQL インスタンスのデータサブスクリプションチャネルの作成」、「PolarDB for MySQL クラスターのデータサブスクリプションチャネルの作成」、または「Oracle データベースのデータサブスクリプションチャネルの作成」をご参照ください。

2. 1 つ以上のコンシューマーグループを作成します。詳細については、「コンシューマーグループの作成」をご参照ください。

   重要

   データサブスクリプションデータを消費する場合、DefaultUserRecord の commit メソッドを呼び出してチェックポイントをコミットする必要があります。コミットしないと、データが重複して消費される可能性があります。

3. ビジネス要件に基づいて SDK サンプルコードを使用します。

   • 新しいデータサブスクリプション SDK パッケージを使用する (推奨)

     1. IntelliJ IDEA を開き、[Create New Project] をクリックして、アプリケーション用のプロジェクトを作成します。

     2. プロジェクトで、プロジェクトオブジェクトモデル (POM) ファイル pom.xml を見つけます。

     3. pom.xml ファイルに次の依存関係を追加します：

        <dependency>
            <groupId>com.aliyun.dts</groupId>
            <artifactId>dts-new-subscribe-sdk</artifactId>
            <version>{dts_new_sdk_version}</version>
        </dependency>

        説明

        最新の Maven 依存関係は、dts-new-subscribe-sdk ページで確認できます。

     4. 新しいサブスクリプション SDK の使用方法の詳細については、「サンプルコードの使用」をご参照ください。

   • 新しいデータサブスクリプション SDK のカスタマイズ版を使用する

     1. SDK サンプルコードパッケージをダウンロードし、解凍します。

        説明

        をクリックし、[Download ZIP] を選択してパッケージをダウンロードします。

     2. 解凍した SDK サンプルコードのディレクトリに移動します。テキストエディターを使用して pom.xml ファイルを開き、データサブスクリプション SDK を最新バージョンに更新します。

        重要

        データサブスクリプション SDK の最新バージョンは Maven の Web サイトから取得できます。詳細については、「データサブスクリプション SDK の Maven ページ」をご参照ください。

     3. IntelliJ IDEA を開き、[Open or Import] をクリックします。

     4. 表示されるダイアログボックスで、解凍した SDK サンプルコードのディレクトリに移動し、フォルダーを展開して pom.xml ファイルを見つけます。

     5. 表示されるダイアログボックスで、[Open as Project] を選択します。

     6. IntelliJ IDEA でフォルダーを展開します。SDK クライアントの使用モードに基づいて、対応する Java ファイル (DTSConsumerAssignDemo.java または DTSConsumerSubscribeDemo.java) を選択してダブルクリックします。

        説明

        DTS は、次の SDK クライアント使用モードをサポートしています：

        • ASSIGN モード：メッセージのグローバルな順序を保証するため、DTS は各サブスクリプショントピックに 1 つのパーティション (パーティション 0) のみを割り当てます。ASSIGN モードで SDK クライアントを使用する場合、クライアントを 1 つのみ起動することを推奨します。

        • SUBSCRIBE モード：メッセージのグローバルな順序を保証するため、DTS は各サブスクリプショントピックに 1 つのパーティション (パーティション 0) のみを割り当てます。SUBSCRIBE モードで SDK クライアントを使用する場合、ディザスタリカバリのためにコンシューマーグループ内で複数の SDK クライアントを起動できます。アクティブなクライアントに障害が発生した場合、別の SDK クライアントが自動的にパーティション 0 に割り当てられ、消費を再開します。

4. Java ファイルで必要なパラメーターを設定します。

   

   表 1. 必須パラメーター

   パラメーター	説明	ソース
   brokerUrl	データサブスクリプションインスタンスのエンドポイントとポート番号。 説明 SDK クライアントを実行している ECS インスタンスとデータサブスクリプションインスタンスが同じクラシックネットワークまたは Virtual Private Cloud (VPC) にある場合は、ネットワークレイテンシを最小限に抑えるために、サブスクリプションに内部エンドポイントを使用することを推奨します。 ネットワークが不安定になる可能性があるため、パブリックエンドポイントの使用は推奨されません。	DTS コンソールで、対象のデータサブスクリプションインスタンスの ID をクリックします。基本情報 ページで、ネットワーク セクションのエンドポイントとポート番号を取得できます。
   topic	インスタンスのサブスクリプショントピック。	DTS コンソールで、対象のデータサブスクリプションインスタンスの ID をクリックします。基本情報 ページの 基本情報 セクションで トピック を取得できます。
   sid	コンシューマーグループの ID。	DTS コンソールで、ターゲットのデータサブスクリプションインスタンスの ID をクリックし、データ消費 をクリックします。コンシューマーグループの [コンシューマーグループ ID] と アカウント を取得できます。 説明 コンシューマーグループのユーザー名のパスワードは、コンシューマーグループの作成時に指定します。
   userName	コンシューマーグループのユーザー名。 警告 このトピックで提供されるクライアントを使用しない場合は、ユーザー名を <Username>-<Consumer Group ID> 形式で設定する必要があります。例：dtstest-dtsae******bpv。そうしないと、接続は失敗します。
   password	ユーザー名のパスワード。
   initCheckpoint	SDK クライアントがデータの消費を開始する消費チェックポイント。UNIX タイムスタンプとして指定します。例：1620962769。 説明 消費チェックポイント情報は、次のシナリオで使用できます。 アプリケーションの中断後に消費を再開し、データ損失を防ぐには、最後に認識された消費チェックポイントを渡します。 クライアントの起動時に、特定の消費チェックポイントを渡して、目的の位置からデータを消費できます。	消費チェックポイントは、データサブスクリプションインスタンスのデータ範囲内 (図を参照) にある必要があり、UNIX タイムスタンプに変換する必要があります。 説明 検索エンジンを使用して UNIX タイムスタンプコンバーターを見つけることができます。
   ConsumerContext.ConsumerSubscribeMode subscribeMode	SDK クライアントの使用モード。有効値： ConsumerContext.ConsumerSubscribeMode.ASSIGN：ASSIGN モード。コンシューマーグループ内の 1 つの SDK クライアントのみがデータサブスクリプションデータを消費できます。 ConsumerContext.ConsumerSubscribeMode.SUBSCRIBE：SUBSCRIBE モード。ディザスタリカバリのために、同じコンシューマーグループで複数の SDK クライアントを起動できます。	N/A

5. IntelliJ IDEA の上部メニューで、[Run] > [Run] を選択してクライアントを実行します。

   説明

   クライアントを初めて実行するとき、必要な依存関係が自動的にロードされてインストールされるまでに時間がかかる場合があります。

   • 結果を次の図に示します。クライアントはソースデータベースから変更データを正常に消費します。

   • SDK クライアントは、送受信されたレコードの総数と総量、および 1 秒あたりのリクエスト数 (RPS) など、データ消費に関する統計を定期的に集計して表示します。

     表 2. データ消費統計

     パラメーター	説明
     outCounts	SDK クライアントが消費したデータレコードの総数。
     outBytes	SDK クライアントが消費したデータの総量。単位：バイト。
     outRps	SDK クライアントがデータを消費する 1 秒あたりのリクエスト数 (RPS)。
     outBps	SDK クライアントのデータ消費レート。単位：バイト/秒 (Bps)。
     inBytes	DTS サーバーが送信したデータの総量。単位：バイト。
     DStoreRecordQueue	DTS サーバーからの受信レコードのための内部データキャッシュキューのサイズ。
     inCounts	DTS サーバーが送信したデータレコードの総数。
     inRps	DTS サーバーがデータを送信する RPS。
     __dt	SDK クライアントがデータを受信したときのタイムスタンプ。単位：ミリ秒。
     DefaultUserRecordQueue	コンシューマーアプリケーションで処理可能なレコードを保持するデータキューのサイズ。

消費チェックポイントの保存とクエリ

データ消費を開始または再開する (初回起動、再起動、内部リトライなど) には、SDK クライアントに消費チェックポイントが必要です。次の表では、データ損失の防止、重複消費の最小化、オンデマンド消費の実現を目的に、シナリオ別のチェックポイントの管理方法とクエリ方法を示します。

消費チェックポイントの永続化

増分データ収集モジュールでディザスタリカバリ イベントが発生した場合 (特に 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) から開始するため、重複消費が発生します。

このシナリオでの重複消費を回避するには、クライアント側で永続チェックポイントストアを構成してください。次の例は、要件に応じて調整できる実装の 1 つを示しています。

1. AbstractUserMetaStore を継承する UserMetaStore クラスを作成してください。

   例えば、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;
           String checkpoint = null;
   
           try {
               pres = con.prepareStatement(sql);
               pres.setString(1, groupID);
               rs = pres.executeQuery();
               
               if (rs.next()) {
                   checkpoint = rs.getString("checkpoint");
               }
           } catch (Exception e) {
               e.printStackTrace();
           } finally {
               close(rs, pres, con);
           }
           return checkpoint;
       }
   }
   

2. consumerContext.java ファイルで、setUserRegisteredStore(new UserMetaStore()) メソッドを使用して外部ストレージを構成してください。

よくある質問

• データサブスクリプションインスタンスの接続の問題はどのように解決できますか。

  エラーメッセージに基づいてトラブルシューティングを行ってください。詳細については、「トラブルシューティング」をご参照ください。

• 消費チェックポイントはどのフォーマットで永続化されますか。

  永続化された消費チェックポイントデータは JSON フォーマットで保存されます。永続化されたチェックポイントは UNIX タイムスタンプで、 SDK に直接渡すことができます。次の応答例では、"timestamp" キーの値 1700709977 が、永続化された消費チェックポイントです。

  {"groupID":"dtsglg11d48230***","streamCheckpoint":[{"partition":0,"offset":577989,"topic":"ap_southeast_1_vpc_rm_t4n22s21iysr6****_root_version2","timestamp":1700709977,"info":""}]}

トラブルシューティング

ERROR
CheckResult{isOk=false, errMsg='telnet dts-cn-hangzhou.aliyuncs.com:18009
failed, please check the network and if the brokerUrl is correct'}
(com.aliyun.dts.subscribe.clients.DefaultDTSConsumer)

telnet real node *** failed, please check the network

ERROR CheckResult{isOk=false, errMsg='build kafka consumer failed, error: org.apache.kafka.common.errors.TimeoutException: Timeout expired while fetching topic metadata, probably the user name or password is wrong'} (com.aliyun.dts.subscribe.clients.DefaultDTSConsumer)

com.aliyun.dts.subscribe.clients.exception.TimestampSeekException: RecordGenerator:seek timestamp for topic [cn_hangzhou_rm_bp11tv2923n87081s_rdsdt_dtsacct-0] with timestamp [1610249501] failed