言葉が出ません！私は Java の面接に来ましたが、なぜビッグデータ用の Kafka について質問するのですか?

みなさんこんにちは。私はNezhaです。数日前、私の友人が面接に行き、Kafka トランザクションについて質問されました。彼女の最初の反応は、「私は Java の面接に来たのに、なぜビッグデータにおける Kafka について聞くのですか?」でした。

しかし、Kafka が Java プログラマーにとって不可欠なミドルウェア テクノロジであることは間違いありません。

Kafka は、今日の時代におけるデータ パイプラインのほぼ第一選択肢です。バックエンド開発やビッグデータ開発を行っている方なら、ご存知かもしれません。オープンソースソフトウェア Kafka の応用はますます広まっています。

Kafka の人気と学習ブームを受けて、Nezha は長年の開発経験を共有し、読者が Kafka の関連知識をより簡単に習得できるようにしたいと考えています。

今日は、カフカの件について体系的に話し、段階的に計画を立て、一つずつ分解してカフカを倒していきます。

今日のビッグデータ時代では、データの信頼性と一貫性が重要になっています。分散ストリーミング データ プラットフォームとして、Kafka はリアルタイム データの高スループット伝送を重視しており、Kafka トランザクション メッセージはこのプロセスで重要な役割を果たします。

この記事では、Kafka トランザクション メッセージについて詳しく説明し、それらがどのようにデータの一貫性を確保するか、そしてさまざまなアプリケーション シナリオでのそれらの適用について説明します。

1. Kafka トランザクションメッセージ

1. Kafkaトランザクションメッセージの導入

Kafka トランザクション メッセージングは​​、データの一貫性を確保するための重要なサポートを提供する重要な機能です。このセクションでは、Kafka トランザクション メッセージングの基本的な概念について詳しく説明します。

Kafka トランザクションメッセージの概念

トランザクション メッセージングがなぜ必要なのでしょうか?

トランザクション メッセージングは​​、データの一貫性を確保するために重要です。一部のアプリケーションでは、メッセージの整合性と信頼性が重要です。メッセージ処理中に障害が発生した場合に、メッセージが失われたり重複したりしないようにする方法は複雑な問題です。 Kafka トランザクション メッセージは、これらの問題を解決する方法を提供し、メッセージ処理の制御性と信頼性を高めます。

トランザクションメッセージの特徴

Kafka トランザクション メッセージングには、次の主な機能があります。

• 原子性: トランザクション メッセージは完全に成功するか、完全に失敗します。これにより、メッセージが部分的に処理されないようになります。
  
• 信頼性: メッセージは、アプリケーションまたはシステムに障害が発生した場合でも、Kafka に書き込まれると処理済みとみなされます。
  

• 秩序性: トランザクション メッセージは、単一のパーティション内で順序どおりに保持されます。これは、順次処理を必要とするアプリケーションにとって重要です。

• べき等性: Kafka プロデューサーはべき等性を持つように構成できるため、同じメッセージが繰り返し送信されなくなります。
  
• 正確に 1 回のセマンティクス: トランザクション メッセージは「正確に 1 回」のセマンティクスをサポートします。つまり、メッセージは正確に 1 回到着するか、まったく到着しないかのいずれかになります。
  

このセクションの目的は、Kafka トランザクション メッセージングの中核となる概念を理解できるようにすることです。次に、それらのアプリケーション シナリオと、非トランザクション メッセージと比較した利点について説明します。

2. トランザクションメッセージの適用シナリオ

トランザクション メッセージは、多くのアプリケーション シナリオで重要な役割を果たします。トランザクション メッセージが特に役立つ一般的な使用例をいくつか示します。

金融取引処理: 金融分野では、矛盾した取引や重複した取引が発生しないように、各取引はアトミックである必要があります。トランザクション メッセージは、金融トランザクションを記録および処理し、トランザクションの整合性を確保するために使用できます。

注文処理: 電子商取引プラットフォームでは、注文が問題なく作成、支払い、発送されるように、注文処理が信頼できるものでなければなりません。トランザクション メッセージを使用すると、注文のさまざまな段階を追跡および処理して、一貫した注文フローを確保できます。

在庫管理: 在庫管理はビジネスにとって非常に重要です。トランザクション メッセージを使用して在庫の変化を追跡し、在庫の正確性と信頼性を確保できます。

ログ記録: ビッグ データおよびログ記録アプリケーションでは、ログの整合性が非常に重要です。トランザクション メッセージを使用すると、ログの整合性を確保し、ログ処理クラスターで障害が発生した場合でも一貫性を維持できます。

システム通知: ユーザーに通知やアラートを送信する必要があるアプリケーションの場合、通知が確実に配信されることが重要です。これを実現するには、トランザクション メッセージングを使用できます。

3. Kafkaトランザクションメッセージの利点

非トランザクション メッセージと比較すると、Kafka トランザクション メッセージには、特にデータの一貫性が求められるアプリケーション シナリオにおいて明らかな利点があります。 Kafka トランザクション メッセージングの利点は次のとおりです。

データの一貫性: トランザクション メッセージングにより、メッセージが完全に処理されるか、または処理されないかが保証されます。これにより、データ処理における不整合が排除され、データの一貫性が維持されます。

信頼性: メッセージは、アプリケーションまたはシステム障害が発生した場合でも、Kafka に書き込まれると処理済みとみなされます。これにより、メッセージの確実な配信が保証されます。

べき等性: Kafka プロデューサーはべき等性を持つように構成できます。つまり、同じメッセージが繰り返し送信されません。これにより、不要なメッセージングが削減され、データの重複が回避されます。

正確に 1 回のセマンティクス: トランザクション メッセージは「正確に 1 回」のセマンティクスをサポートします。つまり、メッセージは正確に 1 回到着するか、まったく到着しないかのいずれかになります。これは、一部のアプリケーションで必要な高レベルのセマンティクスです。

エラー処理: トランザクション メッセージは、メッセージが失われることなく回復または再試行できるように、エラーを処理するメカニズムを提供します。

2. Kafkaトランザクションメッセージの使用

このセクションでは、Kafka トランザクション メッセージングを使用してデータの一貫性を確保する方法について詳しく説明します。

1. トランザクションメッセージをサポートするようにKafkaを構成する

メッセージが一貫して配信および処理されるようにするには、トランザクション メッセージングをサポートするように Kafka を構成することが重要です。このセクションでは、プロデューサーとコンシューマーの両方のセットアップを含め、トランザクション メッセージングをサポートするように Kafka を構成する方法について詳しく説明します。

プロデューサーの設定

プロデューサー側では、トランザクション メッセージを有効にするために特定の構成が必要です。主な構成パラメータは次のとおりです。

• acks: これは、プロデューサーが確認を受け取った後にのみメッセージが正常に送信されたとみなすための設定です。トランザクション メッセージの場合、通常は acks=all に設定され、トランザクションが完全にコミットされた後にのみメッセージが正常に送信されたとみなされます。
• transactional.id: これはプロデューサーインスタンスを識別するために使用される一意の ID です。構成ファイルで transactional.id を設定することは、トランザクション メッセージングを有効にするための重要なステップです。
• enable.idempotence: べき等性とは、同じメッセージが繰り返し送信されないことを意味します。トランザクション メッセージの場合、メッセージが重複して送信されないようにするために、通常は enable.idempotence=true に設定されます。
  

設定例:

 acks=all transactional.id=my-transactional-id enable.idempotence=true

消費者の構成

• isolation.level: これは、コンシューマーの分離レベルを制御するために使用される設定です。トランザクション メッセージの場合、コミットされたトランザクション メッセージのみが読み取られるように、通常は isolation.level=read_committed に設定されます。
• auto.offset.reset: これは、コンシューマーが起動時にメッセージの読み取りを開始する設定です。通常、コミットされたメッセージが失われないように、これは auto.offset.reset=earliest に設定されます。
  

設定例:

 isolation.level=read_committed auto.offset.reset=earliest

トランザクション メッセージングをサポートするように Kafka を構成することは、信頼性の高いメッセージ配信と処理を保証するための重要なステップです。これらの構成設定により、トランザクション メッセージの生成と使用時に正しい動作が保証されます。

2. プロデューサー: トランザクションメッセージの送信

この部分では、Kafka プロデューサーを使用してトランザクション メッセージを送信する方法について詳しく説明します。トランザクション メッセージの送信は、データの一貫性を確保するための重要なステップであり、特別な注意が必要です。詳細な手順と例は次のとおりです。

Kafka プロデューサーの作成

まず、Kafka プロデューサーのインスタンスを作成する必要があります。このプロデューサー インスタンスは、Kafka トピックにメッセージを送信する役割を担います。プロデューサーを作成するには、Kafka クラスターのアドレス、メッセージ キーと値のシリアライザー、トランザクション ID などの構成パラメータが必要です。

以下は Kafka プロデューサーを作成する例です。

 import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.ProducerConfig; import java.util.Properties; public class MyKafkaProducer { public static Producer<String, String> createProducer() { Properties properties = new Properties(); properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "my-transactional-id"); return new KafkaProducer<>(properties); } }

取引を開始

トランザクション メッセージを送信する前に、明示的にトランザクションを開始する必要があります。これは、   beginTransaction  これを実現する方法。トランザクションが開始されると、その後のすべてのメッセージ送信はこのトランザクションに含まれます。

 producer.beginTransaction();

メッセージを送信

トランザクション内で、メッセージの送信を開始できます。これらのメッセージはトランザクションに含まれ、トランザクションが正常にコミットされた場合にのみ Kafka トピックに実際に書き込まれます。

 producer.send(new ProducerRecord<>("my-topic", "key1", "value1")); producer.send(new ProducerRecord<>("my-topic", "key2", "value2"));

トランザクションをコミットまたは中止する

トランザクション メッセージの主な特性は、完全に成功するか、完全に失敗するかのいずれかであるということです。したがって、メッセージが送信された後、メッセージ処理の結果に基づいてトランザクションをコミットするか中止するかを決定する必要があります。これは、commitTransaction メソッドまたは abortTransaction メソッドを呼び出すことによって実現できます。

 try { producer.commitTransaction(); } catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) { // 处理异常，通常中止事务并重试producer.close(); } catch (CommitFailedException e) { // 事务提交失败，通常中止事务并重试producer.close(); }

上記の手順は、Kafka プロデューサーを使用してトランザクション メッセージを送信する方法の基本的な例を示しています。トランザクション メッセージを送信すると、メッセージの信頼性と一貫性が確保されます。これは、アトミック性の保証が必要な場合に特に役立ちます。

3. コンシューマー: トランザクションメッセージの処理

このセクションでは、Kafka コンシューマーを使用してトランザクション メッセージを処理する方法について詳しく説明します。トランザクション メッセージを正しく処理することは、データの一貫性を確保するために重要です。詳細な手順と例は次のとおりです。

Kafka コンシューマーの作成

まず、Kafka コンシューマーのインスタンスを作成する必要があります。このコンシューマー インスタンスは、Kafka トピックからのメッセージの読み取りを担当します。コンシューマーを作成するには、Kafka クラスターのアドレス、メッセージのキーと値のデシリアライザー、コンシューマー グループ ID などの構成パラメータが必要です。

以下は、Kafka コンシューマーを作成する例です。

 import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; import java.time.Duration; import java.util.Collections; import java.util.Properties; public class MyKafkaConsumer { public static Consumer<String, String> createConsumer() { Properties properties = new Properties(); properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); properties.put(ConsumerConfig.GROUP_ID_CONFIG, "my-consumer-group"); properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer"); properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer"); return new KafkaConsumer<>(properties); } }

トピックを購読する

コンシューマーは、トランザクション メッセージを含むトピックを明示的にサブスクライブする必要があります。これは、   subscribe  これを実現する方法。サブスクライブすると、コンシューマーはそのトピックに関するメッセージを受信し始めます。

 consumer.subscribe(Collections.singletonList("my-topic"));

メッセージの処理

トランザクション メッセージが到着すると、コンシューマーはメッセージが正しく処理されていることを確認する必要があります。これには通常、メッセージを処理し、データの一貫性を確保するためのロジックが含まれます。メッセージを処理するロジックは、特定のアプリケーションと要件によって異なります。

 ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100)); for (ConsumerRecord<String, String> record : records) { String key = record.key(); String value = record.value(); // 处理消息的逻辑}

変位を送信

処理されたメッセージを正しく追跡するために、コンシューマーはメッセージ オフセットをコミットする責任があります。これは、   commitSync  または  commitAsync  これを実現する方法。変位のコミットにより、メッセージが重複して処理されないことが保証されます。

 consumer.commitSync();

上記の手順は、Kafka コンシューマーを使用してトランザクション メッセージを処理する方法の基本的な例を示しています。コンシューマーとメッセージ処理を適切に構成することで、メッセージの信頼性と一貫性が確保されます。実際のアプリケーションでは、特定の要件を満たすために、メッセージを処理するロジックはより複雑になります。

III.トランザクションメッセージングのベストプラクティス

このセクションでは、Kafka トランザクション メッセージングの使用方法に関するベスト プラクティスをいくつか紹介します。これには、メッセージの 1 回限りの配信、監視とトラブルシューティング、パフォーマンスの最適化を保証する方法が含まれます。

1. メッセージの配信を確実にする

（１）生産者の冪等性

メッセージが重複して送信されるのを防ぐには、プロデューサーのべき等性を保証することが重要です。プロデューサーのべき等性を確保するために使用できるいくつかの重要な戦略とプラクティスを次に示します。

• 一意のメッセージ ID を割り当てます:  各メッセージには一意のメッセージ ID が割り当てられます。これは、グローバルに一意にすることも、特定の主題に一意にすることもできます。メッセージを送信する前に、プロデューサーは既に送信されたメッセージ レコードをチェックして、現在のメッセージの ID が重複していないことを確認できます。
• べき等 API の使用:   Kafka はべき等プロデューサー API を提供します。プロデューサー設定で冪等性を有効にするには、次のように設定します。   enable.idempotence=true設定すると、メッセージが送信時に重複して処理されないようになります。
• カスタム冪等性を実装する:  場合によっては、冪等性ロジックのカスタム実装が必要になることがあります。これには、メッセージが重複して処理されないようにするために、メッセージ処理側のデータベースまたはストレージで処理されたメッセージのステータスを追跡することが含まれます。
• 適切な再試行メカニズムを設定します。  メッセージの送信に失敗した場合、プロデューサーは、メッセージが最終的に正常に送信されるようにするための適切な再試行メカニズムを備えている必要があります。再試行メカニズムはプロデューサーの構成で設定する必要があります。

（２）消費者の重複排除

メッセージが重複して処理されないようにすることも重要です。消費者の重複排除を実装するための戦略とベスト プラクティスをいくつか紹介します。

• べき等メッセージ処理ロジック:  コンシューマーのメッセージ処理ロジックはべき等である必要があります。つまり、メッセージが何度処理されても、結果は同じになるはずです。これは通常、アプリケーション コードに実装する必要があります。
• メッセージの一意の識別子:  各メッセージにメッセージ ID などの一意の識別子を割り当てます。メッセージを処理する前に、コンシューマーは処理されたメッセージを記録するデータ構造を維持して、メッセージが繰り返し処理されないようにすることができます。
• 消費者の重複排除プロセス:  メッセージを処理する前に、コンシューマーは処理されたメッセージの記録を照会できます。メッセージがレコード内にすでに存在する場合、コンシューマーは処理をスキップするか、さらに処理を実行するかを選択できます。これにより、メッセージの重複処理が防止されます。
• コンシューマー ライブラリは以下をサポートします。  一部のメッセージ キュー処理ライブラリには組み込みの重複排除メカニズムが用意されており、これを使用して重複排除プロセスを簡素化できます。

上記のコンテンツでは、メッセージが 1 回だけ配信されるようにするための詳細な戦略とベスト プラクティスについて説明します。これは、特にトランザクション メッセージ処理において、データの一貫性を確保するための重要なステップです。これらのプラクティスは、特定のアプリケーションやニーズに基づいてカスタマイズできます。

2. トランザクションメッセージの監視とトラブルシューティング

（１）監視ツール

Kafka トランザクション メッセージを監視することは、システムの信頼性を確保する上で重要な部分です。監視ツールと戦略をいくつか紹介します。

• Kafka 組み込みメトリック: Kafka は、トランザクション メッセージのパフォーマンスとステータスを監視するための組み込みメトリックのセットを提供します。これらのメトリックを使用して、メッセージが処理される方法を追跡できます。
  
• ログ ファイル: Kafka のログ ファイルには詳細なイベント情報が含まれており、トラブルシューティングやパフォーマンス分析に使用できます。潜在的な問題がないか確認するために、定期的にログ ファイルを確認してください。
  
• 監視システム: Prometheus や Grafana などの専門的な監視システムを使用して、リアルタイムの監視とアラートを確立します。これらのシステムは、問題を検出し、タイムリーに対処するのに役立ちます。
  

（２）トラブルシューティング

トランザクション メッセージで問題が発生した場合、これらの問題をトラブルシューティングして解決できる必要があります。トラブルシューティングの戦略をいくつか紹介します。

• ログ分析: Kafka ログ ファイルを定期的に分析して、異常やエラー情報を見つけます。これにより、問題を早期に検出し、対処できるようになります。
  
• 監視アラート: 問題が発生したときにすぐに通知されるように監視アラートを設定します。これにより、問題に迅速に対応できるようになります。
  
• バージョンと構成の管理: Kafka とアプリケーションのバージョンと構成が正しく管理されていることを確認します。バージョンが異なったり、構成が矛盾していると、問題が発生する可能性があります。
  

3. トランザクションメッセージのパフォーマンスに関する考慮事項

パフォーマンスは、あらゆるメッセージング システムにとって、特に高スループットと低レイテンシの要件において重要な指標です。パフォーマンスに関する考慮事項と最適化戦略を次に示します。

• プロデューサーのパフォーマンス調整: batch.size、acks などのプロデューサーの構成パラメータを調整することで、メッセージ送信のパフォーマンスを最適化できます。
  
• コンシューマー パフォーマンスの調整: コンシューマー パフォーマンスは、max.poll.records、fetch.min.bytes などの構成パラメータを通じて調整することもできます。
  

（２）スループット最適化

• パーティションとマージ

行サイズ: システムが大量のトランザクション メッセージを処理できるように、パーティションの数とコンシューマーの並列度を適切に選択します。

• 水平スケーリング: システム負荷が増加する場合は、Kafka ブローカーとコンシューマーのインスタンスを追加してスループットを向上させる水平スケーリングを検討してください。
  
• ネットワークとストレージの最適化: ネットワークとストレージ インフラストラクチャが、高スループット メッセージングをサポートするのに十分な速度であることを確認します。
  

上記のベスト プラクティス戦略とパフォーマンス最適化の提案は、Kafka トランザクション メッセージをより適切に使用し、信頼性の高いメッセージ配信と一貫した処理を確保し、パフォーマンス要件を満たすのに役立ちます。慎重な構成、監視、トラブルシューティングを通じて、信頼性が高く、高性能なメッセージング システムを構築できます。

例: Kafka トランザクション メッセージの生成と使用

このセクションでは、Kafka トランザクション メッセージを作成および使用する方法を詳しく説明する 2 つの例を示します。

1. 例1: トランザクションメッセージの生成

例 1 コード: プロデューサー

import org.apache.kafka.clients.producer.*; import java.util.Properties; public class TransactionalProducerExample { public static void main(String[] args) { String bootstrapServers = "localhost:9092"; String topic = "my-transactional-topic"; Properties properties = new Properties(); properties.put("bootstrap.servers", bootstrapServers); properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); properties.put("acks", "all"); properties.put("enable.idempotence", "true"); properties.put("transactional.id", "my-transactional-id"); Producer<String, String> producer = new KafkaProducer<>(properties); producer.initTransactions(); try { producer.beginTransaction(); ProducerRecord<String, String> record = new ProducerRecord<>(topic, "key", "value"); producer.send(record); producer.commitTransaction(); } catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) { // Fenced, sequence issue, or authorization exception producer.close(); } catch (KafkaException e) { // Handle other exceptions producer.close(); } producer.close(); } }

コードの説明:

• この例では、Kafka プロデューサーを作成し、トランザクション メッセージをサポートするように構成し、トランザクション メッセージを生成する方法を示します。
• transactional.id は、プロデューサー トランザクションを識別するために使用される一意の ID です。トランザクション メッセージの一貫性を保証します。
• try ブロックでは、producer.beginTransaction() を使用してトランザクションを開始し、メッセージを送信し、最後に producer.commitTransaction() を使用してトランザクションをコミットします。
• トランザクション中に例外が発生した場合は、catch ブロックで例外を処理し、プロデューサーを閉じます。

2. 例2: トランザクションメッセージの消費

例 2 コード: Consumer

 import org.apache.kafka.clients.consumer.*; import java.time.Duration; import java.util.Collections; import java.util.Properties; public class TransactionalConsumerExample { public static void main(String[] args) { String bootstrapServers = "localhost:9092"; String groupId = "my-consumer-group"; String topic = "my-transactional-topic"; Properties properties = new Properties(); properties.put("bootstrap.servers", bootstrapServers); properties.put("group.id", groupId); properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); Consumer<String, String> consumer = new KafkaConsumer<>(properties); consumer.subscribe(Collections.singletonList(topic)); while (true) { ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100)); for (ConsumerRecord<String, String> record : records) { System.out.printf("Consumed record with key %s and value %s%n", record.key(), record.value()); } } } }

コードの説明:

• この例では、Kafka コンシューマーを作成し、トピックをサブスクライブし、トランザクション メッセージを消費する方法を示します。
• コンシューマーはトピックを継続的にポーリングして新しいメッセージを探します。
• 新しいメッセージが利用可能になると、そのメッセージのキーと値が出力されます。

V. 結論

この記事では、Kafka トランザクション メッセージングの主要な概念、アプリケーション シナリオ、利点、構成、使用方法、ベスト プラクティスについて詳しく説明します。まとめとして、いくつかの重要なポイントを再度強調し、Kafka のトランザクション メッセージングの将来に目を向けてみましょう。

• Kafka トランザクション メッセージングは​​、信頼性の高いメッセージ配信と処理を保証するメカニズムです。これらは、メッセージが完全に成功するか、完全に失敗するかをさらに保証します。
• アプリケーション シナリオ: Kafka トランザクション メッセージは、金融取引、在庫管理、注文処理など、高い信頼性とデータの一貫性が求められるアプリケーションで重要な役割を果たします。
• 利点: トランザクション メッセージは、非トランザクション メッセージよりも高いデータ一貫性と信頼性を提供し、原子性、べき等性、および「一度だけ」のセマンティクスをサポートします。
• 構成: トランザクション メッセージングをサポートするように Kafka を構成するには、 transactional.id 、 enable.idempotenceなどのプロデューサーとコンシューマーの設定が含まれます。
• トランザクション メッセージの生成: Kafka プロデューサーを使用する場合は、トランザクションを初期化し、メッセージを送信し、トランザクションをコミットまたは中止してメッセージの一貫性を確保する必要があります。
• トランザクション メッセージの消費: Kafka コンシューマーを使用する場合は、トピックをサブスクライブし、継続的にポーリングしてメッセージを取得し、メッセージが正しく処理されていることを確認する必要があります。
• ベスト プラクティス: ベスト プラクティスには、システムの安定性と高パフォーマンスを確保するための、1 回のみのメッセージ配信の保証、監視とトラブルシューティング、パフォーマンスの考慮事項が含まれます。