SDK デモを使用した PolarDB-X 1.0 からの変更データの消費 - Data Transmission Service

変更追跡タスクを作成した後、Data Transmission Service (DTS) が提供する SDK デモを使用して、結果のデータ変更を消費します。このトピックでは、PolarDB-X 1.0 や DMS LogicDB などの分散データソースで SDK デモを使用する方法について説明します。

前提条件

JDK 1.8 がインストールされていること。
IntelliJ IDEA がインストールされていること。

注意事項

RAM ユーザーとして変更データを消費するには、そのユーザーに AliyunDTSFullAccess 権限とソースオブジェクトにアクセスするための権限を付与する必要があります。詳細については、「RAM ユーザーへの DTS 管理権限の付与」および「RAM ユーザーの権限管理」をご参照ください。

操作手順

このトピックでは、Windows 向けの IntelliJ IDEA Community Edition 2020.1 を例として、SDK デモを実行して PolarDB-X 1.0 インスタンスから変更データを消費する方法を説明します。

変更追跡インスタンスを作成します。詳細については、「PolarDB-X 1.0 の変更追跡タスクの作成」をご参照ください。
1 つ以上のコンシューマーグループを作成します。詳細については、「コンシューマーグループの作成」をご参照ください。
SDK デモをダウンロードして解凍します。ダウンロードリンクについては、「SDK デモコード」をご参照ください。

重要
変更データを消費する際は、DefaultUserRecord の commit メソッドを呼び出してコンシューマオフセットを送信する必要があります。そうしないと、データが繰り返し消費されるリスクがあります。
IntelliJ IDEA でプロジェクトを開きます。
1. IntelliJ IDEA を開き、[Open or Import] をクリックします。
2. ダイアログボックスで、SDK デモを解凍したディレクトリに移動し、フォルダを展開して、pom.xml ファイルをダブルクリックします。
3. 表示されるダイアログボックスで、[Open as Project] を選択します。

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

aliyun-dts-subscribe-sdk-java-master [dts-new-subscr...]
  .idea
  src
    main
    test
      java
        com.aliyun.dts.subscribe.clients
          DBMapperTest
          DistributedDTSConsumerDemo
          DTSConsumerAssignDemo
          DTSConsumerSubscribeDemo
          UserMetaStore
  .gitignore
  dts-new-subscribe-sdk.iml
  LICENSE
  pom.xml
  README.md
External Libraries
Scratches and Consoles

Java ファイルで必須パラメーターを設定します。

public static void main(String[] args) throws ClientException {
        // PolarDB-X 1.0 インスタンスなどの分散データソースの変更追跡設定を構成します。AccessKey ペア、インスタンス ID、タスク ID、コンシューマーグループなどのパラメーターを設定します。
        String accessKeyId = "LTA***********99reZ";
        String accessKeySecret = "****************";
        String regionId = "cn-hangzhou";
        String dtsInstanceId = "dtse5212sed162****";
        String jobId = "l791216x16d****";
        String sid = "dtsip412t13160****";
        String userName = "xftest";
        String password = "******";
        String proxyUrl = "dts-cn-****.com:18001";
        // 初期コンシューマオフセット。UNIX タイムスタンプ (秒単位) です。例：2019年8月19日 10:03:21 CST の場合は 1566180200。
        String checkpoint = "1639620090";
        // 物理データベース/テーブル名を論理データベース/テーブル名に変換します
        boolean mapping = true;
        // true に設定すると、クライアントは指定されたチェックポイントから強制的に開始されます。チェックポイントのリセットは ASSIGN モードでのみ機能します。
        boolean isForceUseInitCheckpoint = false;
        ConsumerContext.ConsumerSubscribeMode subscribeMode = ConsumerContext.ConsumerSubscribeMode.ASSIGN;
        DistributedDTSConsumerDemo demo = new DistributedDTSConsumerDemo(userName, password, regionId,
                jobId, sid, dtsInstanceId, accessKeyId, accessKeySecret, subscribeMode, proxyUrl,
                checkpoint, isForceUseInitCheckpoint, mapping);
        demo.start();
    }

パラメーター	説明	取得方法
accessKeyId	AccessKey ID。	詳細については、「AccessKey ペアの取得」をご参照ください。
accessKeySecret	AccessKey Secret。	詳細については、「AccessKey ペアの取得」をご参照ください。
regionId	変更追跡インスタンスのリージョン ID。	DTS コンソールで、ターゲットの変更追跡インスタンスの ID をクリックします。タスク管理ページで、リージョン情報を確認します。たとえば、リージョンが中国 (杭州) の場合、このパラメーターを `cn-hangzhou` に設定します。リージョンの一覧については、「サポートされているリージョンのリスト」をご参照ください。
dtsInstanceId	変更追跡インスタンスの ID。	DTS コンソールで、対象の変更追跡インスタンスの ID をクリックします。タスク管理ページで、インスタンス ID とタスク ID を確認できます。
jobId	変更追跡タスクの ID。
sid	コンシューマーグループの ID。	DTS コンソールで、対象の変更追跡インスタンスの ID をクリックし、データ消費をクリックします。コンシューマーグループの SID とアカウントを確認できます。説明コンシューマーグループアカウントのパスワードは、コンシューマーグループの作成時に指定します。
userName	コンシューマーグループのアカウント。
password	コンシューマーグループアカウントのパスワード。
proxyUrl	変更追跡インスタンスのエンドポイントとポート。説明ネットワーク遅延を最小限に抑えるには、SDK クライアントを実行している ECS インスタンスが変更追跡インスタンスと同じクラシックネットワークまたは VPC にある場合、内部エンドポイントを使用します。ネットワークが不安定になる可能性があるため、パブリックエンドポイントの使用は推奨されません。	DTS コンソールで、対象の変更追跡インスタンスの ID をクリックします。タスク管理ページで、エンドポイントとポート番号を確認できます。
checkpoint	コンシューマオフセット。UNIX タイムスタンプ (秒単位) で指定します。SDK クライアントは、この時点からデータの消費を開始します。説明コンシューマオフセットは、次のシナリオで使用されます：アプリケーションが中断された場合、最後のコンシューマオフセットを渡して、データ損失なしで消費を再開します。クライアントの起動時に、特定のコンシューマオフセットを渡して、オンデマンドでデータを消費できます。	コンシューマオフセットは、変更追跡インスタンスのデータ範囲内であり、UNIX タイムスタンプである必要があります。説明検索エンジンを使用して、UNIX タイムスタンプコンバーターを見つけることができます。

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

説明

初回実行時、IntelliJ IDEA は必要な依存関係を自動的にインストールするため、時間がかかる場合があります。

出力には、クライアントがソースインスタンスからのデータ変更を消費できることが示されます。

SDK クライアントは、送受信されたデータレコードの総数とボリューム、およびリクエスト/秒 (RPS) を含む消費統計を定期的に表示します。

表 1. 消費統計

パラメーター	説明
`outCounts`	SDK クライアントによって消費されたデータレコードの総数。
`outBytes`	SDK クライアントによって消費されたデータの総量 (バイト単位)。
`outRps`	SDK クライアントのデータ消費レート (リクエスト/秒 (RPS))。
`outBps`	SDK クライアントのデータ消費レート (ビット/秒 (bps))。
`count`	このパラメーターは予約済みです。
`inBytes`	DTS サーバーから送信されたデータの総量 (バイト単位)。
`DStoreRecordQueue`	DTS サーバー上のデータキャッシュキューの現在のサイズ。
`inCounts`	DTS サーバーから送信されたデータレコードの総数。
`inRps`	DTS サーバーのデータ送信レート (リクエスト/秒 (RPS))。
`inBps`	DTS サーバーのデータ送信レート (ビット/秒 (bps))。
`__dt`	SDK クライアントがデータを受信したときのタイムスタンプ (ミリ秒単位)。
`DefaultUserRecordQueue`	シリアル化後のデータキャッシュキューの現在のサイズ。

オプション： 消費されたデータの処理方法をカスタマイズするには、buildRecordListener() メソッドを変更するか、カスタムクラスを使用します。

public static Map<String, RecordListener> buildRecordListener() {
        // 独自のリスナーを実装できます。
        RecordListener mysqlRecordPrintListener = new RecordListener() {
            @Override
            public void consume(DefaultUserRecord record) {
                OperationType operationType = record.getOperationType();
                if (operationType.equals(OperationType.INSERT)
                        || operationType.equals(OperationType.UPDATE)
                        || operationType.equals(OperationType.DELETE)
                        || operationType.equals(OperationType.HEARTBEAT)) {
                    // レコードを消費します。
                    RecordListener recordPrintListener = new DefaultRecordPrintListener(DbType.MySQL);
                    recordPrintListener.consume(record);
                    // commit メソッドはチェックポイントの更新をプッシュします。
                    record.commit("");
                }
            }
        };
        return Collections.singletonMap("mysqlRecordPrinter", mysqlRecordPrintListener);
    }