Kafka はメッセージ損失の問題をどのように解決しますか?

[[415220]]

この記事はWeChatの公開アカウント「Micro Technology」から転載したもので、著者はMicro Technologyです。この記事を転載する場合は、Micro Technology の公開アカウントにご連絡ください。

みなさんこんにちは、トムです〜

誰もが Kafka メッセージング フレームワークに精通しており、多くの人が仕事でそれに触れたことがあるはずです。その中心的なアイデアは、高性能 MQ サービスを通じて生産システムと消費システムを接続し、強力なスケーラビリティを備えたシステム間の分離を実現することです。

リンクの 1 つが壊れていたらどうなるのかと疑問に思うかもしれません。

この状況はメッセージ損失と呼ばれ、システム間でデータの不整合が発生します。

では、この問題をどう解決すればよいのでしょうか?これを、プロダクション側、MQ サーバー側、コンシューマー側の 3 つの側面から対処する必要があります。

1. 生産

生成側の責任は、生成されたメッセージが MQ サーバーに到達できるようにすることです。ここでは、操作が成功したかどうかを判断するための応答が必要です。

 Future<RecordMetadata> 送信(ProducerRecord<K, V> レコード、コールバック コールバック)

たとえば、上記のコードでは、コールバック関数を使用して、メッセージが正常に送信されたかどうかを判断します。失敗した場合は補償する必要があります。

さらに、送信の柔軟性を向上させるために、Kafkaはさまざまなビジネスが選択できるさまざまなパラメータを提供します。

1.1 パラメータ確認

このパラメータは、メッセージが正常に送信されたと判断される前にメッセージを受信したパーティション レプリカの数を示します。

acks=0の場合、メッセージが送信されれば成功とみなされ、プロデューサーはサーバーノードの応答を待たない。

acks=1、プロデューサーはリーダーパーティションからの応答を受信したときに送信が成功したとみなすことを示します。

acks=-1 の場合、プロデューサーは ISR 内のすべてのレプリカがメッセージを受信した場合にのみ成功と見なします。この構成は最も安全ですが、同期されるノードが増えるためスループットが低下します。

1.2 パラメータの再試行

運用側での再試行回数を示します。再試行回数が尽きてもメッセージが失敗した場合、メッセージはローカル ディスクに一時的に保存され、サービスが復旧した後に再送信されます。推奨値: retries=3

1.3 パラメータ retry.backoff.m

メッセージ送信のタイムアウトまたは失敗後の再試行間隔。一般的に推奨されるセットアップ時間は 300 ミリ秒です。

ここでは、特別な状況に特別な注意を払う必要があります。 MQ サービスが正常に応答しない場合でも、必ずしもメッセージの送信が失敗したことを意味するわけではありません。応答がネットワーク ジッターと一致し、応答がタイムアウトする可能性もあります。

制作側でこれらすべてを実行すると、メッセージが正常に送信されることが保証されますが、メッセージが複数回送信される可能性があり、メッセージが重複することになります。解決策については後で話し合います。

2. MQサーバー

メッセージの保存媒体として、MQ サーバーでもメッセージが失われる可能性があります。たとえば、パーティションが突然クラッシュした場合、このパーティション内のデータが失われないようにするにはどうすればよいでしょうか?この問題をバックアップを通じて解決するために、レプリカの概念を紹介します。

どのようなパラメータを設定できますか?

2.1 パラメータ replication.factor

パーティション レプリカの数 (replication.factor > 1) を示します。リーダー レプリカに障害が発生すると、フォロワー レプリカがリーダーとして選出され、サービスの提供を継続します。

2.2 パラメータ min.insync.replicas

ISR のレプリカの最小数を示します。通常、min.insync.replicas > 1 が設定され、置換を実行してメッセージが失われないようにするために、使用可能なフォロワー レプリカが存在するようになります。

2.3 パラメータ unclean.leader.election.enable

非 ISR セット内のレプリカをリーダ​​ー レプリカとして選出できるかどうか。

true に設定され、フォロワー レプリカの同期メッセージの進行が大幅に遅れている場合、この時点でリーダーとして選出されると、メッセージが失われます。注意してご使用ください。

3. 消費者側

消費者が行う必要があるのは、メッセージを完全に消費して処理することです。しかし、移転を提出する手順があります。

ビジネス処理には長い時間がかかることを考慮して、別のスレッドを開始してメッセージをプルし、ローカル メモリ キューに格納してから、スレッド プールを設定してビジネス ロジックを並列処理する学生もいます。この設計にはリスクが伴います。ローカル メッセージが完全に処理されずにサーバーがクラッシュすると、メッセージは失われます。

正しいアプローチ: メッセージをプル --- ビジネス処理 --- 消費変位を送信

コミット変位に関しては、Kafkaは集中的なパラメータ設定を提供する。

パラメータ enable.auto.commit

消費変位が自動的に送信されるかどうかを示します。

メッセージがプルされたがビジネス ロジックが処理されていない場合、消費変位が送信されたがコンシューマー側がダウンしている場合、コンシューマー側が回復するか、他のコンシューマーがシャードを引き継いでメッセージをプルできなくなり、メッセージが失われます。したがって、通常は enable.auto.commit=false を設定し、消費変位を手動でコミットします。

リスト<文字列>メッセージ = consumer.poll();
 processMsg(メッセージ);
コンシューマー.commitOffset();

この解決策は別の問題を引き起こします。この写真を見てみましょう:

メッセージ4～8を取得して業務処理を行った後、消費変位を送信するとシステムがクラッシュしました。最終送信変位は MQ サーバーに保存されませんでした。次にメッセージがプルされたとき、メッセージは依然としてメッセージ 4 から開始されますが、メッセージのこの部分は処理されているため、重複した消費が発生します。

重複消費を解決し、データの不整合を回避する方法

まず、MQ サーバー上の重複メッセージを解決する必要があります。 Kafka バージョン 0.11.0 以降では、各メッセージには一意のメッセージ ID が付きます。 MQ サービスは、スペース・フォー・タイムを使用して重複メッセージを自動的にフィルタリングし、インターフェースの冪等性を保証します。

しかし、これではメッセージの重複の問題を根本的に解決することはできません。 MQ サービスに重複したメッセージが格納されていない場合でも、コンシューマー側はプル方式を使用します。メッセージが繰り返しプルされると、重複した消費にもつながります。このシナリオの問題をどのように解決するのでしょうか?

解決策 1: 一度だけプルします (コンシューマーがメッセージをプルした後、メッセージを処理する前にオフセットを送信します)。しかし、システムがクラッシュし、業務処理が正常に完了しなかった場合、これらのメッセージは再度取得されなくなり、データの不整合が発生します。このソリューションはほとんど使用されません。

解決策 2: 重複メッセージのプルを許可しますが、コンシューマー側で冪等性制御自体を実行します。一度だけ消費されることが保証されています。

べき等性のある技術的ソリューションは数多くあります。処理識別子を保存するには、データ テーブルまたは Redis キャッシュを使用できます。メッセージがプルされるたびに、処理前に処理ステータスが検証され、その後、メッセージを処理するか破棄するかが決定されます。