RocketMQ、Kafka、Pulsar のトランザクション メッセージに関する簡単な説明

[[427099]]

導入

トランザクションはプログラム実行単位であり、その中ですべての操作が正常に実行されるか、失敗します。 RocketMQ、Kafka、Pulsar はすべて、現在業界で広く使用されているオープン ソースのメッセージ キュー (MQ) コンポーネントです。著者は職場で MQ 選択に関連するコンテンツに遭遇し、「トランザクション メッセージ」の概念は MQ コンポーネントごとに異なる意味合いを持つことを知りました。したがって、この記事では、これら 3 つのメッセージ キュー (MQ) のトランザクション メッセージ間の類似点と相違点を簡単に分析してみます。目的は、メッセージ キュー トランザクション メッセージの全体像を形成し、同様のビジネス ニーズを持つ学生に参考資料やリファレンスを提供することです。

1. メッセージキューの進化

メッセージ キュー (MQ) は、送信中にメッセージを保存するコンテナーまたはサービスを指します。これは、サーバーレスおよびマイクロサービス アーキテクチャに適した非同期のサービス間通信方式です。分散システムが高性能、高可用性、スケーラビリティなどの高度な効果を実現するための重要なコンポーネントです。一般的な主流のメッセージ キューには、ActiveMQ、RabbitMQ、ZeroMQ、Kafka、MetaMQ、RocketMQ、Pulsar などがあります。社内では、TubeMQ、Ckafka、TDMQ、CMQ、CDMQ、Hippo などがあります。

メッセージキューの進化

Kafka: Apache Kafka は、Apache Software Foundation によって開発され、Scala で記述されたオープンソースのメッセージング システム プロジェクトです。 Kafka はもともと LinkedIn によって開発され、2011 年初頭にオープンソース化されました。2012 年 10 月に Apache Incubator を卒業しました。このプロジェクトの目標は、リアルタイム データを処理するための、統合された高スループット、低レイテンシのプラットフォームを提供することです。 Kafka は、分散型、パーティション化型、マルチレプリカのログ配信サービスです。独自の設計によりメッセージング システムの機能を提供しており、全体的なアーキテクチャ図を以下に示します。

Kafka の全体的なアーキテクチャ図

RocketMQ: Apache RocketMQ は、低レイテンシ、強力な一貫性、高いパフォーマンスと信頼性、兆レベルの容量、柔軟なスケーラビリティを備えた分散メッセージングおよびストリーミング プラットフォームです。これは Kafka の設計アイデアに基づいていますが、Kafka のコピーではありません。全体的なアーキテクチャ図を以下に示します。

RocketMQ アーキテクチャ図

Pulsar: Apache Pulsar は、Apache Software Foundation のトップレベル プロジェクトであり、メッセージング、ストレージ、軽量機能コンピューティングを統合した次世代のクラウドネイティブ分散メッセージ ストリーミング プラットフォームです。コンピューティングとストレージを分離したアーキテクチャ設計を採用し、マルチテナント、永続ストレージ、複数のコンピュータ ルームでの地域間データ レプリケーションをサポートします。強力な一貫性、高スループット、低レイテンシ、高スケーラビリティなどのストリーミング データ ストレージ特性を備えています。クラウドネイティブ時代のリアルタイムメッセージストリーミング伝送、ストレージ、コンピューティングに最適なソリューションとされています。全体的なアーキテクチャ図を以下に示します。

パルサーアーキテクチャ図

2. 背景知識

2.1 トランザクションとは何ですか?

2.1.1 トランザクション

トランザクションはプログラム実行単位であり、その中ですべての操作が正常に実行されるか、失敗します。

トランザクションには 4 つの基本特性があり、これらはよく ACID と呼ばれます。

原子性: トランザクションは分割できない全体です。トランザクション内のすべての操作は、成功するか失敗するかのいずれかになります。

一貫性: トランザクションが実行される前と実行された後、データは状態間で一貫している必要があります (A が B に送金する場合、A がお金を差し引いても B が受け取らないということは起こりません)。

分離: 複数の同時トランザクションは互いに分離されており、相互に干渉することはできません。

耐久性: トランザクションが完了すると、データへの変更は永続的に保存され、ロールバックすることはできません。

2.1.2 分散トランザクション

分散トランザクションとは、トランザクション参加者、トランザクションをサポートするサーバー、リソース サーバー、およびトランザクション マネージャーが、異なる分散システムの異なるノードに配置されていることを意味します。分散トランザクションは、分散システム内の異なるノード間のデータの一貫性を確保するためによく使用されます。

分散トランザクションには、一般的に次のソリューションがあります。

XA(2PC/3PC)

最も代表的なものは、Oracle Tuxedo システムによって提案された XA 分散トランザクション プロトコルです。 XA は、トランザクション マネージャーとローカル リソース マネージャーの 2 つの部分に大まかに分かれています。ローカル リソース マネージャーは、多くの場合、データベースによって実装されます。たとえば、Oracle や DB2 などの商用データベースはすべて XA インターフェイスを実装しており、トランザクション マネージャーはグローバル スケジューラーとして各ローカル リソースの送信とロールバックを担当します。 XA プロトコルには通常、2 フェーズ コミット (2PC) と 3 フェーズ コミット (3PC) の 2 つの実装が含まれます。名前が示すように、2 フェーズ コミットには 2 つのコミット フェーズが含まれます。最初のフェーズは準備フェーズ (投票フェーズ) です。 2 番目のフェーズはコミット フェーズ (実行フェーズ) です。実装プロセスは次のとおりです。

2PC

2 フェーズ コミットはアトミック操作を提供できるように見えますが、いくつか欠陥があります。 3 フェーズ コミット (3PC) は、2 フェーズ コミット (2PC) のアップグレードおよび最適化です。興味のある方はさらに詳しく知ることができます。ここでは詳細には触れません。

TCC

TCC (Try-Confirm-Cancel) は、Try、Commit、Cancel 命令の略語であり、補償トランザクションとも呼ばれます。その論理モデルは XA 2 フェーズ コミットに似ており、トランザクション処理フローも非常に似ています。ただし、2PC は DB レベルで適用されますが、TCC はアプリケーション レベルの 2PC として理解できるため、実装するにはビジネス ロジックを記述する必要があります。

TCC の中心的な考え方は、「各操作に対して、対応する確認 (Try) と補償 (Cancel) を登録する必要がある」というものです。

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

いわゆるメッセージ トランザクションは、メッセージ キューに基づく 2 フェーズ コミットです。本質的には、メッセージ キューの特別な使用法です。ローカル トランザクションとメッセージ送信を分散トランザクションに配置して、ローカル操作が成功し、外部メッセージ送信が成功するか、または両方が失敗することを保証します。

メッセージ キューに基づく 2 フェーズ コミットは、同時実行性の高いシナリオでよく使用されます。分散トランザクションをメッセージ トランザクション (システム A のローカル操作 + メッセージの送信) + システム B のローカル操作に分割します。システム B の操作はメッセージによって実行されます。メッセージ トランザクションが成功する限り、システム A の操作は成功し、メッセージが送信される必要があります。このとき、システム B はローカル操作を実行するためのメッセージを受信します。ローカル操作が失敗した場合、システム B の操作が成功するまでメッセージは再送信されます。このようにして、システム A とシステム B の分散トランザクションが偽装されて実現されます。仕組みは次のとおりです。

メッセージトランザクション図

上記のソリューションは A と B の操作を完了できますが、A と B は強く一貫性があるわけではなく、結果的に一貫性があります。これも BASE 理論の要件を満たしています。拡張すると、BASE は、Basically Available、Soft state、Eventually Consistent の 3 つのフレーズの略語になります。 BASE 理論は、CAP における AP の拡張です (CAP では、分散システムは CAP の 3 つの項目のうち 2 つしか同時に満たすことができないことが証明されています)。強力な一貫性を犠牲にして可用性を実現します。障害が発生した場合、部分的な使用不可は許容されますが、コア機能は使用可能であることが保証される必要があります。データは一定期間不整合があっても構いませんが、最終的には整合した状態になります。 BASE理論を満たすトランザクションを「柔軟なトランザクション」と呼びます。

2.2 正確に 1 回のセマンティクスとは何ですか?

分散システムでは、どのノードでも異常が発生したり、クラッシュしたりする可能性があります。メッセージ キューでも同様です。プロデューサーがメッセージを生成しているときに、ブローカーがクラッシュして使用できなくなったり、ネットワークが突然切断されたりするなどの異常な状況が発生する可能性があります。例外が発生したときにプロデューサーがメッセージを処理する方法に応じて、システムは次の 3 つのメッセージ セマンティクスを持つことができます。

2.2.1 少なくとも1回のセマンティクス

プロデューサーは、ブローカーから ACK (メッセージ確認) 通知を受信することで、メッセージがトピックに正常に書き込まれたことを確認します。ただし、プロデューサーが ACK 通知の受信時にタイムアウトになったり、ブローカーからエラー メッセージを受信したりすると、メッセージの再送信が試行されます。ブローカーがトピックにメッセージを正常に書き込んだが、プロデューサーに ACK をまだ送信していないときにクラッシュすると、プロデューサーによって再送信されたメッセージがトピックに再度書き込まれ、最終的にメッセージがコンシューマーに繰り返し配信されることになります。つまり、メッセージは失われませんが、繰り返し送信される可能性があります。

2.2.2 最大1回のセマンティクス

プロデューサーが ACK を受信するときにタイムアウトしたり、ブローカーからエラー メッセージを受信したときにメッセージを再送信しなかったりすると、メッセージは失われ、トピックに書き込まれず、コンシューマーによって消費されない可能性があります。シナリオによっては、重複した消費を避けるために、メッセージの損失を許容することができます。つまり、メッセージは失われる可能性がありますが、複製されることはありません。

2.2.3 正確に1回だけのセマンティクス

** 正確に 1 回のセマンティクスにより、プロデューサーが同じメッセージをサーバーに複数回送信した場合でも、サーバーはそれを 1 回だけ記録します。 ** 正確に 1 回のセマンティクスは最も信頼性が高いですが、理解するのが最も困難でもあります。正確に 1 回のセマンティクスでは、メッセージ キュー サーバー、メッセージ プロデューサー、およびコンシューマー アプリケーションの連携が必要です。たとえば、コンシューマ アプリケーションがメッセージを正常に消費して ACK し、その後消費ポイントを以前のメッセージ ID にロールバックすると、そのメッセージ ID 以降のすべてのメッセージがコンシューマ アプリケーションによって再度消費されます。つまり、メッセージは失われたり、繰り返し送信されたりすることはありません。

3. RocketMQ、Kafka、Pulsar トランザクション メッセージ

3.1 RocketMQ トランザクション メッセージ

RocketMQ はバージョン 4.3.0 ですでに分散トランザクション メッセージをサポートしています。ここで、RocketMQ は 2PC の考え方を採用してトランザクション メッセージの送信を実装し、第 2 フェーズでタイムアウトまたは失敗したメッセージを処理するための補正ロジックを追加します。このプロセスを次の図に示します。

RocketMQ トランザクション メッセージ

具体的なワークフローは、通常のトランザクション メッセージの送信と提出、および異常な状況下でのトランザクション メッセージの補償プロセスに分かれています。

1. メッセージ キューでトランザクション トピックを開きます。 2. トランザクションで実行される最初のサービスは、「セミメッセージ」 (セミメッセージと通常のメッセージの唯一の違いは、トランザクションがコミットされる前はメッセージがコンシューマーに表示されないことです) をメッセージ キューに送信します。 3. セミメッセージが正常に送信された後、セミメッセージを送信したサービスはローカル トランザクションの実行を開始し、ローカル トランザクションの実行結果に基づいてトランザクション メッセージをコミットするかロールバックするかを決定します。

補償プロセス: RocketMQ は、異常な状況を解決するためにトランザクションのバックチェックを提供します。 RocketMQ がコミットまたはロールバック要求を受信しない場合、ブローカーはプロデューサーのローカル トランザクションのステータスを定期的にチェックし、プロデューサーのローカル トランザクションのステータスに基づいて「半分のメッセージ」がコミットされるかロールバックされるかを処理します。注目すべきは、独自のビジネス ロジックに従って逆クエリ ロジック インターフェイスを実装する必要があり、その後 Broker が戻り値に基づいてコミットするかロールバックするかを決定することです。さらに、このリバース クエリ インターフェイスはステートレスである必要があり、どのプロデューサー ノードへの要求でも正しいデータが返されます。 4. ローカル トランザクションが成功すると、この「セミメッセージ」は、分散トランザクションの後続のステップで独自のローカル トランザクションを実行するための通常のメッセージに変換されます。 (ここでのトランザクション メッセージの場合、プロデューサーはコンシューマーの消費失敗によりロールバックしません。トランザクション メッセージを使用するアプリケーションは、高可用性と最終的な一貫性を追求します。メッセージの消費が失敗した場合、RocketMQ は消費が成功するまでメッセージを再プッシュする責任を負います。)

補正プロセスは、メッセージのコミットまたはロールバックがタイムアウトしたり失敗したりする状況を解決するために使用されます。 RocketMQ トランザクション メッセージのメイン フローで、あるフェーズのメッセージがユーザーに表示されない仕組み。その中で、通常のメッセージと比較したトランザクションメッセージの最大の特徴は、1 段階で送信されたメッセージがユーザーには見えないという点です。では、ユーザーには見えないメッセージを作成するにはどうすればよいでしょうか? RocketMQ トランザクション メッセージの動作は次のとおりです。メッセージが「半分のメッセージ」の場合、元のメッセージのトピックとメッセージ消費キューをバックアップし、トピックを RMQ_SYS_TRANS_HALF_TOPIC に変更します。コンシューマー グループがトピックをサブスクライブしていないため、コンシューマーは「半分のメッセージ」メッセージを消費できません。次に、RocketMQ はスケジュールされたタスクを開始し、トピック RMQ_SYS_TRANS_HALF_TOPIC からメッセージをプルして消費し、プロデューサー グループに基づいてサービス プロバイダーを取得してトランザクション ステータスのクエリ要求を送信し、トランザクション ステータスに基づいてメッセージをコミットするかロールバックするかを決定します。

この時点で、ここで説明されているのは、セクション 2.1.2 で説明した分散トランザクションにおけるメッセージ トランザクションであることが誰もが理解できるはずです。目的は、分散トランザクションにおけるシステムの最終的な一貫性を実現することです。

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

RocketMQ のトランザクション メッセージングの目的とは異なり、Kafka のトランザクションは基本的に、そのべき等メカニズムと組み合わせて使用​​され、Exactly-once (セクション 2.2.3 を参照) セマンティクスを実装します。

この機能の開発理由は、次のようにまとめられます。

ストリーム処理の必要性 ストリーム処理の増加に伴い、より強力な処理保証を備えたストリーム処理アプリケーションの需要も高まっています。たとえば、金融業界では、金融機関はストリーム処理エンジンを使用して、ユーザーのローンやクレジットを処理します。このタイプのユースケースでは、各メッセージが例外なく 1 回だけ処理される必要があります。

言い換えると、ストリーム処理アプリケーションがメッセージ A を消費し、その結果をメッセージ B (B = f(A)) として生成する場合、正確に 1 回の処理保証は、B が正常に生成された場合にのみ A を消費済みとしてマークできることを意味し、その逆も同様です。

パルサー取引

トランザクション API を使用すると、ストリーム処理アプリケーションは 1 つのアトミック操作でメッセージを消費、処理、生成できます。これは、トランザクション内のメッセージのバッチを、多数のトピック パーティションから受信、生成、確認できることを意味します。トランザクションに関係するすべての操作は、全体として成功するか失敗します。

現在、Kafka が提供するデフォルトの配信信頼性保証は、少なくとも 1 回です。メッセージが正常に「送信」されたが、ブローカーの応答がプロデューサーに正常に送り返されない場合 (たとえば、ネットワークに瞬間的なジッターがある場合)、プロデューサーはメッセージが実際に正常に送信されたかどうかを判断できません。したがって、再試行することしか選択できないため、Kafka はデフォルトで少なくとも 1 回の保証を提供しますが、これによりメッセージが繰り返し送信されることになります。ほとんどのユーザーは、メッセージが失われたり繰り返し処理されたりしないように、メッセージが 1 回だけ配信されることを望んでいます。つまり、プロデューサーが同じメッセージを繰り返し送信した場合でも、ブローカーは自動的に重複を排除できます。下流の消費者の観点から見ると、メッセージは依然として 1 つだけです。それで、問題は、Kafka が正確に 1 回を達成するにはどうすればよいかということです。簡単に言えば、これはべき等性とトランザクションという 2 つのメカニズムを通じて行われます。

3.2.1 べき等プロデューサー

「べき等」という言葉はもともと数学の概念から来ており、特定の操作や関数を複数回実行しても、そのたびに得られる結果は変わらないことを意味します。べき等性には多くの利点がありますが、その最大の利点は、べき等性操作はシステム状態を破壊しないため、安全に再試行できることです。非べき等操作の場合は、特定の操作を複数回実行することによる状態への影響について依然として考慮する必要がありますが、べき等操作の場合は、これについてまったく考慮する必要はありません。

Kafka では、プロデューサーはデフォルトではべき等ではありませんが、べき等なプロデューサーを作成できます。これは実際にはバージョン 0.11.0.0 で導入された新しい機能です。 enable.idempotence を true に設定すると、プロデューサーは自動的にべき等プロデューサーにアップグレードされ、他のすべてのコード ロジックを変更する必要はありません。 Kafka は自動的にメッセージの重複を排除するのに役立ちます。冪等性を実現するために、Kafka は基盤となる設計アーキテクチャに ProducerID と SequenceNumber を導入しています。 ProducerID: 新しい各プロデューサーが初期化されると、このセッションを識別するための一意の ProducerID が割り当てられます。

SequenceNumber: 各 ProducerID について、プロデューサーがデータを送信する各 Topic および Partition は、0 から単調に増加する SequenceNumber 値に対応します。ブローカーはメモリ内に (pid、seq) マッピングを保持し、メッセージを受信した後に seq をチェックします。プロデューサーがクリア メッセージの ACK を失った場合、またはタイムアウト後に ACK を受信しなかった場合は、再試行する必要があります。

new_seq = old_seq+1: 通常のメッセージ;

new_seq <= old_seq : メッセージを繰り返します。

new_seq > old_seq+1: メッセージが失われました。

さらに、べき等性プロデューサーの範囲を理解する必要があります。まず、単一のパーティション上でのみべき等性を保証できます。つまり、べき等プロデューサーは、トピックのパーティション上に重複したメッセージが表示されないことを確認できますが、複数のパーティション上でべき等性を実現することはできません。第二に、単一セッションでのみべき等性を実現でき、セッション間では実現できません。ここでのセッションは、プロデューサー プロセスの実行として理解できます。 Producer プロセスを再起動すると、このべき等性の保証は失われます。複数のパーティションと複数のセッションでメッセージの重複をゼロにしたい場合はどうすればよいでしょうか?答えは、トランザクション、またはトランザクションプロデューサーに依存することです。これは、べき等プロデューサーとトランザクション プロデューサーの最大の違いでもあります。

3.2.2 トランザクションプロデューサー

トランザクション プロデューサーは、メッセージが複数のパーティションにアトミックに書き込まれることを保証できます。すべてのメッセージが正常に書き込まれるか、またはすべてのメッセージが失敗します。さらに、トランザクション プロデューサーはプロセスの再起動の影響を受けません。プロデューサーが再起動された後も、Kafka は送信されたメッセージの Exactly-once 処理を保証します。通常の Producer コードと比較すると、トランザクション Producer の顕著な特徴は、initTransaction、beginTransaction、commitTransaction、abortTransaction などのトランザクション API の呼び出しです。これらはそれぞれ、トランザクションの初期化、トランザクションの開始、トランザクションのコミット、トランザクションの終了に対応します。

Kafka トランザクション メッセージは、プロデューサー、トランザクション コーディネーター、ブローカー、グループ コーディネーター、コンシューマーなどの共同参加を通じて実装されます。

1) プロデューサー

プロデューサーに固定の TransactionalId を割り当てると、複数のプロデューサー セッション (プロデューサーの再起動/切断と再接続) にわたってプロデューサーの ID を継続的に識別できます。

各プロデューサーはエポックを増分します。トランザクション内の同じ TransactionalId のエポックを識別するために使用されます。トランザクションが初期化されるたびに増加し、サーバーがプロデューサー要求が古い要求であるかどうかを認識できるようにします。エポックを使用してプロデューサーの各「再生」をマークすると、同じプロデューサーが複数のセッションを持つことを防ぐことができます。

プロデューサーは、べき等メッセージの動作に従い、送信された BatchRecord にトランザクション ID とエポックを追加します。

2) 取引コーディネーター

コンシューマー グループの負荷分散のコーディネーターに似たトランザクション コーディネーターが導入されました。トランザクションを実装する各生産エンドには、トランザクション コーディネーターが割り当てられます。メッセージのトランザクション送信は、2 フェーズ送信方式で実現されます。

トランザクション コーディネーターは、特別なトピック (トランザクション トピック) を使用します。トランザクション トピック自体も永続的です。ログ情報はトランザクションのステータス情報を記録し、トランザクション コーディネータによって書き込まれます。

トランザクション コーディネーターは、R​​PC 呼び出しを通じてブローカーとコンシューマーを調整し、トランザクションの 2 フェーズ コミットを実装します。

各ブローカーはトランザクション コーディネーターを起動し、hash(TransactionalId) を使用してプロデューサーに対応するトランザクション コーディネーターを決定し、クラスター全体の負荷を分散します。

3) ブローカー

制御メッセージの導入: これらはクライアントによって生成され、トピックに書き込まれる特別なメッセージですが、コンシューマーには表示されません。これらは、ブローカーが、以前にプルされたメッセージがアトミックにコミットされたかどうかをコンシューマーに通知できるようにするために使用されます。

ブローカーは、トランザクションコーディネータのコミット/アボート制御メッセージを処理し、制御メッセージを通常のメッセージと同様にトピックに書き込み（図のcでマークされたメッセージは、トランザクションコミットのログオフセットを確認するために通常のメッセージと織り交ぜられています）、メッセージコミットオフセットhwを前方にプッシュします。

Kafka トランザクション

4) グループコーディネーター

トランザクション中に消費オフセットがコミットされると、グループ コーディネーターはトランザクション消費オフセットをオフセット ログに書き込みます。トランザクションがコミットされると、トランザクション オフセット確認メッセージがオフセット ログに書き込まれます。

5) 消費者

コンシューマーは、コミットされていないメッセージとトランザクション制御メッセージをフィルタリングし、ユーザーには見えなくなります。

実装方法は2つあります。

- コンシューマーキャッシュ方式

isolation.level=read_uncommitted を設定すると、トピックのすべてのメッセージがコンシューマーに表示されます。コンシューマーは、トランザクション制御メッセージを受信するまでこれらのメッセージをバッファリングします。トランザクションがコミットされると、これらのメッセージが公開されます。トランザクションが中止された場合、これらのメッセージは破棄されます。

- ブローカーフィルタリング方法

isolation.level=read_committed を設定します。現時点では、トピック内のコミットされていないメッセージはコンシューマーには表示されません。メッセージは、トランザクションが終了した後にのみコンシューマーに表示されます。 Broker が Consumer に送信する BatchRecord メッセージには、どのトランザクションが「中止」トランザクションであるかを示すリストが含まれます。コンシューマーは、中止トランザクションのメッセージを単純に破棄できます。

トランザクション メカニズムは、消費者が表示できるメッセージの範囲に影響するため、単純にハイ ウォーターマークに依存するわけではありません。トランザクション コンシューマーの可視性を決定するために、LSO (Log Stable Offset) と呼ばれる変位値に依存します。

3.3 パルサートランザクションメッセージ

Apache Pulsar は 2.8.0 でトランザクション関連の機能を正式にサポートします。 Pulsar が提供するトランザクションは、RocketMQ の 2PC トランザクション実装とは異なります。ローカル トランザクション バックトラッキング メカニズムは存在しませんが、これは Kafka のトランザクション実装メカニズムに似ています。 Apache Pulsar のトランザクションは主に、Pulsar 関数などのストリーム コンピューティング シナリオで正確に 1 回限りのセマンティクスの実装を保証するために使用されます。これは、エンドツーエンドのトランザクション実装のセマンティクスを保証するという、Apache Pulsar 独自のイベント ストリーミングの位置付けとも一致しています。

Pulsar では、トランザクション セマンティクスは次のように定義されています。イベント ストリーム アプリケーションは、メッセージの消費、処理、生成のプロセス全体をアトミック操作として定義できます。つまり、プロデューサーまたはコンシューマーは、複数のトピックとパーティションにまたがってメッセージを処理し、これらのメッセージが 1 つの単位として処理されるようにすることができます。

Pulsar トランザクションには次のセマンティクスがあります。

• トランザクション内のすべての操作は 1 つの単位としてコミットされます。すべてのメッセージが送信されるか、まったく送信されないかのいずれかです。
• 各メッセージは 1 回だけ書き込まれ、処理されるため、障害が発生した場合でもデータの損失や重複は発生しません。
• トランザクションが中止された場合、このトランザクション内のすべての書き込みとコミットはロールバックされます。

トランザクション内のバッチ メッセージは、複数のパーティションにわたって受信、生成、確認することができます。

• コンシューマーはコミットされた（確認された）メッセージのみを読み取ることができます。つまり、ブローカーは、オープントランザクションに属するトランザクション メッセージや中止されたトランザクションに属するメッセージを配信しません。
• 複数のパーティションにわたるメッセージの書き込みはアトミックです。
• 複数のサブスクリプションにわたるメッセージ確認はアトミックです。サブスクライブされたコンシューマーがトランザクション ID を使用してメッセージを確認する場合、メッセージは 1 回だけ正常に確認されます。

Pulsar トランザクション メッセージは、次の主要なポイントで構成されます。

1) 取引ID

トランザクション ID (TxnID) は、Pulsar 内の一意のトランザクションを識別します。トランザクション ID の長さは 128 ビットです。上位 16 ビットはトランザクション コーディネーター ID 用に予約されており、残りのビットは各トランザクション コーディネーター内で単調に増加する番号として使用されます。

2) 取引コーディネーター

トランザクション コーディネーター (TC) は、Pulsar Broker で実行されるモジュールです。

• トランザクションのライフサイクル全体を維持して、トランザクションがエラー状態になるのを防ぎます。
• トランザクション タイムアウトを処理し、トランザクション タイムアウト後にトランザクションが中止されるようにします。

3) トランザクションログ

すべてのトランザクション メタデータはトランザクション ログに保存されます。トランザクション ログは Pulsar トピックによって記録されます。トランザクション コーディネーターがクラッシュした場合、トランザクション ログからトランザクション メタデータを回復できます。

トランザクション ログには、トランザクション内の実際のメッセージではなく、トランザクションの状態が格納されます (実際のメッセージは実際のトピック パーティションに格納されます)。

4) トランザクションキャッシュ

トランザクション内でトピック パーティションに生成されたメッセージは、そのトピック パーティションのトランザクション バッファー (TB) に格納されます。トランザクション バッファ内のメッセージは、トランザクションがコミットされるまでコンシューマーには表示されません。トランザクションが中止されると、トランザクション バッファ内のメッセージは破棄されます。

トランザクション バッファは、進行中および中止されたすべてのトランザクションをメモリに格納します。すべてのメッセージは、実際にパーティション化された Pulsar トピックに送信されます。トランザクションがコミットされると、トランザクション バッファー内のメッセージがコンシューマーに対して具体化 (表示) されます。トランザクションが中止されると、トランザクション バッファ内のメッセージは破棄されます。

5) 確認待ち

保留中の確認応答状態では、トランザクションが完了するまで、トランザクション内のメッセージの確認応答が維持されます。メッセージが保留中の確認応答状態にある場合、メッセージが保留中の確認応答状態から削除されるまで、他のトランザクションはメッセージを確認応答できません。

保留中の確認のステータスは、保留中の確認ログ (カーソル台帳) に保存されます。新しく起動されたブローカーは、保留中の確認ログから状態を回復して、状態の確認が失われないようにすることができます。

処理フローは一般的に次のステップに分かれます。

1. 取引を開きます。
2. トランザクションを使用してメッセージを公開します。
3. トランザクション確認メッセージを使用します。
4. 取引を終了します。

Pulsar のトランザクション処理プロセスは、Kafka のトランザクション処理の考え方とほぼ一致しています。誰もが TC と、すべての TC 操作を永続化してトランザクション ステータスの変更のすべての要求を記録するための対応するトピックを持っています。同様に、トランザクションの開始時には、TC に対応するオーナー ブローカーの場所を照会するための専用のトピックがあります。違いは、まず、Kafka の未確認メッセージはブローカー側で管理されますが、Pulsar の未確認メッセージはクライアント側で管理されることです。トランザクション タイムアウトは、トランザクションが正常に実行されたかどうかを判断するために使用されます。したがって、トランザクション タイムアウトの存在により、クライアント側とブローカー側のトランザクション処理の一貫性が保証されます。 2 つ目: Kafka 自体には単一のメッセージに対する Ack がないため、Kafka のトランザクション処理は順番にしか実行できません。トランザクション要求がブロックされると、後続のすべてのトランザクション要求がブロックされます。ただし、Pulsar はメッセージに対して単一の Ack を実行できるため、ここでの各トランザクションの Ack アクションは独立しており、トランザクションがブロックされることはありません。

IV.結論

RocketMQ と Kafka/Pulsar のトランザクション メッセージの実際のシナリオは異なります。

RocketMQ のトランザクションは、ローカル トランザクションの実行とメッセージの送信という 2 つの操作が成功するか失敗するかを保証するという問題を解決します。さらに、RocketMQ では、トランザクション実行の成功率とデータの一貫性を最大化するために、トランザクション バックチェック メカニズムが追加されました。

Kafka のトランザクションは、トランザクションで送信された複数のメッセージが成功するか失敗するかを保証するという問題を解決します。 (ここでの複数のメッセージは、必ずしも同じトピックとパーティション内にある必要はなく、複数のトピックとパーティションに送信されたメッセージでもかまいません。) もちろん、Kafka トランザクションの実行中にローカル トランザクションを開始して、RocketMQ トランザクション メッセージと同様の効果を実現することもできますが、Kafka にはトランザクション メッセージの逆クエリ メカニズムはありません。直接例外をスローします。ユーザーは、例外に基づいて独自の再試行メソッドを実装し、トランザクションの正常な操作を確保できます。

これらに共通するのは、すべてが 2 フェーズ コミットを通じてトランザクションを実装し、トランザクション メッセージが別々のトピックに保存されることです。違いは、RocketMQ は「セミメッセージ」を通じて実装されるのに対し、Kafka は対応するトピックにメッセージを直接送信し、クライアントを通じてフィルタリングすることです。さらに、それらが使用されるシナリオは非常に異なります。 RockteMQ は主にローカル トランザクションとメッセージに基づいてデータの一貫性を解決しますが、Kafka のトランザクションはリアルタイム ストリーム コンピューティング シナリオに適用される Exactly-once メカニズムを実装するために使用されます。

Pulsar のトランザクション メッセージは Kafka のアプリケーション シナリオおよびセマンティクスに似ていますが、基盤となる実装メカニズムの違いにより、詳細が一部異なります。

今では非常に明確になっていると思います。トランザクション メッセージを選択して適用する方法については、まずビジネス ニーズを理解する必要があります。分散トランザクションの最終的な一貫性を実現したいですか、それとも Exactly-once セマンティクスを実現したいですか?要件を理解すれば、どのコンポーネントを選択すればよいかが非常に明確になります。

参考文献

パルサー公式ドキュメント

メッセージ キュー: メッセージ キューを使用して分散トランザクションを実装するにはどうすればよいですか?