パート1では、ブローカーとしてRabbitMQを使用して、PHPのAMQPプロトコルの理論と単純なユースケースをカバーしました。それでは、いくつかのより高度な例に飛び込みましょう。

キーテイクアウト

PHPとRabbitMQを採(cǎi)用して、複數(shù)の労働者の間で非同期にデータを処理し、高遷移環(huán)境での効率を向上させます。
• rabbitmqに永続的なメッセージを?qū)g裝して、メッセージ「配信_(tái)mode」を2に設(shè)定し、キューを耐久性があると宣言することにより、サーバーのクラッシュ中のデータ損失を防ぐ。
rabbitmqのサービス品質(zhì)（QOS）設(shè)定を使用して、労働者間のメッセージ配信を制御し、単一の労働者が圧倒されないようにします。
• RPC（リモートプロシージャコール）のRabbitMQを利用して、ユーザー認(rèn)証などのタスクに役立つ応答を必要とするメッセージを送信して使用します。
RPC応答の排他的な一時(shí)的なキューを設(shè)定して、クライアントとサーバーの間でメッセージが正しくかつ安全に指示されるようにします。
• 応答の「Correlation_id」をリクエストと一致させ、返信の正しい処理と処理を確保することにより、RPC応答を効果的に処理します。
• 例1：複數(shù)の労働者の間で非同期データを処理するためのリクエストを送信
• 前の部分の例では、1人の生産者、1人の消費(fèi)者がいました。 消費(fèi)者が死亡した場(chǎng)合、消費(fèi)者が再び始まるまでメッセージはキューに積み上げられ続けます。その後、すべてのメッセージを1つずつ処理します。
• これは、1分あたりかなりの量のリクエストがある同時(shí)ユーザー環(huán)境では理想的ではありません。 幸いなことに、消費(fèi)者のスケーリングは非常に簡(jiǎn)単ですが、別の例を?qū)g裝しましょう。
請(qǐng)求書(shū)世代サービスがあるとしましょう。ユーザーは請(qǐng)求書(shū)番號(hào)を提供するだけで、システムはPDFファイルを自動(dòng)的に生成してユーザーにメールで送信します。 生成プロセスが実行されるサーバーがリソース制限である場(chǎng)合、電子メールの生成と送信には數(shù)秒かかる場(chǎng)合があります。 ここで、1秒あたりのいくつかのトランザクションをサポートする必要があると仮定します。サーバーを圧倒することなくこれをどのように達(dá)成できますか？

次のパターンを?qū)g裝する必要があります

プロデューサーのクラスを見(jiàn)てみましょう：

Workersender :: execute（）メソッドは請(qǐng)求書(shū)番號(hào)を受け取ります。 次に、通常どおり、接続、チャネル、キューを作成します。

今回は、メッセージオブジェクトを作成しているときに、コンストラクターが2番目のパラメーターを受信して??いることに注意してください。 この場(chǎng)合、RabbitMQサーバーがクラッシュした場(chǎng)合、メッセージは失われるべきではないと述べたいと思います。 これが機(jī)能するためには、キューも耐久性があると宣言する必要があることに注意してください。

次のコードを使用してフォームデータを受信し、プロデューサーを?qū)g行できます。

<span><span><?php
</span></span><span><span>namespace Acme<span>\AmqpWrapper</span>;
</span></span><span>
</span><span><span>use PhpAmqpLib<span>\Connection\AMQPConnection</span>;
</span></span><span><span>use PhpAmqpLib<span>\Message\AMQPMessage</span>;
</span></span><span>
</span><span><span>class WorkerSender
</span></span><span><span>{
</span></span><span>    <span>/* ... SOME OTHER CODE HERE ... */
</span></span><span>    
</span><span>    <span>/**
</span></span><span><span>     * Sends an invoice generation task to the workers
</span></span><span><span>     * 
</span></span><span><span>     * <span>@param <span>int</span> $invoiceNum
</span></span></span><span><span>     */ 
</span></span><span>    <span>public function execute($invoiceNum)
</span></span><span>    <span>{
</span></span><span>        <span>$connection = new AMQPConnection('localhost', 5672, 'guest', 'guest');
</span></span><span>        <span>$channel = $connection->channel();
</span></span><span>        
</span><span>        <span>$channel->queue_declare(
</span></span><span>            <span>'invoice_queue',    #queue - Queue names may be up to 255 bytes of UTF-8 characters
</span></span><span>            <span>false,              #passive - can use this to check whether an exchange exists without modifying the server state
</span></span><span>            <span>true,               #durable, make sure that RabbitMQ will never lose our queue if a crash occurs - the queue will survive a broker restart
</span></span><span>            <span>false,              #exclusive - used by only one connection and the queue will be deleted when that connection closes
</span></span><span>            <span>false               #auto delete - queue is deleted when last consumer unsubscribes
</span></span><span>            <span>);
</span></span><span>            
</span><span>        <span>$msg = new AMQPMessage(
</span></span><span>            <span>$invoiceNum,
</span></span><span>            <span>array('delivery_mode' => 2) # make message persistent, so it is not lost if server crashes or quits
</span></span><span>            <span>);
</span></span><span>            
</span><span>        <span>$channel->basic_publish(
</span></span><span>            <span>$msg,               #message 
</span></span><span>            <span>'',                 #exchange
</span></span><span>            <span>'invoice_queue'     #routing key (queue)
</span></span><span>            <span>);
</span></span><span>            
</span><span>        <span>$channel->close();
</span></span><span>        <span>$connection->close();
</span></span><span>    <span>}
</span></span><span><span>}</span></span>

快適な入力の消毒/検証のいずれかを使用してください

消費(fèi)者の面で物事が少し面白くなります：

<span><span><?php
</span></span><span><span>/* ... SOME CODE HERE ... */
</span></span><span>
</span><span>        <span>$msg = new AMQPMessage(
</span></span><span>            <span>$invoiceNum,
</span></span><span>            <span>array('delivery_mode' => 2) # make message persistent, so it is not lost if server crashes or quits
</span></span><span>            <span>);
</span></span><span>
</span><span><span>/* ... SOME CODE HERE ... */</span></span>

いつものように、接続を作成し、チャネルを?qū)Г訾?、キューを宣言する必要があります（キューのパラメーターはプロデューサーと同じである必要があります）。

ワーカーの行動(dòng)（いくつかのプロセスの間でメッセージを派遣）を行うには、$ channel-> basic_qos（）：

を使用して、サービス品質(zhì)（QoS）パラメーターを宣言する必要があります。

<span><span><?php
</span></span><span><span>chdir(dirname(__DIR__));
</span></span><span><span>require_once('vendor/autoload.php');
</span></span><span>
</span><span><span>use Acme<span>\AmqpWrapper\WorkerSender</span>;
</span></span><span>
</span><span><span>$inputFilters = array(
</span></span><span>    <span>'invoiceNo' => FILTER_SANITIZE_NUMBER_INT,
</span></span><span><span>);
</span></span><span><span>$input = filter_input_array(INPUT_POST, $inputFilters);
</span></span><span><span>$sender = new WorkerSender();
</span></span><span><span>$sender->execute($input['invoiceNo']);</span></span>

プリフェッチサイズ

：具體的な制限はありません。
• プリフェッチcount：承認(rèn)を送信する前に、労働者ごとに取得するメッセージの數(shù)。これにより、労働者は一度に1つのメッセージを処理できます。
• グローバル：nullとは、上記の設(shè)定がこの消費(fèi)者のみに適用されることを意味します
• 次に、パラメーターに重要な違いがあるため、消費(fèi)を開(kāi)始します。 RabbitMQサーバーにメッセージの処理が完了し、新しいものを受信する準(zhǔn)備ができている場(chǎng)合に、自動(dòng)ACKをオフにします。 さて、どうやってそのackを送信するのですか？ workerreceiver :: process（）メソッド（メッセージが受信されたときにコールバックメソッドとして宣言されます）をご覧ください。 generatedpdf（）およびsendemail（）メソッドへの呼び出しは、両方のタスクを達(dá)成するために費(fèi)やされた時(shí)間をシミュレートするダミーメソッドにすぎません。 $ MSGパラメーターには、プロデューサーから送信されたペイロードが含まれているだけでなく、プロデューサーが使用するオブジェクトに関する情報(bào)も含まれています。 プロデューサーが$ MSG-> DERVILY_INFO ['Channel']で使用するチャネルに関する情報(bào)を抽出できます（これは、$ connection-> channel（）;）で消費(fèi)者向けに開(kāi)いたチャネルと同じオブジェクトタイプです。 プロデューサーのチャンネルにプロセスが完了したことを確認(rèn)する必要があるため、basic_ack（）メソッドを使用して、パラメーターとして配信タグ（$ msg-> deleviry_info ['delivery_tag']）を送信します。 ACKがどのメッセージに屬するかに正しく関連付けるには。
どのようにして労働者を解雇しますか？次のようなファイルを作成するだけで、workerreceiver :: listen（）方法を呼び出します。

PHPコマンド（例：PHP Worker.phpまたは上記のファイルに與えた名前）を使用して、ワーカーを起動(dòng)します。 しかし、待って、目的は2人以上の労働者がいることでしたよね？問(wèn)題ありませんが、同じファイルの複數(shù)のプロセスを作成するために、同じ方法でより多くの労働者を起動(dòng)します。RabbitMQは、QoSパラメーターに従って消費(fèi)者を登録し、それらの間に作業(yè)を分配します。

例2：RPCリクエストを送信し、返信を期待してください

これまでのところ、ユーザーが返信を待つ必要なく、rabbitmqサーバーにメッセージを送信しています。 これは、ユーザーが「OK」メッセージを表示するためだけに費(fèi)やすことをいとわないよりも時(shí)間がかかる非同期プロセスではOKです。 しかし、実際に返信が必要な場(chǎng)合はどうなりますか？複雑な計(jì)算の結(jié)果を使用して、ユーザーに表示できるとしましょう。

アプリケーションの殘りの部分から分離された認(rèn)証メカニズムとして機(jī)能する集中ログインサーバー（シングルサインオン）があるとしましょう。 このサーバーに到達(dá)する唯一の方法は、rabbitmqを使用することです。 このサーバーにログイン資格情報(bào)を送信し、助成金/拒否アクセス応答を待つ方法を?qū)g裝する必要があります。

次のパターンを?qū)g裝する必要があります

いつものように、最初にプロデューサーを見(jiàn)てみましょう：

rpcsender :: executeメソッドを見(jiàn)ると、$ credentialsパラメーターは['username' => 'x' '、' password '=>' y ']の形式の配列であることに注意してください。 繰り返しますが、新しい接続を開(kāi)き、いつものようにチャネルを作成します。

<span><span><?php
</span></span><span><span>namespace Acme<span>\AmqpWrapper</span>;
</span></span><span>
</span><span><span>use PhpAmqpLib<span>\Connection\AMQPConnection</span>;
</span></span><span><span>use PhpAmqpLib<span>\Message\AMQPMessage</span>;
</span></span><span>
</span><span><span>class WorkerSender
</span></span><span><span>{
</span></span><span>    <span>/* ... SOME OTHER CODE HERE ... */
</span></span><span>    
</span><span>    <span>/**
</span></span><span><span>     * Sends an invoice generation task to the workers
</span></span><span><span>     * 
</span></span><span><span>     * <span>@param <span>int</span> $invoiceNum
</span></span></span><span><span>     */ 
</span></span><span>    <span>public function execute($invoiceNum)
</span></span><span>    <span>{
</span></span><span>        <span>$connection = new AMQPConnection('localhost', 5672, 'guest', 'guest');
</span></span><span>        <span>$channel = $connection->channel();
</span></span><span>        
</span><span>        <span>$channel->queue_declare(
</span></span><span>            <span>'invoice_queue',    #queue - Queue names may be up to 255 bytes of UTF-8 characters
</span></span><span>            <span>false,              #passive - can use this to check whether an exchange exists without modifying the server state
</span></span><span>            <span>true,               #durable, make sure that RabbitMQ will never lose our queue if a crash occurs - the queue will survive a broker restart
</span></span><span>            <span>false,              #exclusive - used by only one connection and the queue will be deleted when that connection closes
</span></span><span>            <span>false               #auto delete - queue is deleted when last consumer unsubscribes
</span></span><span>            <span>);
</span></span><span>            
</span><span>        <span>$msg = new AMQPMessage(
</span></span><span>            <span>$invoiceNum,
</span></span><span>            <span>array('delivery_mode' => 2) # make message persistent, so it is not lost if server crashes or quits
</span></span><span>            <span>);
</span></span><span>            
</span><span>        <span>$channel->basic_publish(
</span></span><span>            <span>$msg,               #message 
</span></span><span>            <span>'',                 #exchange
</span></span><span>            <span>'invoice_queue'     #routing key (queue)
</span></span><span>            <span>);
</span></span><span>            
</span><span>        <span>$channel->close();
</span></span><span>        <span>$connection->close();
</span></span><span>    <span>}
</span></span><span><span>}</span></span>

最初の違いは、キューを宣言することからです。 最初に、$ channel-> queue_declare（）の結(jié)果をキャッチするためにlist（）constructを使用していることに注意してください。これは、宣言しながらキュー名を明示的に送信しないため、このキューがどのように識(shí)別されるかを調(diào)べる必要があるためです。 結(jié)果アレイの最初の要素にのみ興味があります。これは、キューの一意の識(shí)別子（amq.gen-_u0kjvm8helfzqk9pz9ggのようなもの）になります。 2番目の変更は、このキューを排他的であると宣言する必要があることです。そのため、他の同時(shí)プロセスの結(jié)果に混同はありません。

<span><span><?php
</span></span><span><span>/* ... SOME CODE HERE ... */
</span></span><span>
</span><span>        <span>$msg = new AMQPMessage(
</span></span><span>            <span>$invoiceNum,
</span></span><span>            <span>array('delivery_mode' => 2) # make message persistent, so it is not lost if server crashes or quits
</span></span><span>            <span>);
</span></span><span>
</span><span><span>/* ... SOME CODE HERE ... */</span></span>

もう1つの大きな変更は、プロデューサーがキューの消費(fèi)者になることです。$ channel-> basic_consume（）を?qū)g行すると、キューを宣言している間に得た$ callback_queue値を提供していることに注意してください。 そして、すべての消費(fèi)者と同様に、プロセスが応答を受信したときに実行するためのコールバックを宣言します。

次に、メッセージの相関IDを作成する必要があります。これは、各メッセージの一意の識(shí)別子にすぎません。 この例では、uniqid（）の出力を使用していますが、好みのメカニズムを使用できます（人種條件を作成しない限り、強(qiáng)力な暗號(hào)セーフRNGである必要はありません）。

ここで、メッセージを作成しましょう。これには、最初の2つの例で慣れていたものと比較して重要な変更があります。認(rèn)証する資格情報(bào)を含むJSONエンコード文字列を割り當(dāng)てることは別として、AMQPMessageコンストラクターに2つのプロパティを定義した配列を提供する必要があります。

• correlation_id：メッセージのタグ
• REPLY_TO：キュー識(shí)別子が生成されたときに生成されます

メッセージを公開(kāi)した後、最初に空になる応答を評(píng)価します。 応答値が空のままですが、$ channel-> wait（）;

チャネルから応答を受信すると、コールバックメソッドが呼び出されます（rpcsender :: onresponse（））。 この方法は、受信した相関IDと生成された相関IDと一致し、それらが同じ場(chǎng)合、応答本體を設(shè)定し、したがってwhileループを破壊します。

RPC消費(fèi)者はどうですか？ ここにあります：

同じ古い接続とチャネル作成：）

<span><span><?php
</span></span><span><span>namespace Acme<span>\AmqpWrapper</span>;
</span></span><span>
</span><span><span>use PhpAmqpLib<span>\Connection\AMQPConnection</span>;
</span></span><span><span>use PhpAmqpLib<span>\Message\AMQPMessage</span>;
</span></span><span>
</span><span><span>class WorkerSender
</span></span><span><span>{
</span></span><span>    <span>/* ... SOME OTHER CODE HERE ... */
</span></span><span>    
</span><span>    <span>/**
</span></span><span><span>     * Sends an invoice generation task to the workers
</span></span><span><span>     * 
</span></span><span><span>     * <span>@param <span>int</span> $invoiceNum
</span></span></span><span><span>     */ 
</span></span><span>    <span>public function execute($invoiceNum)
</span></span><span>    <span>{
</span></span><span>        <span>$connection = new AMQPConnection('localhost', 5672, 'guest', 'guest');
</span></span><span>        <span>$channel = $connection->channel();
</span></span><span>        
</span><span>        <span>$channel->queue_declare(
</span></span><span>            <span>'invoice_queue',    #queue - Queue names may be up to 255 bytes of UTF-8 characters
</span></span><span>            <span>false,              #passive - can use this to check whether an exchange exists without modifying the server state
</span></span><span>            <span>true,               #durable, make sure that RabbitMQ will never lose our queue if a crash occurs - the queue will survive a broker restart
</span></span><span>            <span>false,              #exclusive - used by only one connection and the queue will be deleted when that connection closes
</span></span><span>            <span>false               #auto delete - queue is deleted when last consumer unsubscribes
</span></span><span>            <span>);
</span></span><span>            
</span><span>        <span>$msg = new AMQPMessage(
</span></span><span>            <span>$invoiceNum,
</span></span><span>            <span>array('delivery_mode' => 2) # make message persistent, so it is not lost if server crashes or quits
</span></span><span>            <span>);
</span></span><span>            
</span><span>        <span>$channel->basic_publish(
</span></span><span>            <span>$msg,               #message 
</span></span><span>            <span>'',                 #exchange
</span></span><span>            <span>'invoice_queue'     #routing key (queue)
</span></span><span>            <span>);
</span></span><span>            
</span><span>        <span>$channel->close();
</span></span><span>        <span>$connection->close();
</span></span><span>    <span>}
</span></span><span><span>}</span></span>

キューを宣言するのと同じ

>このキューには事前定義された名前（ '

rpc_queue

’）があります。 自動(dòng)アクセスを無(wú)効にするため、QoSパラメーターを定義します。そのため、資格情報(bào)の検証が完了して結(jié)果が得られることを通知できます。 魔法は宣言されたコールバック內(nèi)から來(lái)ます。資格情報(bào)の認(rèn)証が完了したら（はい、靜的ユーザー名/パスワード値に対してプロセスが実行されることがわかります。このチュートリアルは、クレデンシャルを認(rèn)証する方法に関するものではありません;））作成されました。 $ req-> get（ 'correlation_id'）でリクエストメッセージからこれを抽出し、プロデューサーで行ったのと同じようにこの値を渡すことができます。

<span><span><?php
</span></span><span><span>/* ... SOME CODE HERE ... */
</span></span><span>
</span><span>        <span>$msg = new AMQPMessage(
</span></span><span>            <span>$invoiceNum,
</span></span><span>            <span>array('delivery_mode' => 2) # make message persistent, so it is not lost if server crashes or quits
</span></span><span>            <span>);
</span></span><span>
</span><span><span>/* ... SOME CODE HERE ... */</span></span>

ここで、このメッセージをプロデューサー（「ランダム」名を持つもの）で作成された同じキューに公開(kāi)する必要があります。 $ req-> get（ 'Reply_to'）でキューの名前を抽出し、basic_publish（）のルーティングキーとして使用します。

メッセージを公開(kāi)したら、$ req-> delivice_info ['channel'] - > basic_ack（）でack通知をチャネルに送信する必要があります。 ]したがって、プロデューサーは待つのをやめることができます

繰り返しますが、リスニングプロセスを起動(dòng)すると、準(zhǔn)備ができています。 例2と3を組み合わせて、複數(shù)の労働者を起動(dòng)するだけでスケーリングできるマルチワーカーRPCプロセスを?qū)g行するためのマルチワーカーRPCプロセスを?qū)g行することもできます。

<span><span><?php
</span></span><span><span>chdir(dirname(__DIR__));
</span></span><span><span>require_once('vendor/autoload.php');
</span></span><span>
</span><span><span>use Acme<span>\AmqpWrapper\WorkerSender</span>;
</span></span><span>
</span><span><span>$inputFilters = array(
</span></span><span>    <span>'invoiceNo' => FILTER_SANITIZE_NUMBER_INT,
</span></span><span><span>);
</span></span><span><span>$input = filter_input_array(INPUT_POST, $inputFilters);
</span></span><span><span>$sender = new WorkerSender();
</span></span><span><span>$sender->execute($input['invoiceNo']);</span></span>

仮想ホスト、交換タイプ、サーバー管理など、RabbitMQやAMQPについてははるかに多くのことが言われています。ここおよびドキュメントページで、より多くのアプリケーションパターン（ルーティング、トピックなど）を見(jiàn)つけることができます。 また、RabbitMQを管理するためのコマンドラインツールと、Webベースのインターフェイスもあります。

このチュートリアルシリーズが気に入っていて、MQやより現(xiàn)実的なユースケースについて詳しく見(jiàn)たい場(chǎng)合は、以下のコメントでお知らせください！

PHP Rabbitmqの高度な例に関するよくある質(zhì)問(wèn)（FAQ）

phpにおけるrabbitmqの役割は何ですか？

rabbitmqは、アプリケーションが互いに非同期に通信できるメッセージブローカーです。 PHPアプリケーションでは、高負(fù)荷と複雑なタスクをより効率的に処理できるようにすることにより、PHPアプリケーションで重要な役割を果たします。 RabbitMQは、高度なメッセージキューイングプロトコル（AMQP）を使用して、アプリケーションのさまざまな部分間のメッセージの交換を容易にします。これにより、プロセスのデカップリングが可能になり、アプリケーションがよりスケーラブルで回復(fù)力が高くなります。あなたのマシン。これは、公式のRabbitMQ Webサイトで実行できます。サーバーがインストールされた後、PHP AMQP拡張機(jī)能をインストールできます。これにより、RabbitMQと対話するために必要な機(jī)能が提供されます。これは、Command PECLインストールAMQPを使用してPECLインストーラーを使用して実行できます。 amqpchannelクラスの。このメソッドは、交換の名前、交換のタイプ（直接、トピック、ファンアウト、またはヘッダー）、パッシブ、耐久性、auto_delete、引數(shù)などのオプションのパラメーターを含むいくつかのパラメーターを取ります。 phpのrabbitmq queueにメッセージを送信しますか？

phpのrabbitmqキューにメッセージを送信するには、最初に作成する必要がありますメッセージコンテンツを含むAMQPMessageクラスのインスタンス。次に、amqpchannelクラスのBasic_publishメソッドを使用して、キューにメッセージを送信できます。 BASIC_PUBLISHメソッドは、メッセージ、交換、およびルーティングキーをパラメーターとして使用します。 rabbitmq queue amqpchannelクラスのbasic_consumeメソッドを使用します。この方法では、キュー名、コンシューマータグ、NO_LOCAL、NO_ACK、排他的、およびメッセージが受信されたときに実行されるコールバック関數(shù)など、いくつかのパラメーターが必要です。 ？

PHPを使用したRabbitMQでのエラー処理は、トライキャッチブロックを使用して実行できます。 PHP AMQP拡張は、エラーが発生したときにAMQPExceptionクラスの例外をスローします。これらの例外をキャッチし、アプリケーションのニーズに応じて処理できます。 AMQPMESSAGEクラスの配信のプロパティ2から2。サーバーがクラッシュまたは再起動(dòng)します。

rabbitmqを使用してrabbitmqで優(yōu)先キューを?qū)g裝するには、rabbitmqの優(yōu)先キューをPHPに実裝できます。次に、メッセージを送信するときに、AMQPMessageクラスの優(yōu)先プロパティを指定した最大優(yōu)先度の値に設(shè)定できます。 > RabbitMQは、Reply-toプロパティをコールバックキューに設(shè)定したメッセージを送信することにより、PHPのリモートプロシージャコール（RPC）に使用できます。サーバーはコールバックキューに応答を送信でき、クライアントはそこからの応答を消費(fèi)できます。 RabbitMQマネジメントプラグインは、RabbitMQサーバーの監(jiān)視と管理のためのWebベースのインターフェイスを提供します。また、AMQPCHANNELクラスのメソッドを使用して、承認(rèn)されていないメッセージの數(shù)など、チャネルの狀態(tài)に関する情報(bào)を取得することもできます。

以上がPHPおよびRabbitMQ：高度な例の詳細(xì)內(nèi)容です。詳細(xì)については、PHP 中國(guó)語(yǔ) Web サイトの他の関連記事を參照してください。