14
Kürzlich habe ich eine schnelle Implementierung auf Erzeuger/Verbraucher-Queue-System.Implementierung der verzögerten Warteschlange für PHP AMQP
<?php
namespace Queue;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Wire\AMQPTable;
class Amqp
{
private $connection;
private $queueName;
private $delayedQueueName;
private $channel;
private $callback;
public function __construct($host, $port, $login, $password, $queueName)
{
$this->connection = new AMQPStreamConnection($host, $port, $login, $password);
$this->queueName = $queueName;
$this->delayedQueueName = null;
$this->channel = $this->connection->channel();
// First, we need to make sure that RabbitMQ will never lose our queue.
// In order to do so, we need to declare it as durable. To do so we pass
// the third parameter to queue_declare as true.
$this->channel->queue_declare($queueName, false, true, false, false);
}
public function __destruct()
{
$this->close();
}
// Just in case : http://stackoverflow.com/questions/151660/can-i-trust-php-destruct-method-to-be-called
// We should call close explicitly if possible.
public function close()
{
if (!is_null($this->channel)) {
$this->channel->close();
$this->channel = null;
}
if (!is_null($this->connection)) {
$this->connection->close();
$this->connection = null;
}
}
public function produceWithDelay($data, $delay)
{
if (is_null($this->delayedQueueName))
{
$delayedQueueName = $this->queueName . '.delayed';
// First, we need to make sure that RabbitMQ will never lose our queue.
// In order to do so, we need to declare it as durable. To do so we pass
// the third parameter to queue_declare as true.
$this->channel->queue_declare($this->delayedQueueName, false, true, false, false, false,
new AMQPTable(array(
'x-dead-letter-exchange' => '',
'x-dead-letter-routing-key' => $this->queueName
))
);
$this->delayedQueueName = $delayedQueueName;
}
$msg = new AMQPMessage(
$data,
array(
'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT,
'expiration' => $delay
)
);
$this->channel->basic_publish($msg, '', $this->delayedQueueName);
}
public function produce($data)
{
$msg = new AMQPMessage(
$data,
array('delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT)
);
$this->channel->basic_publish($msg, '', $this->queueName);
}
public function consume($callback)
{
$this->callback = $callback;
// This tells RabbitMQ not to give more than one message to a worker at
// a time.
$this->channel->basic_qos(null, 1, null);
// Requires ack.
$this->channel->basic_consume($this->queueName, '', false, false, false, false, array($this, 'consumeCallback'));
while(count($this->channel->callbacks)) {
$this->channel->wait();
}
}
public function consumeCallback($msg)
{
call_user_func_array(
$this->callback,
array($msg)
);
// Very important to ack, in order to remove msg from queue. Ack after
// callback, as exception might happen in callback.
$msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
}
public function getQueueSize()
{
// three tuple containing (<queue name>, <message count>, <consumer count>)
$tuple = $this->channel->queue_declare($this->queueName, false, true, false, false);
if ($tuple != null && isset($tuple[1])) {
return $tuple[1];
}
return -1;
}
}
public function produce und public function consume Paar arbeitet wie erwartet.
Wenn es jedoch mit einem verzögerten Warteschlangensystem kommt
public function produceWithDelay und public function consume Paar funktioniert nicht wie erwartet. Der Verbraucher, der consume anruft, nicht in der Lage, irgendeinen Gegenstand zu empfangen, sogar auf einen Zeitraum wartend.
Ich glaube etwas nicht richtig mit meiner produceWithDelay Implementierung. Darf ich wissen, was los ist?
Versuchen Sie, Ihre Warteschlange als' $ Channel-> queue_declare ("name", falsch, falsch, falsch, wahr, wahr, array()) zu erklären, ] (https://gist.github.com/tairov/11289983) – Vardius
Es ist nicht notwendig, es von Grund auf neu zu implementieren. So sollten Sie es tun https://stackoverflow.com/a/45549182/579025 –