消费不确认来自 RabbitMq 的消息

Question

我创建了一个简单的发布者和一个使用

basic.consume

在队列上订阅的消费者。

当作业运行无异常时，我的消费者会确认消息。每当我遇到异常时，我都不会确认消息并提前返回。只有已确认的消息才会从队列中消失，因此工作正常。
现在我希望消费者再次获取失败的消息，但重新使用这些消息的唯一方法是重新启动消费者。

我需要如何处理这个用例？

设置代码

$channel = new AMQPChannel($connection);

$exchange = new AMQPExchange($channel);

$exchange->setName('my-exchange');
$exchange->setType('fanout');
$exchange->declare();

$queue = new AMQPQueue($channel);
$queue->setName('my-queue');
$queue->declare();
$queue->bind('my-exchange');

消费者代码

$queue->consume(array($this, 'callback'));

public function callback(AMQPEnvelope $msg)
{
    try {
        //Do some business logic
    } catch (Exception $ex) {
        //Log exception
        return;
    }
    return $queue->ack($msg->getDeliveryTag());
}

生产者代码

$exchange->publish('message');

Answer 1

如果消息未被确认并且应用程序失败，它将自动重新传递，并且信封上的

redelivered

属性将设置为

true

（除非您使用

no-ack = true

标志使用它们）。

更新：

您必须在您的 catch 块中

nack

发送带有重新传递标志的消息

    try {
        //Do some business logic
    } catch (Exception $ex) {
        //Log exception
        return $queue->nack($msg->getDeliveryTag(), AMQP_REQUEUE);
    }

当心无限的 nacked 消息，而重新传递计数在 RabbitMQ 和 AMQP 协议中根本没有实现，除非您使用提供 x-delivery-count 消息头的

Quorum Queues

（请参阅Poison 消息处理）。

如果您不想弄乱此类消息，而只是想添加一些延迟，则可能需要在

sleep()

方法调用之前添加一些

usleep()

或

nack

，但这根本不是一个好主意。

有多种技术可以处理周期重新交付问题：

1。依靠死信交换

• 优点：可靠、标准、清晰
• 缺点：需要额外的逻辑

2。使用每条消息或每个队列 TTL

• 优点：易于实施，也标准、清晰
• 缺点：排长队可能会丢失一些消息

示例（请注意，对于队列 ttl，我们仅传递数字，对于消息 ttl - 任何数字字符串）：

2.1 每条消息 ttl：

$queue = new AMQPQueue($channel);
$queue->setName('my-queue');
$queue->declareQueue();
$queue->bind('my-exchange');

$exchange->publish(
    'message at ' . microtime(true),
    null,
    AMQP_NOPARAM,
    array(
        'expiration' => '1000'
    )
);

2.2。每个队列 ttl：

$queue = new AMQPQueue($channel);
$queue->setName('my-queue');
$queue->setArgument('x-message-ttl', 1000);
$queue->declareQueue();
$queue->bind('my-exchange');

$exchange->publish('message at ' . microtime(true));

3.在消息正文或标头中保留重新发送计数或剩余重新发送数量（也称为 IP 堆栈中的跳数限制或 ttl）

• 优点：在应用程序级别
为您提供对消息生命周期的额外控制
• 缺点：当您必须修改消息并再次发布时，开销很大，特定于应用程序，不清楚

代码：

$queue = new AMQPQueue($channel);
$queue->setName('my-queue');
$queue->declareQueue();
$queue->bind('my-exchange');

$exchange->publish(
    'message at ' . microtime(true),
    null,
    AMQP_NOPARAM,
    array(
        'headers' => array(
            'ttl' => 100
        )
    )
);

$queue->consume(
    function (AMQPEnvelope $msg, AMQPQueue $queue) use ($exchange) {
        $headers = $msg->getHeaders();
        echo $msg->isRedelivery() ? 'redelivered' : 'origin', ' ';
        echo $msg->getDeliveryTag(), ' ';
        echo isset($headers['ttl']) ? $headers['ttl'] : 'no ttl' , ' ';
        echo $msg->getBody(), PHP_EOL;

        try {
            //Do some business logic
            throw new Exception('business logic failed');
        } catch (Exception $ex) {
            //Log exception
            if (isset($headers['ttl'])) {
                // with ttl logic

                if ($headers['ttl'] > 0) {
                    $headers['ttl']--;

                    $exchange->publish($msg->getBody(), $msg->getRoutingKey(), AMQP_NOPARAM, array('headers' => $headers));
                }

                return $queue->ack($msg->getDeliveryTag());
            } else {
                // without ttl logic
                return $queue->nack($msg->getDeliveryTag(), AMQP_REQUEUE); // or drop it without requeue
            }

        }

        return $queue->ack($msg->getDeliveryTag());
    }
);

可能还有其他一些方法可以更好地控制消息重新传递流。

结论：没有灵丹妙药的解决方案。您必须决定哪种解决方案最适合您的需求或找出其他解决方案，但不要忘记在这里分享;)

Answer 2

如果您不想重新启动消费者，那么

basic.recover

AMQP 命令可能就是您想要的。根据AMQP协议：

basic.recover(bit requeue)

Redeliver unacknowledged messages.

This method asks the server to redeliver all unacknowledged messages on a specified channel. 
Zero or more messages may be redelivered. This method replaces the asynchronous Recover.