本記事では、IntelliJ IDEAソフトウェアを例に、このクライアントを実行してSDKを使用してチャネル内のデータをコンシューム・サブスクライブする方法を紹介します。
前提条件
データサブスクリプションタスクが作成されており、タスクが正常に実行できること。
データサブスクリプションタスクにエンドポイントが設定されており、コンシュームタスクが稼働するマシンからサブスクリプションアドレスへの接続が可能であること。
データサブスクリプションタスクにコンシュームグループが作成されていること。
注意事項
データをコンシューム・サブスクライブする際は、レコードのcommit()メソッドを呼び出してライトポイント情報をコミットする必要があります。そうしない場合、データが重複してコンシュームされる可能性があります。
異なるコンシュームグループ間は相互に独立しています。
操作手順
SDKバージョンを確認します。
oms-subscribe-sdkページで最新のMaven依存関係を確認できます。
IntelliJ IDEAソフトウェアを開き、新しいMavenプロジェクトを作成し、SDK依存関係を
pom.xmlに追加します。<dependency> <groupId>com.oceanbase</groupId> <artifactId>oms-subscribe-sdk</artifactId> <version>${oms.subscribe.sdk.version}</version> </dependency>コンシュームコードを編集します。以下の例を参考に、
SdkConfig内の各パラメータに対応する値を編集する必要があります。public static void main(String[] args) { SdkConfig config = new SdkConfig.Builder() .brokerUrl("oms-cn-subsribe-****.com:18000") .topic("oms_***_topic") .groupId("oms****_groupId") .clientId("oms****_clientId") .username("oms_***_username") .password("oms_***_password") .subscribeMode(SubscribeMode.SUBSCRIBE) .startTimestampSeconds(1757576635) .build(); OmsSubscribeClient omsSubscribeClient = new OmsSubscribeClient(config); omsSubscribeClient.addRecordListeners(new TestListener()); omsSubscribeClient.init(); omsSubscribeClient.start(); omsSubscribeClient.stop(); } static class TestListener implements RecordListener { @Override public void consume(Record record) { System.out.println(record.getOperationType()); record.commit("test_info"); } }パラメータ 説明 取得方法 オプション デフォルト値 brokerUrl データサブスクリプションタスク内のサブスクリプションアドレス。 該当するデータサブスクリプションタスクの詳細ページに移動し、サブスクリプションの詳細 > データサブスクリプション タブの サブスクリプションアドレス からネットワークアドレスとポート番号情報を取得します。 必須 なし topic データサブスクリプションタスク内のサブスクリプションTopic。 該当するデータサブスクリプションタスクの詳細ページに移動し、サブスクリプションの詳細 > データサブスクリプション タブの サブスクリプションTopic からサブスクリプションTopicを確認します。 必須 なし groupId 消費管理内の消費グループ名。 該当するデータサブスクリプションタスクの詳細ページに移動し、サブスクリプションの詳細 > データサブスクリプション の 消費管理 タブから消費グループ名と消費グループのアカウント情報を取得します。 必須 なし clientId 消費管理内の ClientId。 subscribeModeがSUBSCRIBEモードの場合にのみ使用されます。なし オプション なし username コンシューマーのアカウント。 新規作成した消費グループで設定した消費グループアカウントです。 必須 なし password 消費グループのパスワード。 新規作成した消費グループで設定した消費グループアカウントのパスワードです。 必須 なし startTimestampSeconds 消費時刻のタイムスタンプであり、SDKクライアントが最初のデータを消費した時間です。形式はUnixタイムスタンプで、単位は秒です。例：1757582867。 消費時刻のタイムスタンプは、サブスクリプションインスタンスのデータ範囲内でなければならず、Unixタイムスタンプに変換する必要があります。
データサブスクリプションタスクの詳細ページの 書き込みデータのタイムスタンプ範囲 から、サブスクリプションインスタンスのデータ範囲を確認できます。オプション なし startOffset 消費タイムスタンプで、Partitionに設定されたOffsetに基づきます。形式はMap
例えば、{0: 1000, 1: 2000}は、Partition 0がOffset 1000から消費を開始し、Partition 1がOffset 2000から消費を開始することを意味します。消費タイムスタンプは、サブスクリプションインスタンスのデータ範囲内でなければなりません。
データサブスクリプションタスクの詳細ページの 書き込みデータのタイムスタンプ範囲 から、サブスクリプションインスタンスのパーティションOffset範囲を確認できます。オプション なし subscribeMode SDKクライアントの消費モードで、ASSIGNモード（デフォルト値）とSUBSCRIBEモードが含まれます。 なし オプション ASSIGN checkpointNotExistScope 起動時に消費タイムスタンプが見つからない場合の動作。BEGINNING、END、EXCEPTIONを含みます。
消費タイムスタンプが見つからない原因には、起動時間が指定されていないか、指定された起動時間に対応するタイムスタンプが存在しない場合が含まれます。なし オプション EXCEPTION properties チューニングパラメータなどの追加のKafkaコンシューマー設定プロパティです。Propertiesオブジェクトの形式で、Kafka Consumerの任意の構成パラメータを設定できます。 Kafka公式ドキュメントのConsumer構成パラメータをご参照ください。 オプション なし pollTimeoutMs Kafkaコンシューマーのpoll操作のタイムアウト時間で、単位はミリ秒です。 なし オプション 500 checkpointCommitInterval Checkpointコミット間隔時間で、単位はミリ秒です。SDKはこの間隔で定期的に消費進捗をコミットします。
注意：コミットされるタイムスタンプの出典は、Recordのcommit()メソッドの呼び出しに依存します。なし オプション 5000 userRegisteredStore ユーザー定義のCheckpointストレージ実装で、Checkpointの永続化方法をカスタマイズするために使用されます。 SerializeCheckpointStoreインターフェースを実装する必要があります。SerializeCheckpointStoreインターフェースを実装し、カスタムのCheckpointストレージロジックを提供します。オプション なし