Magento 2 RabbitMQ 如何及在哪里修改delivery_mode实现消息持久化

你查看的vendor/magento/framework-amqp/Queue.php中的push方法仅适用于直接向指定队列投递消息的小众场景，Magento 90%以上的异步队列消息都是通过交换机路由分发的，这部分的消息属性配置不在该类中定义，所以即使这里默认已经设置了delivery_mode = 2，你实际使用的消息依然不会生效。

负责交换机消息发布的类为Magento\Framework\Amqp\Exchange，其publish方法才是你需要调整的目标，该方法默认没有配置delivery_mode参数，这是消息无法持久化的核心原因。

推荐使用Magento类偏好机制重写逻辑，步骤如下：

1. 在你的自定义模块的etc/di.xml中添加偏好声明：

<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="urn:magento:framework:ObjectManager/etc/config.xsd">
    <preference for="Magento\Framework\Amqp\Exchange" type="你的Vendor\你的模块名\Amqp\Exchange" />
</config>

2. 新建重写类文件app/code/你的Vendor/你的模块名/Amqp/Exchange.php，代码如下：

<?php
namespace 你的Vendor\你的模块名\Amqp;

use Magento\Framework\Amqp\Exchange as BaseExchange;
use Magento\Framework\MessageQueue\EnvelopeInterface;
use PhpAmqpLib\Message\AMQPMessage;

class Exchange extends BaseExchange
{
    /**
     * @inheritdoc
     */
    public function publish($topic, EnvelopeInterface $envelope, $flags = 0, $attributes = [])
    {
        $messageProperties = $envelope->getProperties();
        $msg = new AMQPMessage(
            $envelope->getBody(),
            [
                'correlation_id' => $messageProperties['correlation_id'] ?? '',
                'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT
            ]
        );
        $this->channel->basic_publish($msg, $this->name, $topic, $flags, $attributes);
    }
}

修改完成后依次执行以下命令：

• bin/magento setup:di:compile
• bin/magento cache:clean
• 重启所有队列消费者进程

后续发布的消息将默认携带持久化属性，RabbitMQ服务重启后未消费的消息不会丢失。

内容的提问来源于stack exchange，提问作者user14141548