カフカのタイミングホイール

[[231861]]

Kafka には、遅延生成、遅延プル、遅延削除など、多数の遅延操作があります。 Kafka は、遅延機能を実装するために JDK に付属する Timer または DelayQueue を使用せず、代わりにタイムホイールに基づいてタイマー (SystemTimer) をカスタマイズして遅延機能を実装します。 JDK の Timer および DelayQueue の挿入および削除操作の平均時間計算量は O(nlog(n)) であり、Kafka の高パフォーマンス要件を満たすことができません。ただし、挿入および削除操作の時間計算量は、タイムホイールに基づいて O(1) に削減できます。タイムホイールの応用は Kafka に特有のものではありません。他にも多くの応用シナリオがあります。タイムホイールは、Netty、Akka、Quartz、Zookeeper などのコンポーネントに含まれています。

下の図を参照してください。 Kafka の TimingWheel は、時間制限のあるタスクを格納するための循環キューです。基礎となるレイヤーは配列を使用して実装されます。配列内の各要素には、時間制限付きタスクリスト (TimerTaskList) を格納できます。 TimerTaskList は循環的な双方向リンクリストです。リンクリスト内の各項目は、実際のタイマータスク TimerTask をカプセル化するタイマータスク項目 (TimerTaskEntry) を表します。

タイムホイールは複数のタイムグリッドで構成されており、各グリッドは現在のタイムホイールの基本的な時間範囲 (ティック) を表します。タイミングホイールの時間グリッドの数は固定されており、wheelSize で表現できます。次に、タイミングホイール全体の全体的な時間範囲 (間隔) を、式 tickMs × wheelSize で計算できます。タイムホイールには、タイムホイールの現在の時刻を示すダイヤルポインター (currentTime) もあります。 currentTime は tickMs の整数倍です。 currentTime は、タイムホイール全体を期限切れ部分と期限切れでない部分に分割できます。現在 currentTime によって指されている時間グリッドも期限切れ部分に属しています。つまり、期限が切れたばかりであり、この時間グリッドに対応する TimerTaskList のすべてのタスクを処理する必要があることを意味します。

タイムホイールの tickMs が 1ms で wheelSize が 20 の場合、間隔は 20ms と計算されます。最初、ダイヤルポインタ currentTime はタイムグリッド 0 を指しています。この時点で、タイマーが 2 ミリ秒のタスクが挿入され、タイムグリッド 2 の TimerTaskList に格納されます。時間が経過するにつれて、ポインタ currentTime は継続的に前進します。 2 ミリ秒後、タイムグリッド 2 に到達すると、タイムグリッド 2 に対応する TimeTaskList 内のタスクをそれに応じて期限切れにする必要があります。

この時点で 8 ミリ秒のタイマーを持つ別のタスクが挿入されると、そのタスクはタイムスロット 10 に格納され、さらに 8 ミリ秒後に currentTime はタイムスロット 10 を指すようになります。同時に 19 ミリ秒のタイマーを持つタスクが挿入された場合はどうなりますか?新しい TimerTaskEntry は元の TimerTaskList を再利用するため、すでに期限切れになっている時間スロット 1 に挿入されます。つまり、タイムホイール全体の全体的な範囲は変化しません。ポインター currentTime が進み続けると、現在のタイムホイールが処理できる時間間隔も継続的に後方に移動し、全体的な時間範囲は currentTime と currentTime+interval の間になります。

350 ミリ秒にスケジュールされたタスクがある場合はどうすればよいでしょうか? wheelSize を直接拡張する必要がありますか? Kafka には、数万ミリ秒、あるいは数十万ミリ秒のスケジュールされたタスクが多数あります。 wheelSizeの拡張に制限はありません。スケジュールされたすべてのタスクの有効期限の上限を 100 万ミリ秒などに設定しても、wheelSize が 100 万ミリ秒のこのタイムホイールは大量のメモリ領域を占有するだけでなく、効率も低下します。 Kafka では、この目的のために階層型タイムホイールの概念を導入しています。タスクの有効期限が現在のタイムホイールによって表される時間範囲を超えると、上位のタイムホイールに追加しようとします。

上の図を参照して、前のケースを再利用し、*** レイヤーのタイムホイール tickMs=1ms、wheelSize=20、interval=20ms にします。第 2 層タイムホイールの tickMs は、第 1 層タイムホイールの間隔であり、20ms です。タイムホイールの各レイヤーの wheelSize は 20 に固定されているため、タイムホイールの 2 番目のレイヤーの全体的な時間範囲間隔は 400 ミリ秒になります。同様に、この 400 ミリ秒は第 3 層の tickMs のサイズでもあり、第 3 層のタイムホイールの全体的な時間範囲は 8000 ミリ秒です。

前述の 350ms タイミングタスクの場合、第 1 層のタイムホイールが条件を満たせないことは明らかなので、第 2 層のタイムホイールにアップグレードし、最終的に第 2 層のタイムホイールのタイムグリッド 17 に対応する TimerTaskList に挿入します。この時点でタイミングが 450 ミリ秒の別のタスクがある場合、明らかに第 2 レベルのタイムホイールは条件を満たすことができないため、第 3 レベルのタイムホイールにアップグレードされ、最終的に第 3 レベルのタイムホイールのタイムグリッド 1 の TimerTaskList に挿入されます。有効期限が [400 ミリ秒、800 ミリ秒) の範囲にある複数のタスク (446 ミリ秒、455 ミリ秒、473 ミリ秒のスケジュールされたタスクなど) は、第 3 レベルのタイムホイールのタイムグリッド 1 に配置されることに注意してください。タイムグリッド 1 に対応する TimerTaskList のタイムアウト期間は 400 ミリ秒です。

時間が経過し、TimerTaskList の有効期限が切れると、当初 450 ミリ秒でスケジュールされていたタスクの残り時間が 50 ミリ秒になり、このタスクの有効期限切れ操作は実行できなくなります。ここではタイムホイールのダウングレード操作が行われ、残り時間が 50 ミリ秒のスケジュールされたタスクが階層型タイムホイールに再送信されます。この時点で、第 1 レベルのタイムホイールの全体的な時間範囲は十分ではありませんが、第 2 レベルのタイム範囲は十分であるため、タスクは有効期限が [40 ミリ秒、60 ミリ秒) の第 2 レベルのタイムホイールの時間グリッドに配置されます。さらに 40 ミリ秒後にタスクが再び「検出」されますが、まだ 10 ミリ秒残っているため、有効期限切れ操作をすぐに実行することはできません。したがって、タイムホイールの別のダウングレードがあります。このタスクは、有効期限が [10ms、11ms) でトップレベルのタイムホイールのタイムグリッドに追加されます。さらに 10 ミリ秒後、このタスクは実際に期限切れになり、対応する期限切れ操作が最終的に実行されます。

デザインは生命から生まれます。私たちが普段使っている時計は、3層構造の時計ホイールです。タイムホイールの最初のレイヤーは tickMs=1ms、wheelSize=60、interval=1min で、秒数です。 2 番目のレイヤー tickMs=1min、wheelSize=60、interval=1hour (分)。 3 番目のレイヤーは tickMs=1hour、wheelSize は 12、interval は 12hours で、これは時計です。

Kafka では、最上位レベルのタイムホイールのパラメーターは上記の場合と同じです: tickMs=1ms、wheelSize=20、interval=20ms。各レベルの wheelSize も 20 に固定されているため、各レベルの tickMs と interval もそれに応じて計算できます。 Kafka では、TimingWheel を実装する際にいくつかの細かい点があります。

TimingWheel が作成されると、現在のシステム時刻が最上位のタイミングホイールの開始時刻 (startMs) として使用されます。ここでの現在のシステム時間は、単に System.currentTimeMillis() を呼び出すのではなく、Time.SYSTEM.hiResClockMs を呼び出します。これは、currentTimeMillis() メソッドの時間精度がオペレーティングシステムの特定の実装に依存するためです。一部のオペレーティングシステムでは、ミリ秒レベルの精度を実現できません。 Time.SYSTEM.hiResClockMs は実際には System.nanoTime()/1_000_000 を使用して、精度をミリ秒に調整します。ミリ秒単位の精度を実現できる他の高度な操作もいくつかありますが、お勧めしません。 System.nanoTime()/1_000_000 が最も効果的な方法です。（これについて何かアイデアがあれば、コメント欄で議論してください。）
TimingWheel 内の各双方向循環リンクリスト TimerTaskList には、センチネルノードが存在します。センチネルノードを導入すると境界条件を簡素化できます。センチネルノードはダミーノードとも呼ばれます。追加のリンクリストノードです。最初のノードとして、このノードはその値の範囲に何も格納せず、操作の便宜のためにのみ導入されます。リンクリストにセンチネルノードがある場合、線形リストの最初の要素はリンクリストの 2 番目のノードである必要があります。
第 1 レベルのタイムホイールを除き、他の高レベルのタイムホイールの開始時間 (startMs) は、このレベルのタイムホイールが作成されたときに、最初の前のホイールの currentTime に設定されます。各レイヤーの currentTime は tickMs の整数倍である必要があります。満たされない場合、currentTime は、タイムホイールのタイムグリッドの有効期限の範囲に対応するように tickMs の整数倍にトリミングされます。トリミング方法は次のとおりです: currentTime = startMs - (startMs % tickMs)。 currentTime は時間の経過とともに増加しますが、それが tickMs の整数倍であるという確立された事実は変わりません。ある瞬間の時刻が timeMs である場合、この時点でのタイムホイールの currentTime は timeMs - (timeMs % tickMs) になります。時間が進むたびに、各レベルのタイムホイールの currentTime がこの式に従って進みます。
Kafka のタイマーは、 TimingWheel の最上位レベルのタイムホイールの参照のみを保持すればよく、他の高レベルのタイムホイールを直接保持する必要はありません。ただし、タイムホイールの各レベルには、上位レベルのアプリケーションを指す参照 (overflowWheel) があります。このレベルを呼び出すことにより、タイマーはタイムホイールの各レベルの参照を間接的に保持できます。

これでタイムホイールの詳細説明は終了です。各コンポーネントにおけるタイムホイールの実装は同様です。読者は、この記事で説明されているシナリオ「時間が経つにつれて」または「時間が経つにつれて」に興味があるでしょうか。では、カフカでは時間はどのように進むのでしょうか。これは、JDK で scheduleAtFixedRate を使用してタイミングホイールを 1 秒ごとに進めるのと似ていますか?明らかに、これは合理的ではなく、TimingWheel はその意味のほとんどを失っています。

Kafka のタイマーは、JDK の DelayQueue を使用してタイムホイールを進めます。具体的なアプローチとしては、使用される各 TimerTaskList が DelayQueue に追加されます。「使用された各 TimerTaskList」は、非センチネルノードのスケジュールされたタスク項目 TimerTaskEntry を含む TimerTaskList を具体的に参照します。 DelayQueue は、対応する有効期限に従って TimerTaskList をソートし、有効期限が最も短い TimerTaskList を DelayQueue の先頭に配置します。 Kafka には、DelayQueue 内の期限切れのタスクのリストを取得するためのスレッドがあります。興味深いことに、このスレッドの名前は「ExpiredOperationReaper」であり、これは「期限切れの操作ハーベスター」と直訳でき、「SkimpyOffsetMap」に似ています。「ハーベスター」スレッドは、DelayQueue 内のタイムアウトしたタスクリスト TimerTaskList を取得すると、TimerTaskList の有効期限に応じてタイマーホイールの時間を進めるか、取得した TimerTaskList に対して対応する操作を実行できます。反対の TimerTaskEntry は、必要に応じて有効期限切れ操作を実行するか、必要に応じてタイマーホイールをダウングレードします。

読者はこれを読んだ後、非常に混乱するかもしれません。記事の冒頭で明確に指摘したように、DelayQueue は、Kafka などの高パフォーマンス要件を伴うスケジュールされたタスクには適していません。ここで DelayQueue が導入されるのはなぜでしょうか?スケジュールされたタスク項目 TimerTaskEntry の挿入および削除操作の場合、TimingWheel の時間計算量は O(1) であり、パフォーマンスは DelayQueue よりもはるかに高いことに注意してください。 TimerTaskEntry を DelayQueue に直接挿入すると、パフォーマンスをサポートするのが明らかに難しくなります。特定のルールに従って複数の TimerTaskEntry を TimerTaskList グループに分割し、その TimerTaskList を DelayQueue に挿入したとしても、この TimerTaskList にもう 1 つの TimerTaskEntry を追加したい場合はどうすればよいでしょうか。 DelayQueue の場合、この種の操作は明らかにその能力を超えてしまいます。

この分析から、Kafka の TimingWheel は TimerTaskEntry 操作の挿入と削除に特に使用され、DelayQueue は時間を進めるタスクに特に責任があることがわかります。もう一度、DelayQueue 内の最初のタイムアウトタスクリストの有効期限が 200 ミリ秒で、2 番目のタイムアウトタスクの有効期限が 840 ミリ秒であると想像してください。ここで、DelayQueueの先頭を取得するには、O(1)の時間計算量のみが必要です。時間指定の進捗が毎秒採用される場合、最初のタイムアウトタスクリストが取得されたときに実行される 200 の進捗のうち 199 は「空の進捗」であり、2 番目のタイムアウトタスクが取得されたときに 639 の「空の進捗」を実行する必要があります。これにより、マシンのパフォーマンスリソースが無駄に浪費されます。ここで、DelayQueue は少量のスペースを時間と交換するのを補助するために使用され、それによって「正確な前進」を実現します。カフカにおけるタイマーは「人を知り、うまく活用する」ことだと言えます。 TimingWheel は、最も得意とするタスクの追加と削除の操作を実行するために使用され、DelayQueue は、最も得意とする時間の進行の作業を実行するために使用され、互いに補完し合います。

<<: 注意深い！クラウドコンピューティングの8つの隠れたコストに注意