mrchrisadams
commented
5 years ago Hey @arendjantetteroo , can you let me know where queues are defined in this Symcony/Enqueue/AMQP combo?
Everything I try to set message priority so far seems to result in message priorities being ignored by RabbitMQ. Which means that doing any processing results in blocking incoming requests.
From what I can see you set queues, with priority like so, and you need to set it up when instantiating a queue. Here's a snippet from their docs
<?php
/** @var \Enqueue\AmqpExt\AmqpContext $context */
$fooQueue = $context->createQueue('foo');
$fooQueue->addFlag(AmqpQueue::FLAG_DURABLE);
$fooQueue->setArguments(['x-max-priority' => 10]);
$context->declareQueue($fooQueue);
$message = $context->createMessage('Hello world!');
$context->createProducer()
->setPriority(5) // the higher priority the sooner a message gets to a consumer
//
->send($fooQueue, $message)
;We use a ProducerInterface here in our DefaultController, where we delegate incoming requests to a the greencheck queue here:
private function doGreencheck($url, $ip, $browser, $source = 'api', $blind = false)
{
$message = new Message(JSON::encode(['key' => 0, 'url' => $url, 'ip' => $ip, 'browser' => $browser, 'source' => $source, 'blind' => $blind]));
// We still want incoming requests to take precedence over other jobs, so
$message->setPriority(MessagePriority::VERY_HIGH);
$promise = $this->producer->sendCommand('greencheck', $message, $needReply = true);
$replyMessage = $promise->receive();
$data = JSON::decode($replyMessage->getBody());
return $data['result'];
}Thing is our priority settings are not being respected, and I can't trace where the ProducerInterface is declaring the greencheck queue to see if we're passing in the necessary x-priority arguments needed.
In the logs I see this here:
./bin/console enqueue:consume --setup-broker -vvv
07:58:06 DEBUG [app] [AmqpDriver] Declare router exchange: enqueue.default
07:58:06 DEBUG [app] [AmqpDriver] Declare router queue: enqueue.app.default
07:58:06 DEBUG [app] [AmqpDriver] Bind router queue to exchange: enqueue.app.default -> enqueue.default
07:58:06 DEBUG [app] [AmqpDriver] Declare processor queueWhich makes me think we're using the AMQP driver which does not pass in x-priority info:
class AmqpDriver extends GenericDriver
{
// (snip )
/**
* @return AmqpQueue
*/
protected function doCreateQueue(string $transportQueueName): InteropQueue
{
/** @var AmqpQueue $queue */
$queue = parent::doCreateQueue($transportQueueName);
$queue->addFlag(AmqpQueue::FLAG_DURABLE);
return $queue;
}
}When I think we need to be using the RabbitMQ driver, which does:
class RabbitMqDriver extends AmqpDriver
{
/**
* @return AmqpQueue
*/
protected function doCreateQueue(string $transportQueueName): InteropQueue
{
$queue = parent::doCreateQueue($transportQueueName);
$queue->setArguments(['x-max-priority' => 4]);
return $queue;
}
}Seeing this behaviour yourself.
You can replicate this locally with this one liner if you have a dev environment running with Rabbit, Redis and MySQL:
./bin/console tgwf:greencheck:csvchecker && wget -O - http://localhost:8000/greencheck/google.comthis assumes there is a file called tgwf-check.csv, with a list of entries, of up to 10,000 in them (that's enough on my machine to simulate worker pool of 2 being blocked when an incoming web request comes in)
google.com
youtube.com
facebook.com
baidu.com
wikipedia.org
qq.com
yahoo.com
taobao.com
amazon.comResolving this
I can't see where we explicitly set up a rabbitmq driver, or name queues (otherwise, I'll experiment with making a differently named queue, and allocate some workers there).
Where should I be looking here? It's driving my crazy.
This transfers across the classes referred to in #9