本記事では、IntelliJ IDEAを例に、このクライアントを起動し、SDKを使用してサブスクリプションチャネル内のデータを消費する方法を説明します。
前提条件
データサブスクリプションタスクが作成されており、タスクが正常に実行されていること。
データサブスクリプションタスクにターミナルノードが設定されており、消費タスクが実行されているマシンからサブスクリプションアドレスへの接続が可能であること。
データサブスクリプションタスクに対してコンシューマーグループが作成されていること。
注意事項
データが重複して消費される可能性がありますので、サブスクリプションデータを消費する際は、Recordの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ストレージロジックを提供します。オプション なし