а√天堂www在线а√天堂资源,а√天堂8资源在线官网,oldwoman老熟妇

>兔子在PHP中的作用是什麼？它通過使其能夠更有效地處理高負(fù)載和復(fù)雜任務(wù)，在PHP應(yīng)用中起著至關(guān)重要的作用。 RabbitMQ使用高級消息排隊協(xié)議（AMQP）來促進(jìn)應(yīng)用程序不同部分之間的消息交換。這允許流程解耦，使應(yīng)用程序更具可擴展性和彈性。你的機器。這可以通過官方的RabbitMQ網(wǎng)站完成。安裝服務(wù)器後，您可以安裝PHP AMQP擴展名，該擴展提供了與RabbitMQ交互的必要功能。可以使用命令pecl install amqp。 AMQPCHANNEL課程。此方法採用多個參數(shù)，包括交換的名稱，交換的類型（直接，主題，粉絲或標(biāo)題）以及可選參數(shù)，例如被動，耐用，auto_delete和grignestes。我是否在php中的兔子隊列中發(fā)送消息？帶有消息內(nèi)容的AMQPMESSAGE類。然後，您可以使用AMQPCHANNEL類的Basic_Publish方法將消息發(fā)送到隊列。 BASIC_PUBLISH方法將消息，交換和路由密鑰作為參數(shù)。

>如何確保使用php的兔子中的消息持久性？ AMQPMESSAGE類的velivery_mode屬性到2。這將使RabbitMQ將消息存儲在磁盤上，即使RabbitMQ服務(wù)器崩潰或重新啟動。

>如何通過php？

我如何在PHP中使用RabbitMQ作為RPC？ > RABBITMQ可以通過向回調(diào)隊列發(fā)送帶有回復(fù)屬性設(shè)置的消息，用於PHP中的遠(yuǎn)程過程調(diào)用（RPC）。然後，服務(wù)器可以將響應(yīng)發(fā)送到回調(diào)隊列，並且客戶端可以從那裡消耗響應(yīng)。

>監(jiān)視PHP中的RabbitMQ中的RabbitMQ。 RabbitMQ管理插件，該插件提供了一個基於Web的接口，用於監(jiān)視和管理RabbitMQ服務(wù)器。您還可以使用AMQPCHANNEL類的方法來獲取有關(guān)頻道狀態(tài)的信息，例如未經(jīng)確定的消息的數(shù)量。

首頁

後端開發(fā)

php教程

PHP和RabbitMQ：高級示例

Jennifer Aniston

Feb 19, 2025 am 09:44 AM

PHP and RabbitMQ: Advanced Examples

PHP和RabbitMQ：高級示例

在第1部分中，我們涵蓋了PHP中AMQP協(xié)議的簡單用例，並以RabbitMQ為代理。現(xiàn)在，讓我們深入研究一些更高級的例子。

鑰匙要點

通過發(fā)送需要響應(yīng)的消息，可用於使用RABBITMQ進(jìn)行RPC（遠(yuǎn)程過程調(diào)用）

>>>>> 通過將響應(yīng)中的“相關(guān)性_id”與請求匹配，確保正確處理和處理答復(fù)。

>示例1：發(fā)送請求以在幾個工人中對數(shù)據(jù)異步處理

在上一部分的示例中，我們有一個生產(chǎn)者，一個消費者。如果消費者死亡，則消息將繼續(xù)堆積在隊列中，直到消費者再次開始。然後，它將處理所有消息。

在並發(fā)的用戶環(huán)境中，每分鐘請求相當(dāng)多，這可能是不理想的。幸運的是，擴展消費者非常容易，但讓我們實現(xiàn)另一個示例。

>假設(shè)我們有一項發(fā)票生成服務(wù)，用戶只需要提供發(fā)票號碼，系統(tǒng)將自動生成PDF文件並將其通過電子郵件發(fā)送給用戶。如果生成過程運行的服務(wù)器受到資源有限，則生成和發(fā)送電子郵件可能需要幾秒鐘。現(xiàn)在假設(shè)我們需要每秒支持幾筆交易，我們?nèi)绾卧诓粔旱狗?wù)器的情況下完成此操作？

我們需要實現(xiàn)以下模式：

讓我們看一下我們的製作人班：

> WorkerSender :: execute（）方法將收到一個發(fā)票號。接下來，我們像往常一樣創(chuàng)建一個連接，頻道和隊列。

>請注意，這次，在創(chuàng)建消息對象時，構(gòu)造函數(shù)將接收第二個參數(shù)：array（'veliver_mode'=> 2）。在這種情況下，我們要聲明，如果RabbitMQ服務(wù)器崩潰，則不會丟失消息。請注意，為了使這項工作，隊列也必須被聲明持久。

可以使用以下代碼接收表單數 use class

請使用您感到滿意的輸入消毒/驗證。 > (shù)據(jù)並執(zhí)行生產(chǎn)者：>

<span><span><?php <span>namespace Acme<span>\AmqpWrapper</span>; PhpAmqpLib<span>\Connection\AMQPConnection</span>; <span>use PhpAmqpLib<span>\Message\AMQPMessage</span>; WorkerSender <span>{ <span>/* ... SOME OTHER CODE HERE ... */

<span>/** * Sends an invoice generation task to the workers * * <span>@param <span>int</span> $invoiceNum ;<span><span>     */ <span>public function execute($invoiceNum) <span>{ <span>$connection = new AMQPConnection('localhost', 5672, 'guest', 'guest'); <span>$channel = $connection->channel();

<span>$channel->queue_declare( <span>'invoice_queue',    #queue - Queue names may be up to 255 bytes of UTF-8 characters <span>false,              #passive - can use this to check whether an exchange exists without modifying the server state <span>true,               #durable, make sure that RabbitMQ will never lose our queue if a crash occurs - the queue will survive a broker restart <span>false,              #exclusive - used by only one connection and the queue will be deleted when that connection closes <span>false               #auto delete - queue is deleted when last consumer unsubscribes <span>);

<span>$msg = new AMQPMessage( <span>$invoiceNum, <span>array('delivery_mode' => 2) # make message persistent, so it is not lost if server crashes or quits <span>);

<span>$channel->basic_publish( <span>$msg,               #message <span>'',                 #exchange <span>'invoice_queue'     #routing key (queue) <span>);

<span>$channel->close(); <span>$connection->close(); <span>} <span>}</span></span>

在消費者方面，事情變得有些有趣：

<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>

>像往常一樣，我們必須創(chuàng)建一個連接，得出頻道並聲明隊列（隊列的參數(shù)必須與生產(chǎn)者相同）。

<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>

>為了具有工人行為（幾個程序中的派遣消息），我們必須使用$ channel-> basic_qos（）聲明服務(wù)質(zhì)量（QoS）參數(shù)：>

：沒有具體的限制，我們可以擁有盡可能多的工人
：在發(fā)送確認(rèn)之前，每個工人要檢索多少消息。這將使工人一次處理1條消息。 >
：null表示以上設(shè)置僅適用於該消費者

現(xiàn)在，我們?nèi)绾伟l(fā)送Ack？請查看Workerreceiver :: process（）方法（接收到消息時，該方法稱為回調(diào)方法）。對生成PDF（）和sendemail（）方法的調(diào)用只是虛擬方法，它將模擬完成這兩個任務(wù)所花費的時間。 $ msg參數(shù)不僅包含從生產(chǎn)者發(fā)送的有效載荷，還包含有關(guān)生產(chǎn)者使用的對象的信息。我們可以使用$ msg-> velivery_info ['channel']提取有關(guān)生產(chǎn)者使用的通道的信息（這是我們使用$ Connection-> Channel（）;）為消費者打開的通道的對像類型。由於我們需要向生產(chǎn)者的頻道發(fā)送確認(rèn)我們已經(jīng)完成了該過程的確認(rèn)，因此我們將使用其basic_ack（）方法，將作為參數(shù)發(fā)送作為參數(shù)（$ msg-> delivery_info ['evely_tag']）rabbitmq自動生成的rabbitmq正確地將ACK屬於哪個消息相關(guān)聯(lián)。

我們?nèi)绾谓鈨l工人？只需創(chuàng)建以下文件，請調(diào)用Workerreceiver :: listing（）方法：

現(xiàn)在，使用php命令（例如php worker.php或您給出上述文件的任何名稱）來啟動工人。但是等等，目的是有兩個或更多工人，不是嗎？沒問題，以相同的方式啟動更多的工人，以便擁有同一文件的多個進(jìn)程，而RabbitMQ將根據(jù)QoS參數(shù)註冊消費者並在其中分發(fā)工作。

>示例2：發(fā)送RPC請求並期望回復(fù)

到目前為止，我們一直在向RabbitMQ服務(wù)器發(fā)送消息，而無需等待回復(fù)。對於異步過程，這可能比用戶願意花費更多的時間只是為了查看“確定”消息，這是可以的。但是，如果我們真的需要答復(fù)怎麼辦？假設(shè)一個複雜的計算結(jié)果，因此我們可以將其顯示給用戶？

>假設(shè)我們有一個集中式登錄服務(wù)器（單個登錄），該服務(wù)器將用作與應(yīng)用程序其餘部分隔離的身份驗證機制。到達(dá)該服務(wù)器的唯一方法是通過RabbitMQ。我們需要實現(xiàn)一種將登錄憑據(jù)發(fā)送到該服務(wù)器的方法，並等待授予/拒絕訪問響應(yīng)。

我們需要實現(xiàn)以下模式：

>照常，讓我們首先看一下製作人：>

>查看RPCSENDER ::執(zhí)行方法，請注意，$憑據(jù)參數(shù)是['username'=>'x'，'password'=>'y']的形式的數(shù)組。同樣，我們打開一個新連接並照常創(chuàng)建頻道。

<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>

第一個區(qū)別來自宣布隊列。首先請注意，我們正在使用list（）構(gòu)造來從$ channel-> queue_declare（）捕獲結(jié)果。這是因為我們在聲明時沒有明確發(fā)送隊列名稱，因此我們需要找出如何識別該隊列。我們只對結(jié)果數(shù)組的第一個元素感興趣，該元素將是隊列的唯一標(biāo)識符（例如AMQ.gen-_U0KJVM8HELFZQK9P0Z9GG）。第二個更改是我們需要將此隊列聲明為獨家，因此其他並發(fā)過程的結(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>

>另一個很大的更改是，在執(zhí)行$ channel-> basic_consume（）時，生產(chǎn)者也將成為隊列的消費者，請注意，我們在聲明隊列時提供了$ callback_queue值。像每個消費者一樣，我們將在該過程收到響應(yīng)時聲明一個回調(diào)要執(zhí)行。

接下來，我們必須為消息創(chuàng)建一個相關(guān)ID，這無非是每個消息的唯一標(biāo)識符。在示例中，我們正在使用uniqid（）的輸出，但是您可以使用自己喜歡的任何機制（只要它不創(chuàng)建種族條件，就不需要是強大的，加密的安全RNG）。

<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>

現(xiàn)在，讓我們創(chuàng)建一條消息，與我們在前兩個示例中使用的消息相比，它具有重要的變化。除了分配一個包含我們要身份驗證的憑證的JSON編碼字符串外，我們還必須提供給AMQPMESSAGE構(gòu)造函數(shù)一個具有兩個定義的屬性的數(shù)組：>

：在聲明它時生成的隊列標(biāo)識符>

>一旦我們收到來自頻道的響應(yīng)，將調(diào)用回調(diào)方法（rpcsender :: onResponse（））。此方法將與已接收的相關(guān)ID與生成的相關(guān)ID匹配，如果它們是相同的，則將設(shè)置響應(yīng)主體，從而破壞while循環(huán)。

。 RPC消費者呢？在這裡是：

相同的舊連接和頻道創(chuàng)建：）

>與聲明隊列相同，但是該隊列將具有預(yù)定義的名稱（'

<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

’）。我們將定義QoS參數(shù)，因為我們將停用自動ACK，因此我們可以在驗證憑據(jù)驗證並具有結(jié)果時通知。

魔術(shù)來自聲明的回調(diào)。一旦我們對憑據(jù)進(jìn)行身份驗證（是的，我知道該過程是針對靜態(tài)用戶名/密碼值完成的，本教程並不是關(guān)於如何對憑據(jù)進(jìn)行身份驗證;）），我們必須創(chuàng)建具有相同關(guān)聯(lián)ID的返回消息（生產(chǎn)者）創(chuàng)建。我們可以使用$ req-> get（'correlation_id'）從請求消息中提取此值，以相同的方式傳遞此值。

>現(xiàn)在，我們必須將此消息發(fā)佈到製作人（帶有“隨機”名稱）中創(chuàng)建的同一隊列。我們使用$ req-> get（'reply_to'）提取隊列名稱，並將其用作basic_publish（）。

<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>

中的路由密鑰

>發(fā)布消息後，我們必須使用$ req-> delivery_info ['channel''] - > basic_ack（）將ACK通知發(fā)送到頻道，使用$ req-> evelyse_info ['velivery_tag'' ]這樣生產(chǎn)者可以停止等待。

<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>

再次啟動聆聽過程，您就可以開始了。您甚至可以組合示例2和3以具有多工程師的RPC進(jìn)程來執(zhí)行身份驗證請求，而不是通過啟動幾名工人來縮放。

>關(guān)於RabbitMQ和AMQP的更多要說，例如虛擬主機，交換類型，服務(wù)器管理等……您可以在此處和文檔頁面上找到更多的應(yīng)用程序模式（例如路由，主題）。還有一個命令行工具來管理RabbitMQ，還有一個基於Web的接口。

如果您喜歡這個教程系列，並且想查看有關(guān)MQ和更多現(xiàn)實世界用例的更多信息，請在下面的評論中告訴我們！

>關(guān)於PHP RabbitMQ高級示例的常見問題（常見問題解答）

>兔子在PHP中的作用是什麼？它通過使其能夠更有效地處理高負(fù)載和復(fù)雜任務(wù)，在PHP應(yīng)用中起著至關(guān)重要的作用。 RabbitMQ使用高級消息排隊協(xié)議（AMQP）來促進(jìn)應(yīng)用程序不同部分之間的消息交換。這允許流程解耦，使應(yīng)用程序更具可擴展性和彈性。你的機器。這可以通過官方的RabbitMQ網(wǎng)站完成。安裝服務(wù)器後，您可以安裝PHP AMQP擴展名，該擴展提供了與RabbitMQ交互的必要功能?？梢允褂妹頿ecl install amqp。 AMQPCHANNEL課程。此方法採用多個參數(shù)，包括交換的名稱，交換的類型（直接，主題，粉絲或標(biāo)題）以及可選參數(shù)，例如被動，耐用，auto_delete和grignestes。我是否在php中的兔子隊列中發(fā)送消息？帶有消息內(nèi)容的AMQPMESSAGE類。然後，您可以使用AMQPCHANNEL類的Basic_Publish方法將消息發(fā)送到隊列。 BASIC_PUBLISH方法將消息，交換和路由密鑰作為參數(shù)。

>我如何在php中從php？

中的兔子隊列中消費消息，您可以在使用AMQPCHANNEL類的basic_consume方法兔隊列。此方法採用幾個參數(shù)，包括隊列名稱，消費者標(biāo)籤，no_local，no_ack，exclusive和一個回調(diào)函數(shù)，該函數(shù)將在接收到消息時將執(zhí)行。

>

我如何處理rabbitmq中的PHP中的錯誤可以使用try-catch塊在兔子中使用PHP中的兔子中的錯誤處理？發(fā)生錯誤時，PHP AMQP擴展會引發(fā)AMQPException類的異常。您可以捕獲這些異常並根據(jù)您的應(yīng)用程序的需求處理它們。

>如何確保使用php的兔子中的消息持久性？ AMQPMESSAGE類的velivery_mode屬性到2。這將使RabbitMQ將消息存儲在磁盤上，即使RabbitMQ服務(wù)器崩潰或重新啟動。

>如何通過php？

在兔子中的優(yōu)先級隊列在PHP中實現(xiàn)優(yōu)先級排隊，可以通過在聲明隊列時設(shè)置X-Max-Prifority參數(shù)來實現(xiàn)。然後，在發(fā)送消息時，您可以將AMQPMESSAGE類的優(yōu)先屬性設(shè)置為0和您指定的最大優(yōu)先級之間的值。