Kafkaノート-信頼性、累乗性、トランザクション

ここ数日は忙しいですが、今私に要求しているのは1週間に少なくとも1つの文章を出すことです.だから、まずこのノートを持って前菜を作ります.ソース分析では、明日から2日後に1つ書けると思います.誰も見ていなくても、自分を応援してください.

しんらいせい

メッセージが失われないことを保証する方法

Kafkaは「コミットされた」メッセージ(committed message)に対して限られた永続化保証を行う.
コミットされたメッセージKafkaの複数のBrokerがメッセージを正常に受信し、ログファイルに書き込むと、プロバイダにメッセージが正常にコミットされたことを通知します.
限られた永続化は、1つのメッセージがN個のKafka Brokerに保存されている場合、少なくともこのN個のBrokerが少なくとも1個生存してこそ、メッセージが失われないことを保証することができることを保証する.

データ消失事例

生産者プログラム損失データ

Kafka Producerは非同期送信であるため、producerを呼び出す.send(msg)はメッセージの送信に成功したとは考えられない.
したがって、Producerではコールバック通知付きの送信APIを常に使用し、producerを使用する.send(msg,callback).メッセージの送信に失敗すると、的確に処理することができる.

消費者側のデータ損失

消費者はoffsetを更新してから、メッセージを消費します.この時、消費者が突然ダウンタイムになったら、このニュースは失われます.
だから私たちはまずメッセージを消費して、offsetの位置を更新します.しかし、これにより、メッセージの重複消費が発生します.
もう1つは、consumerがメッセージを取得した後、複数のスレッドが非同期にメッセージを処理し、consumerがoffsetを自動的に更新する場合である.スレッドのいずれかが失敗すると、メッセージが失われます.
このような場合、consumerは自動コミットシフトをオンにするのではなく、アプリケーションに手動でシフトをコミットします.

さいてきインプリメンテーション

producerを使用する.send(msg,callback).

acks=allを設定します.acksはProducerのパラメータであり、すべてのコピーBrokerがメッセージを受信することを意味し、このメッセージは「コミットされた」とみなされます.

retriesを大きな値に設定します.Producerのパラメータで、Producerに対応して自動的に再試行されます.ネットワークジッタが発生した場合、メッセージの送信を自動的に再試行し、メッセージの損失を回避できます.

unclean.leader.election.enable = false.どのBrokerがパーティションに立候補する資格があるかをコントロールするリーダー.あまり遅れているBrokerのリーダー選挙は許されないと述べた.

replicationを設定.factor>=3.Brokerパラメータ、冗長Broker.

min.insyncを設定.replicas>1.Brokerパラメータ.制御メッセージが少なくとも何個のコピーに書き込まれるかを「コミット済み」とします.

replicationを確保する.factor>min.insync.replicas.2つが等しい場合、コピーマウントが1つある限り、パーティション全体が正常に動作しません.推奨設定はreplication.factor=min.insync.replicas+1.

メッセージ消費がコミットされていることを確認します.Consumer側パラメータenbale.auto.commit、falseに設定し、シフトを手動でコミットします.

第2条と第6条を説明する:ISRに1つのコピーしかない場合、acks=allはacks=1に相当し、min.insyncを導入する.replicasの目的は、ISRのすべての書き込みのみを満たすことができず、ISRの書き込み個数がmin.insyncより少なくないことを保証する下限の制限を行うことである.replicas.

べき乗等性

0.11.0.0リリースでは、べき乗等性Producerを作成する機能が導入されています.propsを設定するだけです.put(「enable.idempotence」,true)、またはprops.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,true).
enable.idempotenceがtrueに設定されると、Producerは自動的にべき乗等性Producerにアップグレードされます.Kafkaは自動的に重さを落とす.Brokerはフィールドを多く保存します.Producerが同じフィールド値のメッセージを送信すると、Brokerはこれらのメッセージが重複していることを自動的に知ることができる.
有効範囲:

は、単一パーティション上のべき乗等性のみを保証します.すなわち、1つのべき乗等性Producerは、あるトピックの1つのパーティションに重複メッセージが表示されないことを保証します.

では、単一のセッションのべき乗等性しか実現できません.ここでのセッションは、Producerプロセスの1回の実行を指します.Producerプロセスを再起動した後、べき乗等性は保証されません.

取引

Kafkaは0.11リリースでトランザクションのサポートを開始し、read committed独立性レベルのトランザクションを提供します.複数のメッセージがターゲットパーティションに原子的に書き込まれることを保証し、Consumerはトランザクションが正常にコミットされたメッセージしか表示されないことを保証します.

トランザクション・プログラム

複数のメッセージが複数のパーティションに原子的に書き込まれることを保証します.このメッセージはすべて成功するか、すべて失敗しないか.トランザクションProducerもプロセスの再起動を恐れません.
Producer側の設定:

オープンenable.idempotence = true

Producerエンドパラメータtransactional.id

を設定する.
さらに、initTransaction、beginTransaction、commitTransaction、abortTransactionなどの呼び出しトランザクションAPIも追加され、トランザクションの初期化、トランザクションの開始、トランザクションのコミット、およびトランザクションの終了に対応します.次のようになります.

producer.initTransactions();
try {
            producer.beginTransaction();
            producer.send(record1);
            producer.send(record2);
            producer.commitTransaction();
} catch (KafkaException e) {
            producer.abortTransaction();
}

このコードは、record 1とrecord 2が同じトランザクションとしてKafkaにコミットされるか、すべて成功するか、すべての書き込みに失敗するかを保証します.
Consumer側の設定:isolationを設定する.レベルパラメータ、現在2つの値があります.

read_uncommitted:デフォルト値は、Consumer側がトランザクション型Producerがトランザクションをコミットしても終了しても、書き込まれたメッセージを読み取ることができることを示しています.

read_committed:Consumerはトランザクション型Producerがトランザクション書き込みを正常にコミットしたメッセージのみを読み込むことを示します.トランザクション以外のProducerが書き込むすべてのメッセージが表示されることに注意してください.