非同期通信。メッセージブローカー。Apache Kafka

この出版物は、分散システムの設計、メッセージブローカー、およびApacheKafkaに関心のある人を対象としています。ここでは、独占的な資料やライフハックは見つかりません。この記事の目的は、基礎を築き、言及されたブローカーの内部構造について説明することです。したがって、将来の出版物では、この記事へのリンクを作成して、より高度に専門化されたトピックについて話すことができるようになります。



こんにちは！私の名前はDmitryShelamovで、Vivid.Moneyでカスタマーケア部門のバックエンド開発者として働いています。当社は、ヨーロッパ諸国向けのインターネットバンキングサービスを作成および開発するヨーロッパの新興企業です。これは野心的なタスクです。つまり、その技術的な実装には、高負荷に耐え、ビジネス要件に応じて拡張できる、よく考えられたインフラストラクチャが必要です。



このプロジェクトは、さまざまな言語の数十のサービスを含むマイクロサービスアーキテクチャに基づいています。これらには、Scala、Java、Kotlin、Python、およびGoが含まれます。後者は私がコードを書く場所なので、このシリーズの実際の例では主にGo（およびいくつかのdocker-compose）を使用します。



マイクロサービスの操作には独自の特徴があり、その1つがサービス間の通信の編成です。これらの通信の相互作用モデルは 同期または 非同期であり、システム全体のパフォーマンスと障害耐性に大きな影響を与える可能性があります。

非同期通信

したがって、2つのマイクロサービス（AとB）があると想像してみましょう。それらの間の通信はAPIを介して実行され、マイクロサービスアプローチで規定されているように、相互の内部実装については何も知らないと想定します。それらの間で送信されるデータの形式は事前に合意されています。







私たちの前のタスクは次のとおりです。あるアプリケーションから別のアプリケーションへのデータの転送を、できれば最小限の遅延で整理する必要があります。

最も単純なケースでは、タスクは同期対話によって実現されます 。AがアプリケーションBに要求を送信した後、サービスBがそれを処理し、要求が正常に処理されたかどうかに応じて、応答を送信します。 この応答を期待するサービスA。

要求に対する応答が受信されていない場合、サービスAがBにその要求を繰り返すことができ、（例えば、B改応答を送信する前に接続、またはAがタイムアウトにより脱落）



一方で、そのような相互作用モデルは、要求ごとにデータ配信状態の確実性を与えます送信者が、データが受信者によって受信されたかどうか、および応答に応じてさらにどのようなアクションを実行する必要があるかを確実に知っている場合。

一方で、支払う代償は待っています。要求を送信した後、サービスA（または要求が実行されるスレッド）は、応答を受信するか、内部ロジックに従って要求が失敗したと見なすまでブロックされ、その後、さらにアクションが実行されます。



問題は、待機とダウンタイムが発生するだけでなく、ネットワーク通信の遅延が避けられないことです。主な問題は、この遅延の予測不可能性です。マイクロサービスアプローチの通信の参加者は、互いの実装の詳細を知らないため、要求が定期的に処理されているのか、データを再送信する必要があるのか​​が、要求側に常に明らかであるとは限りません。



この相互作用のモデルに残っているのは、単に待つことだけです。多分ナノ秒、多分1時間。そして、この数字は、データ処理の過程でBがビデオ処理などの重い操作を実行する場合、非常に現実的です。



おそらく、問題はあなたにとって重要ではないように思われました-一方の鉄片がもう一方の鉄片が答えるのを待っています、損失は大きいですか？

この問題をより個人的なものにするために、サービスAが電話で実行されているアプリケーションであり、Bからの応答を待っている間、画面に読み込み中のアニメーションが表示されているとします。サービスBが応答するまでアプリケーションを使い続けることはできず、待つ必要があります。不明な時間。あなたの時間は、コードの実行時間よりもはるかに価値があることを考えると。



このような粗さは次のように解決されます。インタラクションの参加者を2つの「キャンプ」に分割します。最適化しても高速に動作できないもの（ビデオ処理）と、一定時間以上待機できないもの（電話のアプリケーションインターフェイス）があります。

次に、同期を置き換えます それらの間の相互作用に（一部は、データが配信され、受信者サービスによって処理されていることを確認するために、他のを待つことを余儀なくされている場合） は非同期であること、仕事のモデルをされて 送信され、忘れられた-からの応答を待たずに、この場合には、サービスAが作業を続けますB.



しかし、この場合、転送が成功することをどのように確認できますか？たとえば、ビデオをビデオホスティングサービスにアップロードした後、ビデオをダウンロードするサービスがサービスプロセッサからビデオが到達したという確認を受信しなかったため、ユーザーに「ビデオは処理される可能性がありますが、処理されない可能性があります」というメッセージを表示することはできません。何事もなく彼。



この問題の解決策の1つとして、サービスAとサービスBの間にレイヤーを追加できます。これは、送信者と受信者にとって都合のよいレートでのデータ配信の一時的なストレージおよび保証として機能します。したがって、サービスを分離することができ ますが、その同期相互作用は潜在的に問題になる可能性があります。

• 受信サービスが異常終了したときに失われたデータは、送信サービスが作業を継続している間に、ステージングストアから再度取得できるようになりました。したがって、配信保証メカニズムを取得します。
• このレイヤーは、受信者が到着したときではなく、処理されたときにデータが提供されるため、受信者を負荷サージから保護します。
• ヘビーデューティーリクエスト（ビデオレンダリングなど）をこのレイヤーに渡すことができるようになり、アプリケーションの高速部分と低速部分の間の接続性が低下します。

通常のDBMSは、上記の要件に非常に適しています。その中のデータは、情報の損失を心配することなく長期間保存することができます。受信者の過負荷も除外されます。これは、受信者が意図したレコードを読み取るペースと量を自由に選択できるためです。処理の確認は、対応するテーブルの読み取りレコードにマークを付けることで実現できます。



ただし、データ交換ツールとしてDBMSを選択すると、負荷が増加するにつれてパフォーマンスの問題が発生する可能性があります。これは、ほとんどのデータベースがこのユースケース向けに設計されていないためです。また、多くのDBMSには、接続されたクライアントを受信者と送信者（Pub / Sub）に分離する機能がありません。この場合、データ配信ロジックをクライアント側で実装する必要があります。

おそらく、データベースよりも専門的なものが必要です。

メッセージブローカー

メッセージブローカー（メッセージキュー）は、パブ/サブモデルを使用して送信者サービスから受信者サービスにデータを保存および配信する役割を担う別個のサービスです。

このモデルは、非同期通信が2つの役割の次のロジックに従うことを前提としています。

• 出版社は、いくつかの属性によってグループ化されたメッセージとして新しい情報を公開します。
• サブスクライバーは、特定の属性を持つメッセージストリームをサブスクライブし、それらを処理します。

メッセージグループ化属性は キューです。これは、データストリームを分離するために必要です。これにより、受信者は、関心のあるメッセージグループにのみサブスクライブできます。

さまざまなコンテンツプラットフォームでのサブスクリプションと同様に、特定の作成者にサブスクライブすることで、興味のあるものだけを視聴するように選択してコンテンツをフィルタリングできます。







キューは、ライターとリーダーの間に広がる通信チャネルと考えることができます。ライターはメッセージをキューに入れ、その後、このキューにサブスクライブしているリーダーにメッセージを「プッシュ」します。 1人のリーダーが一度に1つのメッセージを受信すると、他のリーダーはそのメッセージにアクセスできなくなります。

一方、メッセージはデータの単位であり、通常はメッセージ本文とブローカーのメタデータで構成されます。



一般に、本文は特定の形式のバイトのコレクションです。

受信者は、メッセージの受信後にさらに処理するために本体を逆シリアル化できるようにするために、この形式を知っている必要があります。

任意の便利な形式を使用できますが、後方互換性について覚えておくことが重要です。これは、たとえば、バイナリProtobufおよびApacheAvroフレームワークによってサポートされています。



AMQP（Advanced Message Queuing Protocol）に基づくほとんどのメッセージブローカーは、この原則に従って動作します。この原則は、キューを介したフォールトトレラントメッセージングの標準を説明するプロトコルです。

このアプローチには、いくつかの重要な利点があります。

• 凝集力が弱い。これは、非同期メッセージ送信によって実現されます。つまり、送信者はデータをドロップし、受信者からの応答を待たずに作業を続けます。受信者は、送信されたときではなく、都合のよいときにメッセージを読み取って処理します。この場合、キューは郵便配達員があなたの手紙を置くメールボックスと比較することができ、あなたはそれがあなたに合うときにそれらを拾います。
• . , ( , ), - .
  
  , . - .
• . , , : , , , -, .
• . “at least once” “at most once”.

せいぜい一度はメッセージの再処理を排除しますが、それらが失われることを許します。この場合、ブローカーは「送信して忘れる」ベースで受信者にメッセージを配信します。受信者が何らかの理由で最初の試行でメッセージを処理できなかった場合、ブローカーはメッセージを再送信しません。 一方、



少なくとも1回は、受信者がメッセージを受信することを保証しますが、同じメッセージが再度処理される可能性があります。



多くの場合、この保証は、受信者が何らかの理由でメッセージを処理できなかった場合にメッセージを再送信するように規定するAck / Nack（確認応答/否定確認応答）メカニズムを使用して実現 されます。

したがって、ブローカーによって送信された（ただしまだ処理されていない）すべてのメッセージには、3つの最終状態があります。受信者がAckを返した （処理が成功した）、Nackを返した （処理が失敗した）、または接続を切断した。最後の2つのシナリオでは、メッセージの再送信と再処理が行われます。



ただし、受信者がメッセージを正常に処理した場合でも、ブローカーはメッセージを再送信できます。たとえば、受信者がメッセージを処理したが、Ack信号をブローカーに送信せずに終了した場合 です。

この場合、ブローカーはメッセージをキューに戻し、その後再び処理されます。開発者が受信者側で重複を排除するメカニズムを提供していない場合、エラーやデータの破損につながる可能性があります。 「exactlyonce」



と呼ばれる別の配信保証があることは注目に値し ます。分散システムで実現することは困難ですが、それも最も望ましいことです。 この点で、後で説明するApache Kafkaは、市場で入手可能な多くのソリューションの背景に対して有利に際立っています。バージョン0.11以降、Kafkaは1回限りの配信保証を提供します



クラスタおよびトランザクション内では、AMQPブローカーはそのような保証を提供できません。Kafkaでのトランザクションは別の投稿のトピックです。今日は、ApacheKafkaについて理解することから始めます。

Apache Kafka

クラスターデバイスの概略図からカフカについての話を始めることは、理解するのに役立つように思われます。







別のKafkaサーバーはブローカーと呼ばれます 。ブローカーはクラスターを形成し、これらのブローカーの1つが、管理操作の一部を引き継ぐコントローラーとして機能します（紫色でマークされています）。



ブローカーコントローラーの選択は、別のサービスの責任です。ZooKeeperは、ブローカーのサービス検出を実行し、構成を保存し、ブローカー間で新しいリーダーの配布に参加し、ほとんどの場合、各リーダーの最後に読み取られたメッセージに関する情報を保存します。これは重要なポイントであり、その調査では、1つのレベルを下げて、個々のブローカーが内部でどのように機能するかを検討する必要があります。

コミットログ

Kafkaの基礎となるデータ構造は、コミットログまたはコミットログと呼ばれます。







コミットログに追加された新しいアイテムは厳密に最後に配置され、その後の順序は変更されないため、個々のログアイテムには常に追加された順序でリストされます。



コミットログの順序付けプロパティを使用すると、たとえば、データベースレプリカ間の最終的な整合性の原則に従ってレプリケーションに使用できます。マスターノードのデータに加えられた変更のログが保存され、スレーブノードで順次適用することで、マスターと合意したものにデータを移動できます。マインド。

Kafkaでは、これらのログはパーティションと呼ばれ、ログに 保存されているデータは メッセージ。



メッセージとは何ですか？これはKafkaのデータの基本単位であり、任意の情報を渡すことができる単なるバイトのコレクションです。その内容と構造はKafkaとは無関係です。メッセージには、バイトのセットでもあるキーを含めることができます。このキーを使用すると、メッセージをパーティションに配布するためのメカニズムをより細かく制御できます。

パーティションとトピック

なぜこれが重要なのでしょうか？一見すると、パーティションはKafkaのキューに類似していないのが事実です。技術的には、メッセージキューはメッセージフローをグループ化および管理する手段であり、特定のリーダーが特定のデータストリームのみをサブスクライブできるようにすることを思い出してください。







したがって、Kafkaでは、キューの機能はパーティションではなく、トピックによって実行され ます。複数のパーティションを1つの共通のストリームに結合する必要があります。前に述べたように、パーティション自体は、コミットログデータ構造に従って順序付けられた形式でメッセージを格納します。したがって、1つのトピックに関連するメッセージは、2つの異なるパーティションに格納でき、リーダーは要求に応じてそこからそれらを引き出すことができます。



したがって、Kafkaの並列処理の単位はトピック（またはAMQPブローカーのキュー）ではなく、パーティションです。このため、Kafkaは、複数のブローカーで同じトピックに関連するさまざまなメッセージを同時に処理でき、トピック全体ではなく個々のパーティションのみを複製できるため、AMQPブローカーと比較して柔軟性とスケーラビリティが向上します。

プルアンドプッシュ

読者に関連して、誤って「プルアウト」という言葉を使用しなかったことに注意してください。

前述のブローカーでは、メッセージは、キューの形式の条件付きパイプを介して受信者にプッシュ（push）することで配信され ます。

Kafkaでは、配信プロセス自体はそうではありません。各リーダー自身が、自分が読み取るパーティションからメッセージをプル（プル）する責任が あります。







製造業者は、メッセージを作成し、それにキーとパーティション番号を添付します。メッセージにキーがない場合は、パーティション番号をランダムに選択できます（ラウンドロビン）。



さらに制御が必要な場合は、メッセージにキーを添付してから、ハッシュ関数を使用するか、メッセージのパーティションを選択するための独自のアルゴリズムを作成できます。形成後、プロデューサーはKafkaにメッセージを送信し、Kafkaはそれをディスクに保存して、どのパーティションに属しているかを記録します。



各受信者は、関心のあるトピックの特定のパーティション（または複数のパーティション）に割り当てられ、新しいメッセージが表示されると、最後に読んだメッセージに注意しながら、コミットログの次の項目を読み取る信号を受け取ります。したがって、再接続すると、次に読むメッセージがわかります。

このアプローチの利点は何ですか？

• . , , . , ( Retention Policy, ), .
• Message Replay. , . , , .
• . , ( ) – , .
• . (batch) , , . : (1 ), .

このアプローチの欠点には、問題のあるメッセージを処理することが含まれます。従来のブローカーとは異なり、壊れたメッセージ（受信者の既存のロジックを考慮して、または逆シリアル化の問題のために処理できない）は、受信者がメッセージを正しく処理することを学習するまで、無期限に再キューイングできません。



Kafkaでは、デフォルトでは、受信者が壊れたメッセージに到達するとパーティションからのメッセージの読み取りが停止し、スキップされてさらに処理するために「隔離」キュー（「デッドレターキュー」とも呼ばれます）にスローされるまで 、パーティションの読み取りを続行します。動作しないでしょう。



また、Kafkaでは、（AMQPブローカーと比較して）メッセージの優先順位を実装するのがより困難です。これは、パーティション内のメッセージが追加された順序で厳密に保存および読み取られるという事実に直接起因しています。 Kafkaでこの制限を回避する方法の1つは、events_low、events_medium、events_highなど、優先度の異なるメッセージに対していくつかのトピックを作成し（トピックは名前のみが異なります）、 コンシューマーアプリケーション側でリストされたトピックの優先度読み取りのロジックを実装することです。



このアプローチのもう1つの欠点は、各リーダーが最後に読み取ったメッセージの記録をパーティションに保持する必要があるという事実に関連しています。パーティションの構造が単純であるため、この情報はと呼ばれる整数値の形式で表示されます。 オフセット（オフセット）。オフセットを使用すると、各リーダーが現在読んでいる投稿を判別できます。オフセットに最も近い類似点は、配列内の要素のインデックスであり、読み取りプロセスは、要素のインデックスとしてイテレーターを使用してループ内の配列をウォークスルーするのと似ています。



ただし、この欠点は、バージョン0.9以降のKafkaが各ユーザーのオフセットを特別なトピック__consumer_offsetsに格納するという事実によって相殺されます （バージョン0.9までは、オフセットはZooKeeperに格納されていました）。

さらに、受信者側で直接オフセットを追跡できます。







スケーリングもより複雑になります。AMQPブローカーでは、メッセージフローの処理を高速化するために、リーダーサービスの複数のインスタンスを追加し、それらを1つのキューにサブスクライブするだけで、ブローカー自体の構成を変更する必要がないことを思い出してください。



ただし、KafkaではAMQPブローカーよりもスケーリングがやや困難です。たとえば、リーダーの別のインスタンスを追加して同じパーティションに設定すると、両方のインスタンスが同じデータセットを読み取るため、効率がゼロになります。

したがって、Kafkaスケーリングの基本的なルールは、トピックの競合するリーダー（つまり、同じ処理ロジック（レプリカ）を実装するサービスのグループ）の数がこのトピックのパーティションの数を超えてはならないということです。超えない場合、リーダーのペアが同じデータのセットを処理します。

消費者グループ

競合するリーダーが1つのパーティションを読み取る状況を回避するために、Kafkaでは、コンシューマーグループ内の1つのサービスの複数のレプリカを組み合わせて、Zookeeperが1つのパーティションに1つのリーダーのみを割り当てるのが通例 です。



リーダーはパーティションに直接バインドされているため（通常、リーダーはトピック内のパーティションの数について何も知りません）、ZooKeeperは、新しいリーダーが接続されると、コンシューマーグループの参加者を再配布して、各パーティションにリーダーが1つだけになるようにします。

読者は、Kafkaに接続するときに消費者グループを指定します。







同時に、1つのパーティションに異なる処理ロジックを持つ複数のリーダーをぶら下げることを妨げるものは何もありません。たとえば、ユーザーアクションごとのイベントのリストをトピックに保存し、これらのイベントを使用して同じデータの複数のビューを生成し（たとえば、ビジネスアナリスト、製品アナリスト、システムアナリスト、およびYarovayaパッケージの場合）、それらを適切なリポジトリに送信するとします。



しかし、ここで、Kafkaがトピックとパーティションの構造を使用しているという事実によって引き起こされる別の問題に直面する可能性があります。 Kafkaは、トピック内のメッセージの順序を保証するのではなく、パーティション内でのみ保証します。これは、たとえば、ユーザーアクションに関するレポートを生成し、それらを現状のストレージに送信する場合に重要になる可能性があります。







この問題を解決するには、逆の方法をとることができます。1つのエンティティに関連するすべてのイベント（たとえば、同じuser_idに関連するすべてのアクション）が常に同じパーティションに追加される場合、それらはトピック内で順序付けられます。同じパーティション内にあり、その順序はKafkaによって保証されています。

これを行うには、メッセージのキーが必要です。たとえば、キーからハッシュを計算してメッセージを追加するパーティションを選択するアルゴリズムを使用する場合、同じキーを持つメッセージは1つのパーティションに分類されることが保証されるため、メッセージの受信者が引き出されます。トピックに追加された順序で同じキーを使用します。

ユーザーアクションに関する一連のイベントがある場合、パーティショニングキーはuser_idにすることができます。

保持ポリシー

次に、保持ポリシーについて説明します。

これは、追加日のしきい値（時間ベースの保持ポリシー）またはディスクの占有スペース（ サイズベースの保持ポリシー）を超えたときにディスクからメッセージを削除するための設定です 。

• TBRPを7日間構成すると、7日より古いすべてのメッセージに後で削除するためのフラグが付けられます。つまり、この設定により、年齢のしきい値を下回るメッセージをいつでも読み取ることができるようになります。時間、分、ミリ秒で設定できます。
• SBRPも同様に機能します。ディスク容量のしきい値を超えると、メッセージは最後から削除するようにマークされます（古い）。メッセージの削除は瞬時に行われないため、占有されているディスク領域は常に設定で指定されているよりもわずかに大きくなることに注意してください。バイト単位で設定します。

保持ポリシーは、クラスター全体と個々のトピックの両方に対して構成できます。たとえば、ユーザーのアクションを追跡するトピック内のメッセージは数日間保存でき、プッシュ通知は数時間保存できます。関連性に応じてデータを削除することで、ディスク領域を節約します。これは、メインディスクストレージとしてSSDを選択するときに重要になる可能性があります。

圧縮ポリシー

ディスク容量を最適化するもう1つの方法は、圧縮ポリシーを使用することです。この設定では、各キーの最後のメッセージのみを保存し、前のメッセージをすべて削除できます。これは、最新の変更のみに関心がある場合に役立ちます。

Kafkaの使用例

• . : . , , , (Clickhouse !) . 
  
  Customer Care Vivid.Money CRM.
• . , . , - ( ) , , .
  
  , ( ) . , , , , .
• . , .
• (commit log). , - / .
  
  , , «» .
  
  Customer Care CRM- .

Kafka

• – , ;
• – (pull) , .  (, ) Consumer Group, ZooKeeper, , , , ;
• . , , , . , () ;
• , , AMQP , – . , ;
• . , , --, – .