我如何使用RabbitMQ延迟消息队列从PHP - How do I use the RabbitMQ delayed message queue from PHP?

How do I use the RabbitMQ delayed message queue from PHP?

我正在尝试使用延迟消息队列从PHP RabbitMQ，但我的消息只是消失。

我用下面的代码声明交换:

$this->channel->exchange_declare(
    'delay',
    'x-delayed-message',
    false,  /* passive, create if exchange doesn't exist */
    true,   /* durable, persist through server reboots */
    false,  /* autodelete */
    false,  /* internal */
    false,  /* nowait */
    ['x-delayed-type' => ['S', 'direct']]);

我用下面的代码绑定队列:

$this->channel->queue_declare(
    $queueName,
    false,  /* Passive */
    true,   /* Durable */
    false,  /* Exclusive */
    false   /* Auto Delete */
);
$this->channel->queue_bind($queueName, "delay", $queueName);

我要发布一条消息，代码如下:

$msg = new AMQPMessage(json_encode($msgData), [
    'delivery_mode' => 2,
    'x-delay' => 5000]);
$this->channel->basic_publish($msg, 'delay', $queueName);

但是消息不会被延迟;仍然是立即送达。我错过了什么?

从这里，

消息创建应该是

require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib'Message'AMQPMessage;
use PhpAmqpLib'Wire'AMQPTable;
$msg = new AMQPMessage($data,
            array(
                'delivery_mode' => 2, # make message persistent
                'application_headers' => new AMQPTable([
                    'x-delay' => 5000
                ])
            )
        );

适合那些需要消息延迟但又不想深入研究细节的人。您只需要做一些事情就可以使它工作:

安装amqp互操作兼容传输，例如enqueue/amqp-bunny和enqueue/amqp-tools。

composer require enqueue/amqp-bunny enqueue/amqp-tools

创建amqp上下文，添加延迟策略并发送延迟消息:

<?php
use Enqueue'AmqpTools'RabbitMqDelayPluginDelayStrategy;
use Enqueue'AmqpBunny'AmqpConnectionFactory;
$context = (new AmqpConnectionFactory('amqp://'))->createContext();
$context->setDelayStrategy(new RabbitMqDelayPluginDelayStrategy())
$queue = $context->createQueue('foo');
$context->declareQueue($queue);
$message = $context->createMessage('Hello world!');
$context->createProducer()
    ->setDeliveryDelay(5000) // 5 sec
    ->send($queue, $message)
;

顺便说一下，这不是唯一可用的策略。有一个是基于RabbitMQ死信队列+ ttl的。也可以用同样的方法。

您需要一个路由密钥从交换器发布到相关队列。

向内置的直接交换器发布信息之所以有效，是因为该交换器是一种特殊情况，它使用路由密钥作为目标队列名称。

对于您创建的所有交换器和队列，您需要使用路由密钥在交换器和队列之间创建绑定。然后使用该路由密钥而不是目标队列名称发布消息。

我不知道创建绑定的PHP代码…但它通常看起来像这样:

channel.bind(exhange_name, queue_name, routing_key)

然后在你发布消息时:

$this->channel->basic_publish($msg, 'delay', $routing_key);