MongoDBサブスクリプションデータの消費
ダウンロード
フォーカスモード
フォントサイズ
操作シナリオ
データ購読Kafka版（現在のKafka ServerバージョンはV2.6.0）では、0.11バージョン以上のKafkaクライアントを使用して購読データを消費できます。このドキュメントでは、消費Demoサンプルを提供しており、データ消費のプロセスを迅速にテストし、データ形式の解析方法を理解するのに役立ちます。
注意事項
Demoにはデータ消費の使用方法のデモは含まれておらず、データの印刷処理のみが行われています。お客様はこれに基づいてデータ処理ロジックを自身で作成する必要があります。他の言語のKafkaクライアントを使用してデータを消費し、解析することもできます。
現在、外部ネットワーク経由でデータ購読のKafkaに接続して消費することはサポートされておらず、Tencent Cloudの内部ネットワークからのアクセスのみがサポートされています。また、購読するデータベースインスタンスが属する地域とデータ消費の地域は同じである必要があります。
DTS購読に内蔵されているKafkaが処理する単一メッセージには一定の上限があります。ソースデータベース内の単一行データが5MBを超えると、購読タスクでエラーが発生する可能性があります。
データの再入可能性を保証するために、DTS購読はCheckpointメカニズムを導入しています。メッセージがKafka Topicに書き込まれる際、通常10秒ごとにCheckpointが1つ挿入され、データ同期の位置を識別するために使用されます。タスクが中断された後に再起動すると、中断位置を認識して再開できます。また、コンシューマー側はCheckpointメッセージを受信すると、Kafkaの消費位置のコミットを一度実行し、消費位置をタイムリーに更新します。
Demoのダウンロードを消費
MongoDBデータ購読は現在、JSONデータ形式のみをサポートしています。以下のDemoサンプルにはJSONプロトコルファイルが既に含まれているため、別途ダウンロードする必要はありません。
Java Demoの操作手順
コンパイル環境：Mavenパッケージ管理ツール、JDK8。ユーザーはパッケージツールを自身で選択できます。以下ではMavenを例として説明します。
実行環境：Tencent Cloud CVM（購読インスタンスと同じ地域である必要があり、Kafkaサーバーの内部ネットワークアドレスにアクセスできます）、JRE8をインストールします。
操作手順は以下の通りです：
1. 新しいバージョンのデータ購読タスクを作成します。詳細については、MongoDBデータ購読の新規作成を参照してください。
2. 1つまたは複数のコンシューマグループを作成します。詳細については、コンシューマグループの追加を参照してください。
3. Java Demoをダウンロードして、そのファイルを解凍してください。
4. 解凍後のディレクトリに入ります。利用しやすいように、ディレクトリにはMavenモデルファイルとpom.xmlファイルがそれぞれ配置されています。ユーザーは必要に応じて選択してください。
Mavenを使用してパッケージングします：mvn clean package。
5. Demoを実行します。
Mavenでパッケージングした後、ターゲットフォルダtargetに入り、以下のコードを実行します：
java -jar consumerDemo-json-1.0-SNAPSHOT.jar --brokers xxx --topic xxx --group xxx --user xxx --password xxx --trans2sql --trans2canalbrokersはデータ購読Kafkaの内部ネットワークアクセスアドレスです。topicはデータ購読タスクの購読topicです。この2つは購読詳細ページで確認できます。group、user、passwordはそれぞれコンシューマグループの名前、アカウント、パスワードです。消費管理ページで確認できます。trans2sqlはSQL文に変換するかどうかを示します。javaコードでは、このパラメータを指定するとSQL文に変換され、指定しない場合は変換されません。trans2canalはCanal形式に変換して出力するかどうかを示します。このパラメータを指定するとCanal形式に変換され、指定しない場合は変換されません。（現在、JSONデータ形式のみがこのパラメータに関係します）6. 消費状況を観察してください。
Golang Demoの操作手順
コンパイル環境：Golang 1.12以降のバージョン、Go Module環境を設定してください。
実行環境：Tencent Cloud CVM（購読インスタンスと同じ地域である必要があり、Kafkaサーバーの内部ネットワークアドレスにアクセスできます）。
操作手順は以下の通りです：
1. 新しいバージョンのデータ購読タスクを作成します。詳細については、データ購読Kafka版を参照してください。
2. 1つまたは複数のコンシューマグループを作成します。詳細については、コンシューマグループの追加を参照してください。
3. Golang Demoをダウンロードして、そのファイルを解凍してください。
4. 解凍後のディレクトリに入り、
go build -o subscribe ./main/main.goを実行して、実行可能ファイルsubscribeを生成します。5. 以下のコードを実行します：
./subscribe --brokers=xxx --topic=xxx --group=xxx --user=xxx --password=xxx --trans2sql=truebrokersはデータ購読Kafkaの内部ネットワークアクセスアドレスです。topicはデータ購読タスクの購読topicです。この2つは購読詳細ページで確認できます。group、user、passwordはそれぞれコンシューマグループの名前、アカウント、パスワードです。消費管理ページで確認できます。trans2sqlはSQL文に変換するかどうかを示します。6. 消費状況を観察してください。
Python3 Demoの操作手順
コンパイル実行環境：Tencent Cloud CVM（購読インスタンスと同じ地域である必要があり、Kafkaサーバーの内部ネットワークアドレスにアクセスできます）、Python3、pip3（依存パッケージのインストールに使用）をインストールします。
pip3を使用して依存パッケージをインストールします：
pip install flagpip install kafka-python
操作手順は以下の通りです：
1. 新しいバージョンのデータ購読タスクを作成します。詳細については、データ購読Kafka版を参照してください。
2. 1つまたは複数のコンシューマグループを作成します。詳細については、コンシューマグループの追加を参照してください。
3. Python3 Demoをダウンロードして、そのファイルを解凍してください。
4. 以下のコードを実行します：
python main.py --brokers=xxx --topic=xxx --group=xxx --user=xxx --password=xxx --trans2sql=1brokersはデータ購読Kafkaの内部ネットワークアクセスアドレスです。topicはデータ購読タスクの購読topicです。この2つは購読詳細ページで確認できます。group、user、passwordはそれぞれコンシューマグループの名前、アカウント、パスワードです。消費管理ページで確認できます。trans2sqlはSQL文に変換するかどうかを示します。5. 消費状況を観察してください。
フィードバック