Redisキュー優先度コード例の実装方法の詳細な説明

redis を使用してメッセージ キューを作成する方法

まず、redis はキャッシュに使用するように設計されていますが、その独自の特性のいくつかにより、メッセージ キューにも使用できます。使用できるブロック API がいくつかあり、メッセージ キューを実行できるのはこれらのブロック API です。

「データベースがすべての問題を解決する」という考えの下で、メッセージキューを使用せずにニーズを完了できると想像してください。すべてのタスクをデータベースに保存し、継続的なポーリングを通じて処理します。このアプローチでタスクを完了できますが、非常に大雑把です。ただし、データベース インターフェイスがブロック メソッドを提供している場合、データベースをメッセージ キューとして使用することもできますが、現在のデータベースにはそのようなインターフェイスがありません。

さらに、FIFO などのメッセージ キューの他の機能も簡単に実装できます。必要なのは、先頭からデータをフェッチし、末尾からデータを詰め込むだけです。

Redis は、そのリスト オブジェクト blpop brpop インターフェイスと Pub/Sub (パブリッシュ/サブスクライブ) のいくつかのインターフェイスのおかげでメッセージ キューを実行できます。これらはすべてブロッキング バージョンであるため、メッセージ キューとして使用できます。

rabbitmq の優先事項アプローチ

現在、成熟したメッセージキュー製品が多数あり、有名なものは Rabbitmq です。使い方は比較的簡単で、機能も比較的豊富で、一般的な状況では十分です。ただし、非常に厄介な点は、優先順位をサポートしていないことです。

たとえば、電子メールを送信するタスクにおいて、一部の特権ユーザーは、自分の電子メールがよりタイムリーに送信されること、または少なくとも通常のユーザーより優先されることを望んでいます。デフォルトでは、rabbitmq はそれを処理できません。rabbitmq にスローされるタスクは FIFO 先入れ先出しです。ただし、これらの優先事項をサポートするためにいくつかの回避策を使用できます。複数のキューを作成し、rabbitmq コンシューマに対応するルーティング ルールを設定します。

例えば、デフォルトでこのようなキューがあり、[タスク1、タスク2、タスク3]をシミュレートするリストを使用し、コンシューマはFIFOの原則に従って順番にタスクを1つずつ取り出して処理します。高優先度のタスクが入った場合、最後にのみ処理できます [task1、task2、task3、higitask1]。ただし、高優先度のキューと通常の優先度のキューの 2 つのキューが使用される場合。 通常の優先度 [task1、task2、task3]、高優先度 [hightask1] 次に、コンシューマが任意のキューからデータをランダムにフェッチできるようにコンシューマのルーティングを設定します。

そして、アイドル状態のときに優先度の低いキューからのデータを処理しない、優先度の高いキューを特別に処理するコンシューマを定義できます。これは銀行のVIPカウンターに似ていますが、VIPが来ると、一般会員の前にある番号取り機からチケットを取り出すことはありません。 VIP チャンネルに直接アクセスすることで、より速くアクセスできるようになります。

rabbitmq を使用して優先メッセージ キューをサポートする場合は、前述した同じ銀行の VIP メンバーと同様に、異なるチャネルを経由します。ただし、この方法は相対的な優先度を使用するだけであり、絶対的な優先度制御を実現することはできません。たとえば、ある優先度の高いタスクが他の通常のタスクよりも先に処理されることを期待します。この場合、上記の解決策は機能しません。 。 Rabbitmq のコンシューマーは、キューが空いているときに対象のキューからキュー内の最初のデータを「ランダムに」選択することしか知らないため、どのキューを最初に取得するかを制御することはできません。あるいは、よりきめ細かい優先制御。 または、システムに 10 を超える優先順位が設定されています。このように、rabbitmq を使用して実現することも困難です。

しかし、redis をキューとして使用すると、上記の要件を達成できます。

メッセージキューが必要な理由

システムへのメッセージキューメカニズムの導入は、システムにとって非常に大きな改善です。たとえば、Web システムでは、ユーザーが特定の操作を実行した後、ユーザーのメールボックスに電子メール通知を送信する必要があります。同期方式を使用すると、メールが送信され、完了後にユーザーにフィードバックが送信されるまでユーザーを待機させることができますが、これにより、ネットワークの不確実性によりユーザーが長時間待機することになり、ユーザー エクスペリエンスに影響を与える可能性があります。 。

一部のシナリオでは、同期メソッドを使用して完了を待つことが不可能であり、それらの操作はバックグラウンドで長時間を必要とします。たとえば、極端な例では、オンライン コンパイル システム タスクの場合、バックグラウンド コンパイルが完了するまでに 30 分かかります。このシナリオの設計では、同期的に待機してからフィードバックを提供することはできません。まずユーザーにフィードバックしてから非同期処理が完了し、その後、処理が完了するのを待ってからユーザーにフィードバックする必要があります。状況。

さらに、メッセージ キューは、システムの処理能力が制限されている状況に適しています。キュー メカニズムは、最初にタスクを一時的に保存するために使用され、その後、システムがキューに入れられたタスクを 1 つずつ順番に処理します。これにより、システムのスループットが不足している場合でも、同時実行性の高いタスクを安定して処理できます。

メッセージキューはキューイングメカニズムとして使用できます。システムがキューイングメカニズムを使用する必要がある限り、メッセージキューを使用できます。

Redis メッセージキューの優先順位の実装

Redis の基本的な基本のいくつかの説明

redis> blpop tasklist 0
"im task 01"

この例では、blpop コマンドを使用して、ブロック方式でタスクリスト リストから最初のデータをフェッチします。最後のパラメーターは待機タイムアウトです。 0 に設定すると、無期限に待機することを意味します。さらに、redis に保存されるデータは string 型のみであるため、タスクを転送するときは string のみを渡すことができます。必要なのは、単に責任のあるデータを json 形式の文字列にシリアル化し、それをコンシューマー側で変換することだけです。

ここでのサンプル言語は Python を使用し、redis にリンクされたライブラリは redis-py を使用します。ある程度のプログラミング スキルがある場合は、好みの言語に切り替えても問題ありません。

1. 単純な FIFO キュー

import redis, time
def handle(task):
    print task
    time.sleep(4)
 
def main():
    pool = redis.ConnectionPool(host='localhost', port=6379, db=0)
    r = redis.Redis(connection_pool=pool)
    while 1:
        result = r.brpop('tasklist', 0)
        handle(result[1])
 
if name == "main":
    main()

上記の例は、最も単純なコンシューマーでもあり、無限ループを通じて Redis キューからデータをフェッチします。キューにデータがない場合はタイムアウトせずにそこでブロックされ、データがあれば取​​り出して実行されます。

通常、それは取り出された複雑な文字列になります。フォーマットしてから処理関数に渡す必要がある場合がありますが、簡単にするために、この例では通常の文字列を使用します。また、この例の処理関数は処理を実行しませんが、時間のかかる操作をシミュレートするためにスリープが使用されます。

プロデューサーをシミュレートするために別の Redis クライアントを開きます。組み込みクライアントで十分です。タスクリストキューにさらにデータを入れます。

redis> lpush tasklist 'im task 01'
redis> lpush tasklist 'im task 02'
redis> lpush tasklist 'im task 03'
redis> lpush tasklist 'im task 04'
redis> lpush tasklist 'im task 05'

その後、コンシューマ側では、これらのシミュレートされたタスクが 1 つずつ消費されるのがわかります。

2. 単純な優先度キュー

優先度の高いタスクを優先度の低いタスクよりも先に処理することだけを必要とする単純な要件を想定します。この場合、他のタスクの順序は重要ではありません。優先度の高いタスクが見つかったときに、それをキューの後ろにプッシュするのではなく、キューの先頭にプッシュするだけで済みます。

私たちのキューは Redis リストを使用しているため、実装は簡単です。優先度が高い場合は rpush を使用し、優先度が低い場合は lpush を使用します。すると、常に高優先度が低優先度よりも先に実行されることがわかります。ただし、このソリューションの欠点は、優先度の高いタスクの実行順序が先入れ後出しになることです。

3. より完全なキュー

例 2 では、優先度の高いタスクがキューの先頭に配置され、優先度の低いタスクが最後に配置されます。これは、優先度の高いタスク間の順序を保証するものではありません。

すべてのタスクの優先度が高い場合、それらの実行順序が逆になるとします。これは明らかにキューの FIFO 原則に違反しています。

しかし、少し改善することで、キューを改善することができます。

rabbitmq の使用と同様に、高優先度キューと低優先度キューの 2 つのキューを設定します。優先度の高いタスクは優先度の高いキューに配置され、優先度の低いタスクは優先度の低いキューに配置されます。 redis と Rabbitmq の違いは、キューコンシューマーにどのキューから最初に読み取るかを要求できることです。

redis> lpush tasklist 'im task 01'
redis> lpush tasklist 'im task 02'
redis> rpush tasklist 'im high task 01'
redis> rpush tasklist 'im high task 01'
redis> lpush tasklist 'im task 03'
redis> rpush tasklist 'im high task 03'

最初のキューが 2 番目のキューからフェッチされない場合、上記のコードは 2 つのキュー「high_task_queue」と「low_task_queue」からブロック的にデータをフェッチします。

したがって、目標を達成するには、キュー コンシューマにそのような改善を加えるだけで済みます。

def main():
    pool = redis.ConnectionPool(host='localhost', port=6379, db=0)
    r = redis.Redis(connection_pool=pool)
    while 1:
        result = r.brpop(['high_task_queue', 'low_task_queue'], 0)
        handle(result[1])

上記のテストを通じて、高優先度が最初に実行され、高優先度の間でも FIFO 原理が保証されていることがわかります。

このソリューションを使用すると、高、中、低、またはそれ以上の 3 つのレベルなど、さまざまな段階で優先キューをサポートできます。

4. 多くの優先順位がある状況

このような需要があり、優先順位が高、中、低、または 0 ～ 10 の単純な固定レベルではないとします。ただし、0 ～ 99999 など、非常に多くのレベルがあります。その場合、3 番目のオプションは適切ではなくなります。

redis にはソートセットのようなソート可能な

がありますが、インターフェイスのブロックバージョンがないのは残念に思えます。したがって、他の方法で目的を達成するには、リスト型を使用するしかありません。 簡単な方法は、キューを 1 つだけ設定し、優先度に従ってソートされるようにすることです。次に、

メソッドを使用してタスクの適切な場所を見つけ、lset コマンドを使用してタスクを対応する場所に挿入します。 たとえば、キュ​​ーには書き込み優先度[1、3、6、8、9、14]のタスクが含まれており、優先度7のタスクが来ると、独自のバイナリアルゴリズムを使用してキューからデータを1つずつ取得します。対象データと比較し、対応する位置を計算し、指定した位置に挿入します。

二分探索は比較的高速で、redis自体もメモリ上にあるため、理論上は速度が保証されます。ただし、データ量が非常に大きい場合は、何らかの方法で調整することもできます。

3 番目のオプションを思い出してください。3 番目のオプションを組み合わせると、オーバーヘッドが大幅に削減されます。たとえば、データ量が 100,000 のキューの場合、優先順位も 0 ～ 100,000 のランダムな範囲になります。 10 または 100 の異なるキューを設定できます。0 ～ 10,000 の優先タスクはキュー 1 に配置され、10,000 ～ 20,000 のタスクはキュー 2 に配置されます。このように、キューが異なるレベルに分割された後は、単一のキューのデータが大幅に削減されるため、二分探索マッチングの効率が高くなります。ただし、データが占有するリソースは基本的に変わりません。10 万個のデータが同じ量のメモリを占有する必要があります。システム内にさらに多くのキューが存在するだけです。

以上がRedisキュー優先度コード例の実装方法の詳細な説明の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。