rabbitmq 示例_PHP和RabbitMQ：高级示例

ゝ一世哀愁。 2022-12-03 01:56 203阅读 0赞

rabbitmq 示例

In [part 1][] we covered the theory and a simple use case of the AMQP protocol in PHP with RabbitMQ as the broker. Now, let’s dive into some more advanced examples.

在[第1部分中，][part 1]我们介绍了以RabbitMQ作为代理PHP中AMQP协议的理论和简单用例。 现在，让我们深入研究一些更高级的示例。

## 示例1：发送请求以在多个工作程序之间异步处理数据 **(**Example 1: send request to process data asynchronously among several workers**)** ##

In the example of the previous part, we had one producer, one consumer. If the consumer died, messages would continue to stack up in the queue until the consumer started again. It would then process all the messages, one by one.

在上一部分的示例中，我们有一个生产者，一个消费者。 如果使用者死亡，则消息将继续堆积在队列中，直到使用者再次启动。 然后它将一一处理所有消息。

This can be less than ideal in a concurrent user environment with a fair amount of requests per minute. Fortunately, scaling the consumers is super easy, but let’s implement another example.

在每分钟有大量请求的并发用户环境中，这可能不理想。 幸运的是，扩展消费者非常容易，但是让我们实现另一个示例。

![enter image description here][]

Let’s say we have an invoice generation service, where the users just need to provide the invoice number, and the system will automatically generate a pdf file and email it to the user. Generating and sending the email could take even several seconds if the server on which the generation process runs is resource limited. Now suppose we are required to support several transactions per second, how do we accomplish this without overwhelming the server?

假设我们有一个发票生成服务，用户只需要提供发票编号，系统就会自动生成一个pdf文件并将其通过电子邮件发送给用户。 如果运行生成过程的服务器资源有限，则生成和发送电子邮件可能甚至需要花费几秒钟的时间。 现在假设我们需要每秒支持多个事务，我们如何在不使服务器不堪重负的情况下完成此任务？

We need to implement the following pattern:

我们需要实现以下模式：

![enter image description here][enter image description here 1]

Let’s look at our producer class:

让我们看一下我们的生产者类：

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

The `WorkerSender::execute()` method will receive an invoice number. Next we create a connection, channel and queue as usual.

`WorkerSender::execute()`方法将收到发票编号。 接下来，我们照常创建连接，通道和队列。

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

Please notice that this time, while creating the message object, the constructor receives a second parameter: `array('delivery_mode' => 2)`. In this case we want to state that the message should not be lost if the RabbitMQ server crashes. Please be aware that in order for this to work, the queue has to be declared durable, too.

请注意，这一次，在创建消息对象时，构造函数将接收第二个参数： `array('delivery_mode' => 2)` 。 在这种情况下，我们要声明的是，如果RabbitMQ服务器崩溃，则消息不会丢失。 请注意，为了使此工作正常进行，还必须将队列声明为持久队列。

The following code can be used to receive the form data and execute the producer:

以下代码可用于接收表单数据并执行生产者：

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

Please use whichever input sanitization/validation you feel comfortable with.

请使用您喜欢的任何输入消毒/验证。

Things get a little bit interesting on the consumer side:

消费者方面的事情变得有些有趣：

<?php
    namespace Acme\AmqpWrapper;
    
    use PhpAmqpLib\Connection\AMQPConnection;
    use PhpAmqpLib\Message\AMQPMessage;
    
    class WorkerReceiver
    {
        /* ... SOME OTHER CODE HERE ... */
        
        /**
         * Process incoming request to generate pdf invoices and send them through 
         * email.
         */ 
        public function listen()
        {
            $connection = new AMQPConnection('localhost', 5672, 'guest', 'guest');
            $channel = $connection->channel();
            
            $channel->queue_declare(
                'invoice_queue',    #queue
                false,              #passive
                true,               #durable, make sure that RabbitMQ will never lose our queue if a crash occurs
                false,              #exclusive - queues may only be accessed by the current connection
                false               #auto delete - the queue is deleted when all consumers have finished using it
                );
                
            /**
             * don't dispatch a new message to a worker until it has processed and 
             * acknowledged the previous one. Instead, it will dispatch it to the 
             * next worker that is not still busy.
             */
            $channel->basic_qos(
                null,   #prefetch size - prefetch window size in octets, null meaning "no specific limit"
                1,      #prefetch count - prefetch window in terms of whole messages
                null    #global - global=null to mean that the QoS settings should apply per-consumer, global=true to mean that the QoS settings should apply per-channel
                );
            
            /**
             * indicate interest in consuming messages from a particular queue. When they do 
             * so, we say that they register a consumer or, simply put, subscribe to a queue.
             * Each consumer (subscription) has an identifier called a consumer tag
             */ 
            $channel->basic_consume(
                'invoice_queue',        #queue
                '',                     #consumer tag - Identifier for the consumer, valid within the current channel. just string
                false,                  #no local - TRUE: the server will not send messages to the connection that published them
                false,                  #no ack, false - acks turned on, true - off.  send a proper acknowledgment from the worker, once we're done with a task
                false,                  #exclusive - queues may only be accessed by the current connection
                false,                  #no wait - TRUE: the server will not respond to the method. The client should not wait for a reply method
                array($this, 'process') #callback
                );
                
            while(count($channel->callbacks)) {
                $this->log->addInfo('Waiting for incoming messages');
                $channel->wait();
            }
            
            $channel->close();
            $connection->close();
        }
        
        /**
         * process received request
         * 
         * @param AMQPMessage $msg
         */ 
        public function process(AMQPMessage $msg)
        {
            $this->generatePdf()->sendEmail();
            
            /**
             * If a consumer dies without sending an acknowledgement the AMQP broker 
             * will redeliver it to another consumer or, if none are available at the 
             * time, the broker will wait until at least one consumer is registered 
             * for the same queue before attempting redelivery
             */ 
            $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
        }
        
        /**
         * Generates invoice's pdf
         * 
         * @return WorkerReceiver
         */ 
        private function generatePdf()
        {
            /**
             * Mocking a pdf generation processing time.  This will take between
             * 2 and 5 seconds
             */ 
            sleep(mt_rand(2, 5));
            return $this;
        }
        
        /**
         * Sends email
         * 
         * @return WorkerReceiver
         */ 
        private function sendEmail()
        {
            /**
             * Mocking email sending time.  This will take between 1 and 3 seconds
             */ 
            sleep(mt_rand(1,3));
            return $this;
        }
    }

As usual, we have to create a connection, derive a channel and declare a queue (the queue’s parameters have to be de same as the producer).

与往常一样，我们必须创建一个连接，派生一个通道并声明一个队列(该队列的参数必须与生产者相同)。

<?php
    /* ... SOME CODE HERE ... */
    
            $channel->basic_qos(
                null,   #prefetch size - prefetch window size in octets, null meaning "no specific limit"
                1,      #prefetch count - prefetch window in terms of whole messages
                null    #global - global=null to mean that the QoS settings should apply per-consumer, global=true to mean that the QoS settings should apply per-channel
                );
                
    /* ... SOME CODE HERE ... */

In order to have worker behavior (dispatch messages among several proceses), we have to declare the Quality of Service (qos) parameters with `$channel->basic_qos()`:

为了具有工作者行为(在多个过程中调度消息)，我们必须使用`$channel->basic_qos()`声明服务质量(qos)参数：

*  *Prefetch size*: no specific limit, we could have as many workers as we need
    
    *预取大小* ：没有具体限制，我们可以根据需要*容纳*任意数量的工人
 *  *Prefetch count*: how many messages to retrieve per worker before sending back an acknowledgement. This will make the worker to process 1 message at a time.
    
    *预取计数* ：发送回确认之前，每个工作人员要检索多少条消息。 这将使工作人员一次处理1条消息。
 *  *Global*: null means that above settings will apply to this consumer only
    
    *全局* ：null表示以上设置仅适用于此使用者

Next, we will start consuming, with a key difference in the parameters. We will turn off automatic ack’s, since we will tell the RabbitMQ server when we have finished processing the message and be ready to receive a new one.

接下来，我们将开始消耗，并在参数上有一个关键区别。 我们将关闭自动确认，因为我们将在处理完消息并准备接收新消息时告诉RabbitMQ服务器。

Now, how do we send that ack? Please take a look at the `WorkerReceiver::process()` method (which is declared as a callback method when a message is received). The calls to `generatedPdf()` and `sendEmail()` methods are just dummy methods that will simulate the time spent to accomplish both tasks. The `$msg` parameter not only contains the payload sent from the producer, it also contains information about the objects used by the producer. We can extract information about the channel used by the producer with `$msg->delivery_info['channel']` (which is the same object type to the channel we opened for the consumer with `$connection->channel();`). Since we need to send the producer’s channel an acknowledgement that we have completed the process, we will use its `basic_ack()` method, sending as a parameter the delivery tag (`$msg->delivery_info['delivery_tag']`) RabbitMQ automatically generated in order to associate correctly to which message the ack belongs to.

现在，我们如何发送该确认？ 请看一下`WorkerReceiver::process()`方法(在收到消息时被声明为回调方法)。 对`generatedPdf()`和`sendEmail()`方法的调用只是虚拟方法，它们将模拟完成这两项任务所花费的时间。 `$msg`参数不仅包含从生产者发送的有效负载，而且还包含有关生产者使用的对象的信息。 我们可以使用`$msg->delivery_info['channel']` (与我们使用`$connection->channel();`为消费者打开的通道相同的对象类型)提取有关生产者使用的通道的信息。 由于我们需要向生产者的频道发送确认已完成该过程的确认，因此我们将使用其`basic_ack()`方法， `$msg->delivery_info['delivery_tag']`顺序自动生成的交付标签( `$msg->delivery_info['delivery_tag']` )RabbitMQ作为参数发送正确关联确认属于哪个消息。

How do we fire up the workers? Just create a file like the following, invoking the `WorkerReceiver::listen()` method:

我们如何解雇工人？ 只需创建如下文件，并调用`WorkerReceiver::listen()`方法即可：

<?php
    chdir(dirname(__DIR__));
    
    require_once('vendor/autoload.php');
    
    use Acme\AmqpWrapper\WorkerReceiver;
    
    $worker = new WorkerReceiver();
    
    $worker->listen();

Now use the php command (e.g. `php worker.php` or whichever name you have given to above file) to fire up the worker. But wait, the purpose was to have two or more workers, wasn’t it? No problem, fire up more workers in the same way in order to have multiple processes of the same file, and RabbitMQ will register the consumers and distribute work among them according to the QoS parameters.

现在使用php命令(例如`php worker.php`或您在上面文件中指定的名称)启动工作程序。 但是等等，目的是要有两个或更多的工人，不是吗？ 没问题，以相同的方式激发更多的工人，以使同一文件具有多个进程，RabbitMQ将注册使用者并根据QoS参数在其中分配工作。

## 示例2：发送RPC请求并期望得到答复 **(**Example 2: send RPC requests and expect a reply**)** ##

So far, we have been sending messages to the RabbitMQ server without the user having to wait for a reply. This is ok for asynchronous processes that might take more time than the user is willing to spend just to see an ‘OK’ message. But what if we actually need a reply? Let’s say some result from a complex calculation, so we can show it to the user?

到目前为止，我们一直在向RabbitMQ服务器发送消息，而无需用户等待答复。 对于异步过程，这可能会花费更多时间，而异步过程可能会花费比用户仅愿意看到“ OK”消息更多的时间。 但是，如果我们实际上需要回复怎么办？ 假设某个复杂的计算得出了一些结果，那么我们可以将其显示给用户吗？

![enter image description here][enter image description here 2]

Let’s say we have a centralized login server (single sign on) that will work as an authentication mechanism isolated from the rest of our application(s). The only way to reach this server is through RabbitMQ. We need to implement a way to send the login credentials to this server and wait for a grant/deny access response.

假设我们有一个集中式登录服务器(单点登录)，该服务器将作为与其他应用程序隔离的身份验证机制。 到达此服务器的唯一方法是通过RabbitMQ。 我们需要实现一种将登录凭据发送到此服务器并等待授予/拒绝访问响应的方法。

![enter image description here][enter image description here 3]

We need to implement the following pattern:

我们需要实现以下模式：

As usual, let’s look at the producer first:

像往常一样，让我们先来看生产者：

<?php
    namespace Acme\AmqpWrapper;
    
    use PhpAmqpLib\Connection\AMQPConnection;
    use PhpAmqpLib\Message\AMQPMessage;
    
    class RpcSender
    {
        private $response;
    
        /**
         * @var string
         */
        private $corr_id;
    
        /* ...SOME OTHER CODE HERE... */
    
        /**
         * @param array $credentials
         * @return string
         */
        public function execute($credentials)
        {
        	$connection = new AMQPConnection('localhost', 5672, 'guest', 'guest');
    		$channel = $connection->channel();
    		
    		/*
    		 * creates an anonymous exclusive callback queue
    		 * $callback_queue has a value like amq.gen-_U0kJVm8helFzQk9P0z9gg
    		 */
    		list($callback_queue, ,) = $channel->queue_declare(
    			"", 	#queue
    			false, 	#passive
    			false, 	#durable
    			true, 	#exclusive
    			false	#auto delete
    			);
    			
    		$channel->basic_consume(
    			$callback_queue, 			#queue
    			'', 						#consumer tag
    			false, 						#no local
    			false, 						#no ack
    			false, 						#exclusive
    			false, 						#no wait
    			array($this, 'onResponse')	#callback
    			);
    			
    		$this->response = null;
    		
    		/*
    		 * $this->corr_id has a value like 53e26b393313a
    		 */
    		$this->corr_id = uniqid();
    		$jsonCredentials = json_encode($credentials);
    
    		/*
    		 * create a message with two properties: reply_to, which is set to the 
    		 * callback queue and correlation_id, which is set to a unique value for 
    		 * every request
    		 */
    		$msg = new AMQPMessage(
    		    $jsonCredentials,    #body
    		    array('correlation_id' => $this->corr_id, 'reply_to' => $callback_queue)    #properties
    			);
    		    
    		/*
    		 * The request is sent to an rpc_queue queue.
    		 */
    		$channel->basic_publish(
    			$msg,		#message 
    			'', 		#exchange
    			'rpc_queue'	#routing key
    			);
    		
    		while(!$this->response) {
    			$channel->wait();
    		}
    		
    		$channel->close();
    		$connection->close();
    		
    		return $this->response;
        }
    
        /**
         * When a message appears, it checks the correlation_id property. If it
         * matches the value from the request it returns the response to the
         * application.
         *
         * @param AMQPMessage $rep
         */
        public function onResponse(AMQPMessage $rep) {
        	if($rep->get('correlation_id') == $this->corr_id) {
    			$this->response = $rep->body;
    		}
    	}
    }

Looking at `RpcSender::execute` method, please be aware that the `$credentials` parameter is an array in the form of `['username'=>'x', 'password'=>'y']`. Again, we open a new connection and create a channel as usual.

在`RpcSender::execute`方法时，请注意， `$credentials`参数是`['username'=>'x', 'password'=>'y']`形式的数组。 同样，我们像往常一样打开一个新连接并创建一个频道。

<?php
    //...
            list($callback_queue, ,) = $channel->queue_declare(
                "",     #queue
                false,  #passive
                false,  #durable
                true,   #exclusive
                false   #auto delete
                );
    //...

The first difference comes from declaring a queue. First notice that we are using the `list()` construct to catch the result from `$channel->queue_declare()`. This is because we do not explicitly send a queue name while declaring it so we need to find out how this queue is identified. We are only interested in the first element of the result array, which will be an unique identifier of the queue (something like `amq.gen-_U0kJVm8helFzQk9P0z9gg`). The second change is that we need to declare this queue as exclusive, so there is no mix up in the results from other concurrent processes.

第一个区别来自声明队列。 首先请注意，我们正在使用`list()`构造函数来捕获`$channel->queue_declare()` 。 这是因为我们在声明队列名称时没有显式发送队列名称，因此我们需要找出如何标识此队列。 我们只对结果数组的第一个元素感兴趣，它将是队列的唯一标识符(类似于`amq.gen-_U0kJVm8helFzQk9P0z9gg` )。 第二个变化是我们需要将此队列声明为互斥，因此其他并发进程的结果不会混淆。

Another big change is that the producer will be a consumer of a queue too, when executing `$channel->basic_consume()` please notice that we are providing the `$callback_queue` value we got while declaring the queue. And like every consumer, we will declare a callback to execute when the process receives a response.

另一个重大变化是，生产者也将成为队列的使用者，在执行`$channel->basic_consume()`请注意，我们将提供在声明队列时获得的`$callback_queue`值。 与每个使用者一样，我们将声明一个回调，以在流程收到响应时执行。

<?php
    //...
            /*
             * $this->corr_id has a value like 53e26b393313a
             */
            $this->corr_id = uniqid();
    //...

Next, we have to create a correlation id for the message, this is nothing more than a unique identifier for each message. In the example we are using `uniqid()`’s output, but you can use whichever mechanism you prefer (as long as it does not create a race condition, does not need to be a strong, crypto-safe RNG).

接下来，我们必须为消息创建一个关联ID，这仅是每个消息的唯一标识符。 在该示例中，我们使用`uniqid()`的输出，但是可以使用您喜欢的任何一种机制(只要它不创建竞争条件，就不必是强大的，加密安全的RNG)。

<?php
    //...
            $msg = new AMQPMessage(
                $jsonCredentials,    #body
                array('correlation_id' => $this->corr_id, 'reply_to' => $callback_queue)    #properties
                );
    //...

Now let’s create a message, which has important changes compared to what we were used to in the first 2 examples. Aside from assigning a json-encoded string containing the credentials we want to authenticate, we have to provide to the `AMQPMessage` constructor an array with two properties defined:

现在，让我们创建一条消息，与前两个示例中使用的消息相比，它具有重要的变化。 除了分配一个包含要验证的凭据的json编码的字符串外，我们还必须向`AMQPMessage`构造函数提供一个数组，该数组定义了两个属性：

*  *correlation\_id*: a tag for the message
    
    *related\_id* ：消息的标签
 *  *reply\_to*: the queue identifier generated while declaring it
    
    *reply\_to* ：声明时生成的队列标识符

After publishing the message, we will evaluate the response, which will be empty at the beginning. While the response value remains empty, we will wait for a response from the channel with `$channel->wait();`.

发布消息后，我们将评估响应，该响应在开始时为空。 当响应值保持为空时，我们将使用`$channel->wait();`的响应`$channel->wait();` 。

Once we receive a response from the channel, the callback method will be invoked (`RpcSender::onResponse()`). This method will match the received correlation id against the one generated, and if they are the same, will set the response body, thus breaking the while loop.

一旦我们从通道收到响应，则将调用回调方法( `RpcSender::onResponse()` )。 此方法会将接收到的相关ID与生成的相关ID进行匹配，如果它们相同，则将设置响应主体，从而中断while循环。

What about the RPC consumer? Here it is:

RPC使用者呢？ 这里是：

<?php
    namespace Acme\AmqpWrapper;
    
    use PhpAmqpLib\Connection\AMQPConnection;
    use PhpAmqpLib\Message\AMQPMessage;
    
    class RpcReceiver
    {
        /* ... SOME OTHER CODE HERE... */
    
        /**
         * Listens for incoming messages
         */
        public function listen()
        {
            $connection = new AMQPConnection('localhost', 5672, 'guest', 'guest');
            $channel = $connection->channel();
            
            $channel->queue_declare(
                'rpc_queue',    #queue 
                false,          #passive
                false,          #durable
                false,          #exclusive
                false           #autodelete
                );
            
            $channel->basic_qos(
                null,   #prefetch size
                1,      #prefetch count
                null    #global
                );
                
            $channel->basic_consume(
                'rpc_queue',                #queue
                '',                         #consumer tag
                false,                      #no local
                false,                      #no ack
                false,                      #exclusive
                false,                      #no wait
                array($this, 'callback')    #callback
                );
                
            while(count($channel->callbacks)) {
                $channel->wait();
            }
            
            $channel->close();
            $connection->close();
        }
    
        /**
         * Executes when a message is received.
         *
         * @param AMQPMessage $req
         */
        public function callback(AMQPMessage $req) {
            
            $credentials = json_decode($req->body);
        	$authResult = $this->auth($credentials);
    
        	/*
        	 * Creating a reply message with the same correlation id than the incoming message
        	 */
        	$msg = new AMQPMessage(
        	    json_encode(array('status' => $authResult)),            #message
        	    array('correlation_id' => $req->get('correlation_id'))  #options
        	    );
        
        	/*
        	 * Publishing to the same channel from the incoming message
        	 */
        	$req->delivery_info['channel']->basic_publish(
        	    $msg,                   #message
        	    '',                     #exchange
        	    $req->get('reply_to')   #routing key
        	    );
        	    
        	/*
        	 * Acknowledging the message
        	 */
        	$req->delivery_info['channel']->basic_ack(
        	    $req->delivery_info['delivery_tag'] #delivery tag
        	    );
        }
    
        /**
         * @param \stdClass $credentials
         * @return bool
         */
        private function auth(\stdClass $credentials) {
        	if (($credentials->username == 'admin') && ($credentials->password == 'admin')) {
        	    return true;
        	} else {
        	    return false;
        	}
        }
    }

Same old connection and channel creation :)

相同的旧连接和频道创建:)

Same as declaring the queue, however this queue will have a predefined name (‘*rpc\_queue*‘). We will define the QoS parameters since we will deactivate automatic acks, so we can notify when we are finished verifying the credentials and have a result.

与声明队列相同，但是此队列将具有预定义的名称(' *rpc\_queue* ')。 我们将定义QoS参数，因为我们将停用自动确认，因此我们可以在完成验证凭证并得到结果时通知。

<?php
    //...
            $msg = new AMQPMessage(
                json_encode(array('status' => $authResult)),            #message
                array('correlation_id' => $req->get('correlation_id'))  #options
                );
    //...

The magic comes from within the declared callback. Once we are done authenticating the credentials (yes, I know the process is done against static username/password values, this tutorial is not about how to authenticate credentials ;) ), we have to create the return message with the same correlation id the producer created. We can extract this from the request message with `$req->get('correlation_id')`, passing this value the same way we did it in the producer.

魔术来自已声明的回调内。 一旦我们完成了对凭据的身份验证(是的，我知道该过程是针对静态用户名/密码值完成的，本教程不涉及如何对凭据进行身份验证;))，我们必须创建具有与生产者相同的相关ID的返回消息创建。 我们可以使用`$req->get('correlation_id')`从请求消息中提取该值，并以与在生产者中相同的方式传递该值。

<?php
    //...
            $req->delivery_info['channel']->basic_publish(
                $msg,                   #message
                '',                     #exchange
                $req->get('reply_to')   #routing key
                );
    //...

Now we have to publish this message to the same queue that was created in the producer (the one with the ‘random’ name). We extract the queue’s name with `$req->get('reply_to')` and use it as the routing key in `basic_publish()`.

现在，我们必须将此消息发布到在生产者中创建的相同队列(名称为“ random”的那个)。 我们使用`$req->get('reply_to')`提取队列的名称，并将其用作`basic_publish()`的路由键。

Once we published the message, we have to send the ack notice to the channel with `$req->delivery_info['channel']->basic_ack()`, using the delivery tag in `$req->delivery_info['delivery_tag']` so the producer can stop waiting.

发布消息后，我们必须使用`$req->delivery_info['delivery_tag']`的传递标签，通过`$req->delivery_info['channel']->basic_ack()`将确认通知发送到该通道。生产者可以停止等待。

Again, fire up a listening process and you are ready to go. You can even combine examples 2 and 3 to have a multi-worker rpc process to perform the authentication requests than can be scaled just by firing up several workers.

再次，启动监听过程，您就可以开始了。 您甚至可以将示例2和3组合在一起，以使多工作人员rpc流程执行身份验证请求，而不仅仅是通过解雇几个工作人员就可以对其进行扩展。

There is far more to be said about RabbitMQ and AMQP, like virtual hosts, exchange types, server administration, etc… you can find more application patterns (like routing, topics) [here][] and at the [documentation page][]. There is also a [command line tool][] to manage RabbitMQ, as well as a [web based interface][].

关于RabbitMQ和AMQP，还有更多要说的，例如虚拟主机，交换类型，服务器管理等……您可以[在此处][here]和[文档页面][documentation page]找到更多的应用程序模式(例如路由，主题)。 还有一个用于管理RabbitMQ的[命令行工具][command line tool] ，以及一个[基于Web的界面][web based interface] 。

If you liked this tutorial series and would like to see more about MQ and more real world use cases, please let us know in the comments below!

如果您喜欢本教程系列，并且希望了解有关MQ和更多实际使用案例的更多信息，请在下面的评论中告诉我们！

> 翻译自: [https://www.sitepoint.com/php-rabbitmq-advanced-examples/][https_www.sitepoint.com_php-rabbitmq-advanced-examples]

rabbitmq 示例

[part 1]: https://www.sitepoint.com/use-rabbitmq-php/
[enter image description here]: http://s6.postimg.org/ac06ys4m9/Screen_Shot_2014_09_10_at_11_15_32_AM.png
[enter image description here 1]: http://s6.postimg.org/ey9mszdcx/007_second_example.png
[enter image description here 2]: https://s6.postimg.org/8lh5xan35/Screen_Shot_2014_09_10_at_11_25_17_AM.png
[enter image description here 3]: https://s6.postimg.org/aqeuk8bxd/008_third_example.png
[here]: https://www.rabbitmq.com/getstarted.html
[documentation page]: https://www.rabbitmq.com/documentation.html
[command line tool]: https://www.rabbitmq.com/management-cli.html
[web based interface]: https://www.rabbitmq.com/management.html
[https_www.sitepoint.com_php-rabbitmq-advanced-examples]: https://www.sitepoint.com/php-rabbitmq-advanced-examples/