php – 从RabbitMq消费不确认消息

$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');

如果消息未被确认并且应用程序发生故障,则将自动重新发送,并且将信封上的重新传递属性设置为true(除非使用no-ack = true标志消费).

UPD：

您必须在catch块中使用重新发送标志来消息

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

当RabbitMQ和AMQP协议中没有实现重新传递计数时,请注意无限制的Nacked消息.

如果你不想混淆这样的消息,只是想添加一些延迟,你可能想在nack方法调用之前添加一些sleep()或usleep(),但这根本不是一个好主意.

有多种技术来处理循环重传问题：

依靠Dead Letter Exchanges

>职业：可靠,标准,清晰
cons：需要额外的逻辑

2.使用per message or per queue TTL

>职业：容易实施,也是标准,明确
cons：长排队你可能会丢掉一些信息

示例(注意,对于队列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)

>专业人士：在应用程序级别上给予您对消息生命周期的额外控制
> cons：显着的开销,而你必须修改消息并再次发布,具体应用,不清楚

码：

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

$exchange->publish(
    'message at ' . microtime(true),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(),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());
    }
);

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

结论：没有银弹解决方案.你必须决定什么解决方案适合你的需要最好或找出其他的东西,但不要忘记在这里分享)