Go 言語を使用して Kafka を操作し、メッセージの損失を防ぐ方法

[[423396]]

背景

現在、一部のインターネット企業は、コアビジネスを実行するためにメッセージキューを使用しています。コアビジネスであるため、データの最終的な一貫性には敏感です。途中でデータが失われると、ユーザーからのクレームにつながり、年末の業績は325になります。以前、数人の友人とチャットしていました。これらの企業はすべて、メッセージキューとして Kafka を使用しています。 Kafka を使用するとメッセージは失われますか?メッセージが失われた場合、どのように補償措置をとるのでしょうか?今回は一緒に解析していき、Goを使ってデータを失うことなくKafkaを操作する方法を紹介します。

この記事は、https://github.com/Shopify/sarama に基づいて Kafka を操作します。

Kafka アーキテクチャ入門

Wikipedia による Kafka の紹介:

Kafka は、Apache Software Foundation によって開発され、Scala と Java で記述されたオープンソースのストリーム処理プラットフォームです。このプロジェクトの目標は、リアルタイムデータを処理するための、統合された高スループット、低レイテンシのプラットフォームを提供することです。その永続性レイヤーは本質的に「分散トランザクションログアーキテクチャを備えた大規模なパブリッシュ/サブスクライブメッセージキュー」であり、ストリーミングデータを処理するためのエンタープライズレベルのインフラストラクチャとして非常に価値があります。さらに、Kafka は Kafka Connect を介して外部システムに接続 (データ入出力用) することができ、Java ストリーム処理ライブラリである Kafka Streams も提供します。この設計はトランザクションログに大きく影響されます。

Kafka の全体的なアーキテクチャは比較的シンプルで、主にプロデューサー、ブローカー、コンシューマーで構成されています。

スクリーンキャプチャ 2021-09-12 午前10時00分13秒

アーキテクチャ図に従って各モジュールを説明します。

プロデューサー: 選択したトピックにデータを公開できるデータのプロデューサー。
コンシューマー: データコンシューマーはコンシューマーグループによって識別されます。トピック内の各レコードは、サブスクリプションコンシューマーグループ内のコンシューマーインスタンスに割り当てられます。コンシューマーインスタンスは、複数のプロセスまたは複数のマシンに分散できます。
ブローカー: メッセージミドルウェア処理ノード (サーバー)。ノードはブローカーです。 Kafka クラスターは 1 つ以上のブローカーで構成されます。

他にもいくつか概念を紹介します:

トピック: メッセージの集合として理解できます。トピックはブローカーに保存されます。トピックには複数のパーティションを含めることができます。トピックには、メッセージをプッシュする複数のプロデューサーが存在する場合があります。トピックには、そこからメッセージをプルする複数のコンシューマーが存在する場合があります。トピックは 1 つ以上のブローカーに存在できます。
パーティション: トピックのサブセットです。異なるパーティションが異なるブローカーに割り当てられ、水平拡張することで Kafka の並列処理機能が向上します。同じトピックの下にある異なるパーティション情報は異なり、同じパーティション情報は順序付けられます。各パーティションには 1 つ以上のレプリカがあり、その中からリーダーが選出されます。ファウラーはリーダーからデータを取得して自身のログを更新し (各パーティションは論理的にログフォルダーに対応します)、コンシューマーはリーダーから情報を取得します。

Kafka がメッセージを失う 3 つのノード

プロデューサープッシュメッセージノード

まず、プロデューサーの一般的な執筆プロセスを見てみましょう。

プロデューサーはまずKafkaクラスタからパーティションのリーダーを見つける
プロデューサーはリーダーにメッセージを送信し、リーダーはメッセージをローカルに書き込む
フォロワーはリーダーからメッセージをプルし、ローカルログに書き込み、リーダーはackを送信する。
リーダーは ISR 内のすべてのレプリカから ACK を受信した後、高水準点を上げてプロデューサーに ACK を送信します。

スクリーンキャプチャ 2021-09-12 午前11時16分43秒

このプロセスを通じて、Kafka が最終的に ack を返してプッシュメッセージの結果を確認することがわかります。ここで Kafka は次の 3 つのモードを提供します。

応答なし 必要なAcks = 0
ローカル要求されたAcksの待機 = 1
 WaitForAll 必須Acks = -1

NoResponse RequiredAcks = 0: これは、データプッシュの成功または失敗は私に関係ないことを意味します
WaitForLocal RequiredAcks = 1: ローカル (リーダー) がメッセージが正常に受信されたことを確認すると、戻ることができます。
WaitForAll RequiredAcks = -1: すべてのリーダーとフォロワーがメッセージを正常に受信した場合にのみ返されます。

したがって、これら 3 つのモードに基づいて、プロデューサーがメッセージをプッシュするときに一定の損失の可能性があると推測できます。分析は次のとおりです。

モード 1 を選択した場合、このモードではデータが失われる可能性が高く、再試行できません。
モード 2 を選択した場合、このモードでは、リーダーがダウンしていない限り、データが失われないことが保証されます。ただし、リーダーがダウンし、フォロワーがデータを同期していない場合、データが失われる可能性が一定程度あります。
モード 3 を選択した場合、この状況ではデータが失われることはありませんが、データが重複する可能性があります。リーダーとフォロワーがデータを同期するときにネットワークの問題が発生すると、データの重複が発生する可能性があります。

したがって、実稼働環境では、メッセージの信頼性を確保するためにモード 2 またはモード 3 を選択できます。具体的な選択はビジネスシナリオに基づいて行う必要があります。スループットを重視する場合は、モード 2 を選択します。スループットを気にしない場合は、モード 3 を選択します。データが失われないように完全に保証したい場合は、最も信頼性の高いモード 3 を選択します。

Kafkaクラスタ自体の障害が原因である

データを受信した後、Kafka クラスターはデータを永続的に保存し、最終的にデータはディスクに書き込まれます。ディスクへの書き込み時に、オペレーティングシステムはまずデータをキャッシュに書き込むため、ディスクへの書き込み手順によってデータが失われる可能性もあります。オペレーティングシステムがキャッシュ内のデータをディスクに書き込むタイミングは不確実です。したがって、この場合、Kafka マシンが突然クラッシュすると、データ損失も発生します。しかし、これが起こる可能性は非常に低いです。一般的には、社内の Kafka マシンがバックアップされます。この状況は非常に極端であり、無視することができます。

コンシューマープルメッセージノード

メッセージがプッシュされると、データがパーティションに追加され、オフセットが割り当てられます。このオフセットは、現在のコンシューマーによって消費される場所を表します。このパーティションを通じてメッセージの順序も保証されます。コンシューマーがメッセージをプルした後、自動送信または手動送信を設定できます。送信が成功すると、オフセットは次のようにシフトします。

スクリーンキャプチャ 2021-09-12 午後3時37分33秒

したがって、自動送信ではデータが失われ、手動送信ではデータが重複することになります。分析は次のとおりです。

自動コミットを設定すると、メッセージをプルするときにオフセットはコミットされますが、消費ロジックの処理時に失敗し、データが失われます。
手動送信を設定する場合、メッセージを処理した後にコミットを送信すると、コミットステップで障害が発生すると、重複消費の問題が発生します。

データ損失と比較すると、重複した消費はビジネスの期待に沿ったものです。いくつかのべき等設計によってこの問題を回避できます。

実際の戦闘

完全なコードはgithubにアップロードされています: https://github.com/asong2020/Golang_Dream/tree/master/code_demo/kafka_demo

プッシュメッセージの損失問題を解決する

これは主に次の 2 つの方法で解決されます。

この問題は、RequiredAcks モードを設定することで解決できます。 WaitForAll を選択すると、データが正常にプッシュされることが保証されますが、レイテンシに影響します。
再試行メカニズムを導入し、再試行回数と再試行間隔を設定する

そこで、次のコードを記述します (クライアント作成部分を抽出します)。

関数 NewAsyncProducer() sarama.AsyncProducer {
 cfg := sarama.NewConfig()
 バージョン、エラー:= sarama.ParseKafkaVersion(VERSION)
 err != nil{の場合
  log.Fatal( "NewAsyncProducer の kafka バージョンの解析に失敗しました" 、 err.Error())
ゼロを返す
 }
 cfg.Version = バージョン
 cfg.Producer.RequiredAcks = sarama.WaitForAll // 選択できる 3 つのモード
 cfg.Producer.Partitioner = sarama.NewHashPartitioner
 cfg.プロデューサー。 .Successes = trueを返します 
 cfg.プロデューサー。 .Errors = trueを返します 
 cfg.Producer.再試行。最大= 3 // 再試行を3回に設定
 cfg.Producer.Retry.Backoff = 100 *時間.ミリ秒
 cli、エラー:= sarama.NewAsyncProducer([]string{ADDR}、cfg)
 err != nil{の場合
  log.Fatal( "NewAsyncProducer が失敗しました" , err.Error())
ゼロを返す
 }
 CLIを返す
}

プルメッセージの損失問題を解決する

この解決策はかなり粗雑です。自動コミットモードを直接使用し、消費ごとにオフセットを手動でコミットします。しかし、重複消費の問題が発生します。ただし、べき等演算を使用すると簡単に解決できます。

コード例:

 func NewConsumerGroup(グループ文字列) sarama.ConsumerGroup {
 cfg := sarama.NewConfig()
 バージョン、エラー:= sarama.ParseKafkaVersion(VERSION)
 err != nil{の場合
  log.Fatal( "NewConsumerGroup の kafka バージョンの解析に失敗しました" 、 err.Error())
ゼロを返す
 } 
 
 cfg.Version = バージョン
 cfg.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRange
 cfg.Consumer.Offsets.Initial = sarama.OffsetOldest
 cfg.Consumer.Offsets.Retry.Max = 3
 cfg.Consumer.Offsets.AutoCommit.Enable = true // 自動コミットを有効にするには、MarkMessage を手動で呼び出す必要があります。
 cfg.Consumer.Offsets.AutoCommit.Interval = 1 *時間。秒// 間隔
 クライアント、エラー:= sarama.NewConsumerGroup([]string{ADDR}、グループ、cfg)
 err != nil の場合 {
  log.Fatal( "NewConsumerGroup が失敗しました" , err.Error())
 }
顧客を返す
}

上記は主に ConsumerGroup 部分の作成に関するものです。注意深い読者は、ここで自動送信を使用していることに気付いたはずです。手動での送信はどうですか?これは、kafka ライブラリの特性が異なるためです。この自動送信は、送信するために MarkMessage() メソッドと組み合わせて使用する必要があります (質問のある友人はそれを練習したり、ソースコードを確認したりできます)。そうでない場合、消費ロジックを次のように記述する必要があるため、送信は失敗します。

 func (e EventHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) エラー {
 msg := range の場合、claim.Messages() {
  var データ common.KafkaMsg
  エラーの場合:= json.Unmarshal(msg.Value, &data);エラー != ゼロ {
 errors.Newを返します( "メッセージのアンマーシャルに失敗しました。err は " + err.Error())
  }
  // データを操作し、代わりに印刷を使用する
  log.Print( "consumerClaimデータは " ) 
 
  // メッセージが正常に処理されたら、処理済みとしてマークし、自動的に送信します
  session.MarkMessage(メッセージ、 "" )
 }
ゼロを返す
}

または、手動で送信する方法を直接使用して問題を解決することもできます。これには 2 つの手順のみが必要です。

ステップ 1: 自動送信をオフにする:

 consumerConfig.Consumer.Offsets.AutoCommit.Enable = false // 自動コミットを無効にして手動に変更します

ステップ 2: 消費ロジックに次のコードを追加します。手動送信モードでは、コミットする前にマークする必要もあります。

 session.MarkMessage(メッセージ、 "" )
セッション.コミット()

完全なコードは GitHub からダウンロードして検証できます。

要約する

この記事では、主に次の 2 つの知識ポイントについて説明します。

Kafka はメッセージ損失を引き起こす可能性がある

データ損失を回避するために Go で Kafka を設定する方法

日常のビジネス開発では、多くの企業が分離のためにメッセージキューを使用しています。そうなると注意を払わなければなりません。 Kafka をメッセージキューとして使用しても、データが失われないことは保証されません。補償は手動で設定する必要があります。忘れないでください。そうしないと、別の P0 事故が発生します。

<<: ウェブサイトを構築する前に、クラウドサーバーと仮想ホストの4つの違いを見てみましょう

>>: クラウドコンピューティングはどのように進化するのでしょうか?