転送: RabbitMQ と PHP (1)

RabbitMQ と PHP (1)

このプロジェクトは、ユーザー メッセージ通知を処理するキューとして RabbitMQ を使用します。メッセージはフロントエンドの PHP コードによって生成され、メッセージの処理には Python が使用されます。これにより、コードの一貫性の問題が発生します。メッセージ定義を調整する場合は、PHP と Python の両方を変更する必要があります。私は過去 2 日間、PHP を使用してメッセージを生成および処理 (消費) する方法を勉強してきました。情報を確認すると、PHP のメッセージ キューの処理に関する情報が非常に少ないことがわかりました。初心者が混乱しやすい点について説明する必要があります。

1. RabbitMQ の原理と動作例、2. 具体的なサービスのインストールとデーモンモードとして PHP を使用してメッセージを処理する方法の 2 つのパートに分かれる予定です。

RabbitMQ は、アーラン言語で開発された人気のオープンソース メッセージ キュー システムであり、AMPQ (Advanced Message Queuing Protocol) を完全に実装しています。 Web サイトは次のとおりです: http://www.rabbitmq.com/ チュートリアルとサンプル コード (Python および Java) があります。

AMPQ プロトコルは、さまざまなメッセージ キューの要件を満たすために概念的に複雑です。まず、rabbitMQ はデフォルトで何も設定せずに起動しますが、動作するにはクライアントの接続、スイッチの設定などが必要です。これらの基本的な考え方が明確になっていないと、その後のプログラミング設計で問題が発生しやすくなります。

1、仮想ホスト: 仮想ホスト

RabbitMQ エンティティは複数の仮想ホストを持つことができ、ユーザーと権限の設定は仮想ホストに依存します。一般的な PHP アプリケーションの場合、ユーザー権限の設定は必要ありません。デフォルトで存在する「/」を使用するだけで済みます。簡単な設定例:

$conn_args = array(
'host' => '127.0.0.1',
'port' => '5672',
'login' => 'guest',
' パスワード' => 'guest',
'vhost'=>'/'
);

2、接続とチャネル: 接続とチャネル

接続は、クライアントとサーバー間の物理的な接続を指します。 1 つの接続上に複数のチャネルを確立できます。これは論理接続として理解できます。一般的なアプリケーションでは、1 つのチャネルで十分であり、さらにチャネルを作成する必要はありません。サンプルコード:

//接続とチャネルを作成します
$conn = new AMQPConnection($conn_args);
if (!$conn->connect()) {
die("ブローカーに接続できません!n");
}
$channel = new AMQPChannel($conn);

3、交換とルーティングキー: スイッチとルーティングキー

異なるタイプのメッセージを区別するために、スイッチとルーティングの 2 つの概念が設定されます。たとえば、タイプ A のメッセージを「C1」という名前のスイッチに送信し、タイプ B のメッセージを「C2」という名前のスイッチに送信します。クライアントが C1 に接続してキュー メッセージを処理する場合、タイプ A のメッセージのみを取得します。さらに、タイプ A のメッセージが多数ある場合は、区別をさらに改良する必要があります。たとえば、クライアントが K 人のユーザーに対するタイプ A のメッセージのみを処理する場合、この目的にルーティングキーが使用されます。

$e_name = 'e_linvo'; //スイッチ名
$k_route = array(0=> 'key_1', 1=> 'key_2') //ルーティングキー
//スイッチを作成します
$ex = new AMQPExchange($channel);
$ex->setName($e_name);
$ex->setType(AMQP_EX_TYPE_DIRECT); //ダイレクトタイプ
$ex->setFlags(AMQP_DURABLE); //Persistence
"交換ステータス:".$ex->declare()."n";
for($i=0; $i echo "メッセージの送信:".$ex-> ; public($message . date('H:i:s'), $k_route[i%2])."n";
}
上記のコードからわかるように、メッセージを送信するときは、 「スイッチ」で十分です。スイッチの背後に対応する処理キューがあるかどうかについては、送信側は気にする必要はありません。 routingkey は空の文字列にすることができます。この例では、以下のルーティング キーの役割を理解しやすくするために、2 つのキーを使用してメッセージを交互に送信します。

スイッチには 2 つの重要な概念があります:

A、タイプ。 3 つのタイプがあります。ファンアウト タイプが最も単純で、このモデルはルーティング キーを無視します。ダイレクト タイプは、特定のルーティング キーを使用して最もよく使用されます。このモデルでは、メッセージの受信時に「key_1」をバインドすると、最後のメッセージはトピックです。このモードはダイレクトと似ていますが、「key_*」などのワイルドカード マッチングをサポートします。 key_1 と key_2 を受け入れます。トピックは見栄えがしますが、不正確になる可能性があるため、ダイレクトを使用することをお勧めします。

B、粘り強さ。永続化のために指定されたスイッチは再起動時にのみ再構築できます。それ以外の場合は、クライアントが世代を再宣言する必要があります。

特に明確な概念が必要です。スイッチの永続性はメッセージの永続性と等しくありません。永続化できるのは永続化キュー内のメッセージだけです。キューがない場合、メッセージには保存する場所がありません。PHP では、メッセージ自体にも永続化スイッチに配信されるメッセージが永続化されます。永続的なメッセージ。アドホックは必要ありません。

4、キュー: キュー

たくさん話した後、キューについて話しました。実際、キューは受信者 (消費者) 専用であり、要求に基づいて受信者によって作成されます。スイッチは、キューが作成された場合にのみ、新しく受信したメッセージをキューに送信します。キューが作成される前に、スイッチはメッセージを書き込みません。つまり、キューが確立される前に送信されたメッセージはすべて破棄されます。下の図は、Queue が ReceiveMessage の一部である公式の図よりも明確です。

キューの作成とメッセージの受信の例を見てみましょう:

$e_name = 'e_linvo'; //スイッチ名
$q_name = 'q_linvo' //キュー名
$k_route = ''; //ルートキー

//接続とチャネルを作成します
$conn = new AMQPConnection($conn_args);
if (!$conn->connect()) {
die("ブローカーに接続できません!n") ;
}
$channel = new AMQPChannel($conn);

//スイッチを作成します
$ex = new AMQPExchange($channel);
$ex->setName($e_name);
$ex-> setType(AMQP_EX_TYPE_DIRECT) ; //直接型
$ex->setFlags(AMQP_DURABLE); //Persistence
echo "Exchange Status:".$ex->declare()."n";

//キューを作成
$q = new AMQPQueue($channel);
$q->setName($q_name);
$q->setFlags(AMQP_DURABLE); //Persistence

//スイッチとキューをバインドし、ルーティングキー
echo 'Queue Bind: '.$q->bind($e_name, $k_route)."n";

//ブロッキングモードでメッセージを受信
echo "Message:n";
$q-> ;consume(' processMessage', AMQP_AUTOACK); //自動 ACK 応答

$conn->disconnect();

/**
*消費コールバック関数
*メッセージの処理
*/
function processMessage($envelope, $queue) {
var_dump( $envelope-> ;getRoutingKey);
$msg = $envelope->getBody();
echo $msg."n"; //メッセージの処理
}
上記の例からわかるように、スイッチは次のようになります。メッセージ送信者によって作成されますが、メッセージ利用者によっても作成できます。

キューを作成した後 (行:20)、キューを機能させるには、キューをスイッチにバインドする必要があります (行:25)。ルーティングキーもここで指定します。一部の情報では「bindingkey」と記載されていますが、これは実際には同じものです。2 つの名詞を使用すると混同しやすくなります。

メッセージを処理するには 2 つの方法があります:

A、1 回限り。 $q->get([...]) を使用すると、メッセージが取得されたかどうかに関係なく、すぐに返されます。通常、このメソッドは、ポーリング

B を使用してメッセージ キューを処理するために使用されます。 $q->consum( callback, [...] ) を使用すると、プログラムはメッセージを受信するたびに、コールバックで指定された関数が一度呼び出され、一定の時間になるまで終了しません。コールバック関数は FALSE を返します。

コールバックについて一言: PHP の call_back は、次のような配列の使用をサポートしています。 $c = new MyClass(); $c->counter = 100; $q->consume( array($c ,'myfunc) ') ) このようにして、作成した処理クラスを呼び出すことができます。 MyClass の myfunc のパラメータ定義は、上記の例の processMessage のパラメータ定義と同じです。

上記の例では、 $routingkey = '' を使用することは、すべてのメッセージを受信することを意味します。これを $routingkey = 'key_1' に変更すると、結果の唯一の内容が routingkey を key_1 に設定していることがわかります。

注: routingkey = 'key_1' と routingkey = 'key_2' は 2 つの異なるキューです。前提: client1 と client2 は両方とも key_1 のキューに接続されています。メッセージは client1 によって処理された後、client2 によって処理されません。 Routingkey = '' は別の代替方法です。 client_all は '' にバインドされます。すべてのメッセージが処理されると、client1 と client2 にはメッセージがなくなります。

プログラム設計の観点からは、交換の名前、さまざまな種類のタグを区別するためのキーの使用方法、メッセージが生成される場所にメッセージ送信コードを挿入する方法を計画する必要があります。バックエンド処理では、キーごとに 1 つ以上のクライアントを起動して、メッセージ処理のリアルタイム性を向上させることができます。マルチスレッドのメッセージ処理に PHP を使用する方法については、次のセクションで説明します。

その他のメッセージ モデルについては、以下を参照してください: http://www.rabbitmq.com/tutorials/tutorial-two-python.html