Stworzyłem prostego wydawcę i konsumenta, który subskrybuje kolejkę za pomocąbasic.consume.

Mój konsument potwierdza wiadomości, gdy zadanie działa bez wyjątku. Ilekroć spotykam wyjątek, nie potwierdzam wiadomości i wracam wcześniej. Tylko potwierdzone wiadomości znikają z kolejki, więc działa poprawnie.
Teraz chcę, aby konsument ponownie odebrał nieudane wiadomości, ale jedynym sposobem ponownego rozważenia tych wiadomości jest ponowne uruchomienie konsumenta.

Jak muszę podejść do tego przypadku użycia?

Kod konfiguracji

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

Kod konsumenta

$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());
}

Kod producenta

$exchange->publish('message');