SignalsyMQ-PHP + Redis（および小さなZend Framework）のシンプルで高速なメッセージキュー

読者への挨拶。 現在、現在支配的なMVCモデルとは対照的に、 信号アーキテクチャ （信号/スロット）に基づいて独自のフレームワークを開発しています。 彼が私たちのスタートアップで戦闘の慣らしを受けている間に、この間、ライブラリが不足していることに気付きました-実際のタスクと将来のビジョンは、異なる機能が必要であるが、1つのテーマで統一されていることを示しています-リアルタイムで多くのクライアントへの情報の処理と配信（はい、コメットに似たもので、彼もそこにいます）。 したがって、次のプロジェクトの主要なコンポーネントとなる、高速で柔軟かつスケーラブルなメインコンポーネントであるメッセージキューの実装を試みることが決定されました。

どうしたの？ SignalsyMQ Alphaは、PHP / Redis / Zend Frameworkベースのメッセージキューです。

メッセージキューについてはすでに書きました（ レビューの最初の記事 、 2番目 ）が、すべてのシステムを批判的にレビューし、それらのいくつかを試してみると、メインアプリケーションのプラットフォームにできるだけ近い、このニッチが小さく、柔軟で構成可能であることに気付きました（この場合、PHP）、オープンで無料のままです。 標準メッセージ（JSON形式で、アプリケーション内の単なる連想配列）を処理し、メッセージ配信の柔軟な構成、キューの永続的な保存、およびそれらとの非常に高速な処理を提供するシンプルなサーバーがありません。 同時に、シンプルなプロトコルと、一般的に最大限のシンプルさが必要です。

これに最も近いのは、Rubyベースの製品であるStarling MQとMemcacheQです。 ところで、両方のシステムはmemcacheプロトコルで動作します。これは、異種環境に埋め込む機能にプラスの影響を与えます。 しかし、memcacheqはその開発ポリシーに恥ずかしかったのですが、Twitterの使用経験はありますが、それはかなり具体的な製品であり、絶対に必要です。別のプラットフォーム。 したがって、最良の伝統では、メッセージキューの独自の実装を記述することが決定されました。

Redisプロジェクトは、構造化データを保存および処理するための開発機能を備えた非常に高速かつ柔軟なNoSQLシステムの出発点として取り上げられました（Habréでこれについて書く時間はありませんでした）。 2番目の「クジラ」はRediskaでした-便利な構文と多くの興味深い機能（たとえば、複数のサーバー、組み込みのシリアル化、サーバー間でのキーの配布など）をサポートするRedisと連携するためのPHPライブラリですが、ライブラリは活発に開発されています。したがって、機能は常に追加されます）。 3番目の「クジラ」はZend Frameworkで 、現在ほとんどのプロジェクトで使用していますが、ここではサポートの役割しか果たしていません（いくつかのユーティリティクラスが使用されています）。

ZFには、メッセージキューの独自の実装-Zend_Queueが既にあることがすぐにわかります。これは、通常のPHP配列から始まり、MemcacheQ、データベース、Zend Platformで終わる、メッセージを保存するためのさまざまなバックエンドをサポートします。 ただし、実際には、MemcacheQで最も有望なインターフェイスを試すことができませんでした。メッセージフローが密集している場合にデータベースを操作することは、単に不可能です。スクリプト全体が失敗するまで、作業は非常に遅くなります。 Zend_Queueインターフェース自体は非常に抽象的です。私の場合は、いくつかの高レベルの関数が必要だったため、既存のコードを大幅に拡張する必要があります。 ちなみに、RediskaライブラリにはRedis上でキューを操作するための特別なアダプターがありますが、1週間苦労し、テストケースが動作しなかった後、私は最終的に決定しました-それが私のシステムを書いています！

まず、一般的なアーキテクチャと機能を決定しましょう。 基本的な概念は、メッセージ、キュー（チャネル）、サーバー、およびストレージです。

• メッセージは特定のデータ構造であり、この例では、保存する情報を含む連想配列です。 メッセージは、いくつかのサービスフィールドと本文で構成されます。 メッセージの本文には、受信者に送信されるデータ自体が保存されます。最も重要なことは、変更されていないことです。 サービスフィールドは処理中に変更される場合があります。 これまでのところ、私に必要なフィールドは、メッセージのタイプ（任意の行）、メッセージのソース（また行）、メッセージを配信するチャネル、およびメッセージを作成するためのタイムスタンプです。 メッセージ本文もフィールドの1つですが、空の場合もあります。特定の受信者に対してのみ意味があり、システム全体の責任は単に保存して配信することです。
• キュー （チャネル）は一意のアドレスであり、メッセージが配信されるキューの名前です。 私は最もシンプルで最も理解しやすい形式のアドレス指定-URLのようなものを取りました。 たとえば、/ users / aleksは、ユーザーaleksのメッセージキューです。 内部でいくつかの追加処理が発生しますが、これは他のスクリプトがキューで動作する方法には影響しません。 また、一度に複数のキューにメッセージを送信することもできます-このためには、メタ文字「*」（任意の数の文字）、「？」 （任意の単一文字）および '[a..z]'-配信されるメッセージの名前に含まれる必要がある任意の文字セット。 これは私の開発が同じZend_Queueと比較して好意的です-たとえば、名前がaleksであるすべてのユーザーにメッセージを簡単に送信できます（ユーザー名でキューを呼び出す場合）。 キューは、長さ（そこに格納できるメッセージの最大数）によっても特徴付けられます。 メッセージがメモリ（さらにはシリアル化された形式）に保存されることを考えると、メモリの消費を考慮する必要があるため、プロジェクトの詳細に基づいてキューの長さを選択してください。 キューがオーバーフローした場合、最も古いメッセージが削除されます。
• 実際、 キューサーバーはライブラリスクリプトそのものです。 現在の実装はRedisサーバーで直接動作します。次のステップは専用サーバーの作成です。 つまり、クライアントアプリケーションはソケットを介してサーバーに接続し、メタ情報付きのメッセージを送信します（少なくとも、メッセージを送信するチャネルを示します）。残りはサーバーのビジネス（またはデーモン）です。 機能を拡張するために、ソケット、XML / JSON-RPC、HTTP-REST、SOAPなどを介して、さまざまなデータ形式（シリアル化）といくつかの作業プロトコルの両方をサポートする予定です。
• ストレージ -現在のバージョンでは1つまたは複数のRedisサーバーを実行していますが、将来的には、Zend_Queueと同様に、メッセージを保存するためのいくつかのオプションを実装する可能性があります。

アーキテクチャとRediska_Queueとの違いについてのもう1つの発言は、ソースから理解しているように、速度のために貴重な配布プロパティを配置することです。 たとえば、格納されているすべてのキューの配列（メッセージではなく、キューの名前のみ）を返すgetQueuesコマンドは、キューリストのローカルコピーを処理します。 したがって、最初のキューが機能している間に別のクライアントが新しいキューを作成した場合、彼はそれを知りません。 Webアプリケーションの場合、ページが形成されている間にリクエストが処理されるとき、これは重要ではないかもしれませんが、デーモンとして動作し、多くのクライアントからの多くのキューとコマンドを処理できるメッセージサーバーが必要です。 これには追加のリソースが必要ですが、Redisのおかげですが、すべての操作には複雑さの非常に良い指標があります。

したがって、一般的なAPIは、Zend_Queueインターフェースに可能な限り近いいくつかのメソッドで構成されていますが、いくつかの拡張機能があります。これについては以下で説明します。

• getInstance（$ config） -ライブラリとアダプターの1つのインスタンスを使用してRedisにアクセスするため、すべての作業は、既に初期化されたインスタンスの作成または使用を決定する静的メソッドを介して行われます。 構成では、サーバーパラメーター（アドレス、ポート）が設定され、複数のサーバーの場合はシリアル化機能とキー配布メカニズムが設定されます。
  

    private $ _options = Array（
     'maxQueues' => 100000、//キューの最大数
     'maxMessagesPerQueues' => 1000、// 1つのキュー内のメッセージの最大数
     'expireQueues' => 15552000、//キューの最大有効期間は180日間
     'expireMessages' => 604800、// 1つのメッセージの最大有効期間
     'maxMessageLength' => 32768、//メッセージの最大サイズ（シリアル化後）
     'log' => false、//指定されている場合、メッセージを記録するZend_Logオブジェクト、
     'redis' =>配列（
  	 'namespace' => 'smq_'、//キーを減らすために選択されたキーの名前空間
  	 'サーバー' =>配列（
   	   array（ 'host' => 'localhost'、//ホスト
  	           'port' => 6379、//ポート
  	           'weight' => 1、//キー配布のためのサーバーの重み
  	           'password' => '' //パスワード
  	  ）
     ）、
    'keyDistributor' => 'crc32'、//キー配布タイプ
    'serializer' =>配列（ 'Signalsy_MQ'、 '_ serialize'）、
    'unserializer' =>配列（ 'Signalsy_MQ'、 '_ unserialize'））
   ）; 

  
  ロガー（Zend_Logを使用）を設定して、すべての重要なイベント、エラー、さらにはメッセージ自体を記録することもできます（デバッグではなく、プロダクションでRedisにデータ損失または複製を防ぐために組み込まれたappendonly.logメカニズムを設定するだけで十分です）バックアップサーバーへ）。
• _serialize / _unserialize-格納されたメッセージをリポジトリに追加し、そこから取得するときに使用される組み込みのシリアル化関数。 Redisは値として数値と文字列のみを保存できるため、保存する前にデータをシリアル化する必要があります。 これはサーバー自体によって透過的に行われます。これらのアクションのメソッドを指定するだけでよく、そうでない場合はデフォルトでPHPシリアル化が使用されます。 将来的にはクライアントとリポジトリの間の中間層の数を最小限に抑えたいため、AJAXアプリケーションにはより便利で「ネイティブ」なJSONを選択しました。 次のバージョンでは、Zendインキュベーターの新しいZend_Serializerコンポーネントが使用され、他のアルゴリズムを使用できるようになります。 さらに、PHPに保存されたデータは、JSONパーサーがある任意の言語の他のプログラムで読み取ることができます。これは、分散システムにとって非常に重要です。
• createQueueは、名前が渡されるキューを作成するメインメソッドです。 最初に、名前の正確さがチェックされ（さまざまな特殊文字が削除されるなど）、システム内のそのようなキューの存在もチェックされます。 作成後、最初のサービスメッセージがキューに配置され、タイムスタンプが示されます。 これは、Redisが最初の要素を追加するときに空のリスト（リストデータ構造）を自動的に作成するため、空の文字列などの公式なものを指定しないように、特別なユーティリティメッセージタイプでキューを初期化することが決定されたためです。
• isExists-指定されたキューがサーバー上に存在するかどうかを確認します。 Zend_Queue用のRediskaアダプターの実装とは対照的に、データベースメモリ内のそのようなオブジェクトの存在の確認やデータ型の一致など、より深刻なチェックを実行します。
• count-キュー内のメッセージの数を取得できます。
• deleteQueue-サーバーストレージからキューを削除します。キュー内のすべてのメッセージも削除されます。 おそらく、クラッシュの場合にデータが関連するように、データベースの状態の強制記録をそのような変更で初期化する必要があります。
• sendは、指定されたチャネルにメッセージを送信するメインメソッドです。 必須パラメーター-メッセージに加えて、キューシステムの正しい動作を保証する追加のチェックを指定できますが、作業が大幅に遅くなる可能性があります。 skipCheckQueueLengthは、構成で設定された制限を超えた場合にキューの長さとその切り捨てをチェックする役割を果たします。 切り捨てられると、最も古いメッセージが削除されます。 このパラメーターが整数の場合、この要求でこのようなチェックが実行される可能性を設定しますが、これはサーバーへの1つまたは2つの要求を必要とするため、速度と信頼性の間の一種の妥協です。 skipCheckQueueExistsパラメーターを使用すると、キューの存在のチェックを無効にできます。 前述のように、メッセージは標準的な配列であり、シリアル化後にキューに格納されます。 必須チャネルフィールドで、メッセージを配置するキュー（またはマスクを使用した複数）を指定します。 マスクが指定されている場合、指定されたパラメーターに一致するすべてのキュー名をこのマスクから抽出し、メッセージ内のチャネルフィールドを調整した後、ループ内の各キュー名にメッセージを追加します。 現時点では、サービスヘッダーの変更や分析は行われません。メッセージの本文自体も、文字列または正しいシリアル化を可能にするPHPの任意のデータ型のいずれかです。 繰り返しますが、他の実装との主な違いは、1回の呼び出しで複数の（またはすべての）キューにメッセージを送信できることです。 ロギングオプションが設定されている場合、メッセージはZend_Logを使用してログに書き込まれます
• receive -2番目に重要な操作-キューから1つ以上のメッセージを受信します。 ここではマスクはサポートされなくなり、メッセージは特定の指定されたキューからのみ受信できます。 また、サーバー上のキューの存在を確認する必要があります（特に最大のパフォーマンスを得るには、特にRedisサーバーで作業する人がいない場合、これらのすべての確認は作業の初期化時に1回のみ実行できます）。 受信するために、受信できるメッセージの最大数を示しますが、実際には、たとえばキューに同じ数のメッセージがある場合、受信できるメッセージの数は少なくなります。 別の興味深いオプション（最適化に関して機能が完全になるまで非表示にしません）は、指定した日付以降に送信されたメッセージのみを受信することです。 たとえば、過去1時間に送信されたメッセージを100個まで受信したいとします。 最初に、指定された数のメッセージがサンプリングされ、次に作成時間の基準を満たさないメッセージをチェックして破棄します。 キューからのメッセージは読み取り専用であり、読み取り直後に削除されないことに注意してください-このためには、特別なコマンドを使用する必要があります。 キュー自体は永続的であり、そのステータスはクライアントによって手動で（またはキューの長さをチェックした結果として）変更されるだけです。 将来的には、期限切れメカニズムが実装されます。Redisで最初にサポートされますが、クライアントパーツではまだ実装されていません（個々のリストアイテム用）。
• deleteMessage-キューからメッセージを削除します。
• getQueues-サーバーで使用可能なすべてのキューのリストを表示できます。
• getQueuesByPattern-指定されたパターンに一致するキューのリストを表示します（メタキャラクター「*」、「？」、および文字グループ[a..z]を使用します）

正直なところ、 プロジェクトのソースコードは、アクセス可能ですが、実際の作業の準備ができていません。むしろ、単なるプロトタイプと最初のアルファバージョンです。 まだ最適化されていない場所、Zend_Queueのピースなど、さまざまなドキュメントがまだありますが、それらのインターフェイスには、不完全に考え出された作業とデータ構造のアルゴリズムがあります（たとえば、deleteMessageには、チャネルサービスフィールドには2つのパラメーター、キュー名とメッセージが必要です）メッセージには既に彼のキューの名前などがあります）。 それにもかかわらず、現在のコードはすでに結果を示しており、動作しており、実験に適しています。

現在パフォーマンスについて話すことも本当に不可能ですが、Amazon EC2スモールインスタンス（32ビット、Debian、PHP 5.2.11、Redis 1.1.9）を使用していくつかのテストを実施しました。ランダムキュー（1000の多数のメッセージを選択的に記録するモードのシングルスレッドサーバーメッセージおよび各バッチ100）は、1秒あたり1500-3000メッセージ内の速度を示します（記録のため、他のシステムはまだサーバー上で回転しているため、変動は負荷に大きく依存します。また、このモードではRedis自体の保存間隔が設定されるため、非常に敏感です）

各クライアントが少数のメッセージと1つのキューで動作するが、多数の並列クライアントがある場合、原則として、 この構成のテストによると 、引き出し可能な最大値（1秒あたり5から7千メッセージ、テストとサーバーが同じマシン上にある場合）。 マルチスレッドを使用して、Redisを別のサーバーに移動し、サーバーを可能な限り他のタスクからアンロードすることで、より大きな結果を得ることができます。さらに、パフォーマンスはサーバーのCPUパラメーターに依存します-大きいインスタンス（2コアと7 GB RAM）はすでに5-6より速い速度（もちろん、これは単なる平均的な指標であり、現実とはかけ離れています）。

これらは実験です。 メッセージキューのトピックは興味深いものです。たとえば、仕事でZFを積極的に使用している開発者であっても、Zend_Queueコンポーネントとそのアプリケーションについてほとんど誰も知らないなど、さらに記事を書く必要があります。 さて、サーバーには多くの改善と改善があり、それをネットワークデーモンに変え、さまざまなプロトコル（おそらくMQの標準であるStomp、AMQP）で動作する特別なサーバーを構築し、バックグラウンド保存中の輻輳のレプリケーションとアカウンティングを設定し、いくつかの並列サーバーなど あなたはそれについて読んでみたいですか？