如何推遲？ - php-amqplib

Question

我想知道如何延遲Amqpphplib。

我用這個偉大的咖啡腳本教程：

https://github.com/jamescarr/rabbitmq-scheduled-delivery

，但它似乎並不與PHP-amqplib工作。

消息按我的意願過期，但似乎「x-dead-letter-exchange」不做這項工作。我使用RabbitMQ管理控制檯，並且實時看到所有隊列的創建和刪除。但我的信息會在到期後進入即時隊列。我使用RabbitMQ 3.2.3版本，PHP-amqplib 2.2。*版本。

這裏是我的代碼：

Connection類：

class Connection 
{ 
/** 
* @var $ch 
*/ 
public $ch; 

/** 
* @var $consumer_tag 
*/ 
public $consumer_tag; 

/** 
* @var $exchange 
*/ 
public $exchange; 

/** 
* @var $conn 
*/ 
public $conn; 

public function __construct($host, $port, $user, $password, $vhost) 
{ 

    $this->exchange = 'immediate'; 
    $this->queue = 'right.now.queue'; 
    $this->consumer_tag = 'consumer'; 


    $this->conn = new AMQPConnection($host, $port, $user, $password, $vhost); 
    $this->ch = $this->conn->channel(); 

    $this->ch->exchange_declare($this->exchange, 'direct', false, true, false); 

    $this->ch->queue_declare($this->queue, false, true, false, false, false); 

    $this->ch->queue_bind($this->queue, $this->exchange); 


} 

public function createDelayedQueue ($name, $delay_seconds) { 
    $this->ch->queue_declare($name, false, false, false, true, true, array(
     "x-dead-letter-exchange" => array("S", $this->exchange), 
     "x-message-ttl" => array("I", $delay_seconds*1000), 
     "x-expires" => array("I", $delay_seconds*1000+1000) 
    )); 
} 
} 

發佈代碼

$name = 'send.later.'.$ts; 
$amqp->createDelayedQueue($name, 2); 
$msg = new AMQPMessage($msg_body, array('content_type' => 'text/plain', 'delivery_mode' => 2)); 
$amqp->ch->basic_publish($msg); 

消費者代碼

$amqp = $this->getContainer()->get('amqp_connexion'); 

    $amqp->ch->basic_consume($amqp->queue, $amqp->consumer_tag, false, false, false, false, function ($msg) { 

     echo $msg->body; 
     echo "\n--------\n"; 
    }); 

    $output->writeln('Listening '.$amqp->queue.'...'); 

    // Loop as long as the channel has callbacks registered 
    while (count($amqp->ch->callbacks)) { 
     $amqp->ch->wait(); 
    } 

Answer 1

我只是寫了PHP的簡化工作版本：

/////// simplified /////// 

// include the AMQPlib Classes || use an autoloader 

// queue/exchange names 
$queueRightNow = 'right.now.queue'; 
$exchangeRightNow = 'right.now.exchange'; 
$queueDelayed5sec = 'delayed.five.seconds.queue'; 
$exchangeDelayed5sec = 'delayed.five.seconds.exchange'; 

$delay = 5; // delay in seconds 

// create connection 
$AMQPConnection = new \PhpAmqpLib\Connection\AMQPConnection('localhost',5672,'guest','guest'); 

// create a channel 
$channel = $AMQPConnection->channel(); 

// create the right.now.queue, the exchange for that queue and bind them together 
$channel->queue_declare($queueRightNow); 
$channel->exchange_declare($exchangeRightNow, 'direct'); 
$channel->queue_bind($queueRightNow, $exchangeRightNow); 

// now create the delayed queue and the exchange 
$channel->queue_declare(
     $queueDelayed5sec, 
     false, 
     false, 
     false, 
     true, 
     true, 
     array(
      'x-message-ttl' => array('I', $delay*1000), // delay in seconds to milliseconds 
      "x-expires" => array("I", $delay*1000+1000), 
      'x-dead-letter-exchange' => array('S', $exchangeRightNow) // after message expiration in delay queue, move message to the right.now.queue 
     ) 
); 
$channel->exchange_declare($exchangeDelayed5sec, 'direct'); 
$channel->queue_bind($queueDelayed5sec, $exchangeDelayed5sec); 

// now create a message und publish it to the delayed exchange 
$msg = new \PhpAmqpLib\Message\AMQPMessage(
    time(), 
    array(
     'delivery_mode' => 2 
    ) 
); 
$channel->basic_publish($msg,$exchangeDelayed5sec); 


// consume the delayed message 
$consumeCallback = function(\PhpAmqpLib\Message\AMQPMessage $msg) { 
    $messagePublishedAt = $msg->body; 
    echo 'seconds between publishing and consuming: ' 
     . (time()-$messagePublishedAt) . PHP_EOL; 
}; 
$channel->basic_consume($queueRightNow, '', false, true, false, false, $consumeCallback); 

// start consuming 
while (count($channel->callbacks) > 0) { 
    $channel->wait(); 
} 

Answer 2

如果選擇amqp interop運輸爲主，你將不再需要在所有挖成的細節。只有幾件事要做：

安裝enqueue/amqp-lib（順便說一句，你可以使用其他傳輸基於amqp分機和偉大的兔子lib）運輸和enqueue/amqp-tools。

composer require enqueue/amqp-lib enqueue/amqp-tools 

創建AMQP情況下，添加一個拖延戰術，併發送延遲的消息：

<?php 
use Enqueue\AmqpTools\RabbitMqDlxDelayStrategy; 
use Enqueue\AmqpBunny\AmqpConnectionFactory; 

$context = (new AmqpConnectionFactory('amqp://'))->createContext(); 
$context->setDelayStrategy(new RabbitMqDlxDelayStrategy()) 

$queue = $context->createQueue('foo'); 
$context->declareQueue($queue); 

$message = $context->createMessage('Hello world!'); 

$context->createProducer() 
    ->setDeliveryDelay(5000) // 5 sec 
    ->send($queue, $message) 
; 

順便說一句，這不是可用這只是策略。有一個基於RabbitMQ延遲插件。它可以用同樣的方式。