RabbitMQ ест память

#rabbitmq

Запущен сервер на Python с использованием RabbitMQ. Через некоторое время (около
месяца) вот процесс съедает всю память и очереди останавливаются.
Вот этот процесс:

29431 rabbitmq  20   0 4330792 1,992g   4596 S  12,6 12,7 792:49.90 beam.smp 


После убиства процесса все становится хорошо. 

Судя по beam.smp, работает виртуальная машина erlang. Но почему она не освобождает
память - непонятно.
    

Ответы

Ответ 1

Разобрался с вопросом. Дело было в том, что раббит написан на erlang. В прекрасном
языке erlang сборщик мусора в виртуальной машине работает по старым и не особо эффективным
алгоритмам. Он начинает чистить память только когда она заканчивается. Однако есть
способы указать явно когда он запускается. 

Таким образом, через настройки RabbitMQ можно достучаться до виртуальной машины Erlang.
Для этого необходимо создать (если его нет) файл /etc/rabbitmq/rabbitmq.config (Debian,
Ubuntu). В нем можно указать процент от общей памяти или конкретное количество, сообщающее
когда следует запускать сборщик мусора. Следует помнить, что учитывается не оперативная
память, а виртуальная(!). Я сделал 20% от общей виртуальной памяти так:

[{rabbit, [{vm_memory_high_watermark, 0.2}]}].


Моя проблема заключалась в том, что значение по умолчанию 0.4, а это слишком много
для нагруженного сервера в production...

Убедиться в том, что предел потребляемой виртуальной памяти не превышает норму можно
командой sudo rabbitmqctl status. Нужный параметр: vm_memory_limit.

Вопрос по поводу очередей в RabbitMQ

#java #rabbitmq

Здравствуйте. Я пытаюсь написать очередь на rabbitmq.
На всех туториалах которых я смотрел, все работает следующим образом:


Есть код (отправитель), который посылает сообщение в очередь Rabbit'a и сразу завершается.
Есть код (получатель), который работает постоянно слушая очередь, чтобы забрать с
нее сообщение.
После того того как ОН получит сообщение из очереди, есть возможность дать сообщение
rabbit'у , что он его получил. И тогда rabbit удалит его из очереди. При этом сам получатель
не знает доставлено ли сообщение или нет (оно просто удалилось из очереди).


Но мне этого не достаточно. Мне нужно, чтобы отправитель узнал, что сообщение доставлено.

То есть, попытаюсь привести пример.

Отправитель кладет в очередь два сообщения:


"привет"
"пока"


Получатель получил сообщения, и отвечает :


"привет ок"
"пока не ок"


Подскажите, можно это реализовать как-то?
    

Ответы

Ответ 1

Вам нужно понять идеологию очередей, как они работают и в первую очередь -- "подтверждение
сообщений" (acknowledgement, ack) как описано в этой статье.

При выполнении задания - отправлять в другую очередь "задание id=123 выполнено" после
успешного выполнения задания. 

Плюс опционально: 


держать очередь для нотификаций об ошибках
периодически проверять "задания, которые зависли более чем на Х минут".


И ещё раз не могу не обратить ваше внимание на то, что у вас ДВЕ очереди сообщений:
одна в RabbitMQ, другая - в MSSQL. 

Синтетический пример. Допустим, у вас архив страничек в базе MSSQL. И тут же вы добавляете
табличку, в которой ставите задания для ботов, которые качают новые статьи. Скачал
- добавил новую запись в базу.

Вы сделали избыточную структуру. Можно было обойтись базой в MSSQL и хранить задания
ТОЛЬКО в RabbitMQ. И держать сто подписчиков, добавлять и удалять их, чтобы качать
быстрее. Обработчик взял задание, обработал, сохранил в базу, пошёл за следующим заданием.
Взял другое, ан видит - статья пять минут назад обновлялась, это меньше чем заданный
интервал свежести -- удалил из очереди.

Можно было бы не привлекать Rabbit MQ, оставить всё на базе MSSQL.

А у вас две очереди по факту. Такое ну бывает иногда приходится делать, если у вас
MSSQL - в одной организации, а Rabbit MQ - это сторонний сервис. Или разные отделы,
которые используют разные техологии (одни Linux, другие Windows), бывает и так. Но
поскольку я не знаю ваших особенностей -- то просто указываю на излишнее дублировани,
вероятно оно вам самому в глаза не бросается.


Ответ 2

Вот два варианта вам:


На отправителе поднимите слушатель(получатель) который будет слушать очередь B. Т.е.
вы отправляете в очередь A, её обрабатывает получатель A и после отработки шлет в очередь
B, которую слушает(обрабатывает) получатель B
Введите в сообщение какой-нибудь идентификатор. После обработки ложите данный идентификатор
какое нибудь хранилище(БД к примеру) со статусом обработки. Тем самым отправитель по
повторному запросу сможет получить данные из данного хранилища по идентификатору указанному
в сообщении

Таймер в RabbitMQ

#python #rabbitmq

Как в RabbitMQ на python сделать отправку сообщений в очередь так, чтобы оно пришло
в очередь только через определенное время, например, 4 часа ?
    

Ответы

Ответ 1

RabbitMQ не поддерживает такую "родную" функцию.

Предложение приведенное выше проблематично так как требует базы данных и постоянную
обработку сообщений. Такое решение намного более "рискованно". Не говоря уже о том
что просто дополнительное обрабатывающее средство + база данных ну просто ни к чему.

Но функцию возможно эмулировать. При этом сообщение будет отослано и будет находиться
на сервере до тех пор пока не наступит время сообщение доставить. Следующая версия
NServiceBus.RabbitMQ транспорта будет поддерживать Native Delays при помощи Topic Exchanges. 

Имплементация не зависит от языка посылающей аппликации.

Идея заключается в создании header exchanges где каждый уровень представляет бит
из задержки. Если бит 0, exchange направляет сообщение на следующий уровень. Если бит
1, то в queue с TTL соответствующий задержке на этом уровне. Каждый queue с DLQ binding
к следующему exchange.

Схематически, это будет выглядеть так:



Заметка: можно также имплементировать с Header Exchanges

С 32-ух битной задержкой (32 уровня exchanges) задержка получится от 1 до 2^31 секунд.


Ответ 2

Сообщение попадёт в очередь как только вы его туда отправите..

Что то вы намудрить хотите. Тогда уже добавляйте дату в сообщение и при чтении из
очереди если 4 часа не прошло - отправляйте клон задачи в конец очереди, текущую завершайте.

Или, пишите в какую-нибудь промежуточную таблицу в базе данных. Читайте её постоянно
(например, по крону). Когда выгребутся сообщения с нужной датой - - создавайте задачу
в очередь а строку из базы удаляйте:)


Ответ 3

Есть отличный плагин, который позволяет это реализовать - Scheduling Messages with
RabbitMQ.

Пример использования (Java)

Определим точку обмена:

// ...
Map args = new HashMap();
args.put("x-delayed-type", "direct");
channel.exchangeDeclare("my-exchange", "x-delayed-message", true, false, args);
// ...


Добавим пару сообщений:

byte[] messageBodyBytes = "delayed payload".getBytes("UTF-8");
Map headers = new HashMap();
headers.put("x-delay", 5000);
AMQP.BasicProperties.Builder props = new AMQP.BasicProperties.Builder().headers(headers);
channel.basicPublish("my-exchange", "", props.build(), messageBodyBytes);

byte[] messageBodyBytes2 = "more delayed payload".getBytes("UTF-8");
Map headers2 = new HashMap();
headers2.put("x-delay", 1000);
AMQP.BasicProperties.Builder props2 = new AMQP.BasicProperties.Builder().headers(headers2);
channel.basicPublish("my-exchange", "", props2.build(), messageBodyBytes2);


Задержка указывается в заголовке сообщения (x-delay) в секундах, соответственно.

Больше примеров и информации можно найти на странице самого проекта - Scheduling
Messages with RabbitMQ.

Как зайти на страницу RabbitMQ?

#java #python #rabbitmq

Установил RabbitMQ. Проверил - все работает. Сообщения отправляются и принимаются.
Но вот не могу зайти на страницу сервера RabbitMQ. Ввожу в браузер http://0.0.0.0:15672/
и ничего. Браузер говорит, что такой страницы не существует. ЧТо я делаю не так ? 
    

Ответы

Ответ 1

По умолчанию, веб-интерфейс отключен, но его можно включить командой на *nix системах:

rabbitmq-plugins enable rabbitmq_management


На Windows включается командой из папки sbin, где установлен сервер:

rabbitmq-plugins.bat enable rabbitmq_management 


После чего необходимо перезапустить RabbitMQ и войти через браузер на http://localhost:15672.
По умолчанию, логин и пароль для входа в RabbitMQ Managment Plugin: guest/guest

С версии 3.3 было введено ограничение на доступ по guest/guest с хоста отличного
от localhost. Но вы можете создать пользователя test c паролем test и зайти под ним
с любого хоста: 

rabbitmqctl add_user test test \n
rabbitmqctl set_user_tags test administrator 
rabbitmqctl set_permissions -p / test ".*" ".*" ".*"


Ответ 2

http://127.0.0.1:15672/


или

http://localhost:15672/


Если RabbitMQ стоит на другой машине, то соответственно нужно указать адрес машины

Сервер очередей под задачу

#клиент_сервер #очередь #rabbitmq #backgroundworker #gearman

Пытаюсь выбрать наиболее подходящий сервер очередей для такой задачи:
Есть алгоритм, который выполняется для определенного интервала, нужно разбить этот
интервал и вынести вычисления в воркеры (количество которых желательно задавать динамически
в зависимости от размера интервала), далее дождаться выполнения всех воркеров и выполнить
действия над вернувшимися значениями.

Я почитал, что для этого в принципе подходит gearman, но хотелось бы услышать еще
и ваши мнения. Также читал про популярный rabbitMq, но не нагуглил в нем возможности
задавать несколько воркеров.

Язык программирования, в принципе, не важен, главное чтоб многопоточность поддерживал
(многоядерную).
    

Ответы

Ответ 1

RabbitMQ позволит тебе подключить к одной очереди несколько consumer-ов (см. http://www.rabbitmq.com/tutorials/tutorial-two-python.html
Fair Dispatch). По поводу динамических консьюмеров - тут твое приложение само должно
решать когда их запускать и как их выключать. Можешь запустить кол-во консьюмеров равное
кол-ву ядер на машине и потом собирать данные.

А твоя задача напоминает реализацию MapReduce. Может тебе имеет смысл взглянуть на
Hadoop?

Если все же хочешь написать сам, то по поводу языка можно выбрать Erlang/OTP (на
нем написан RabbitMQ). Работа с потоками там построена очень хорошо и объединить несколько
машин в кластер, с шарингом потоков, не составит труда. Твое приложение не будет думать
о том, где и что выполняется, а просто ждать завершения всех воркеров и вернет результат.
Но сломать мозг придется, прежде чем написать функциональный код :)


Ответ 2

Из того, что уже советовали - RabbitMQ. Если нагрузки маленькие и вам необязательна
гарантируемость доставки, бурите его. В случае с Rabbit смиритесь с тем, что сообщения
могут потеряться, смиритесь с тем, что при больших нагрузках кластер начнёт отказывать,
смиритесь с тем, что при разделении сети будут проблемы.

Можно ещё взять Apache Kafka. Там с этим лучше, но его настраивать очень сложно,
чтоб он хорошо работал.

Для вашей же задачи подойдёт Hadoop или Spark.

Если хоститесь на амазоне, то с их лямбдами вроде как подобное можно реализовать.

PHP + Rabbitmq. Как получить количество сообщений в очереди?

#php #rabbitmq

Как из php скрипта определить количество сообщений в очереди? Вот мой код:

channel();
$channel->basic_consume('sms_inbox', '', true, false, false, false, $callback);

    

Ответы

Ответ 1

Это костыль для решения этого вопроса. Как сделать вывод количества сообщений в очереди
средствами php без функции exec() у меня так и не вышло

$content = exec('rabbitmqctl list_queues|grep sms_inbox');
preg_match_all('/sms_inbox(.+)/', $content, $result, PREG_PATTERN_ORDER);
$cnt = trim($result[1][0]);


Ответ 2

$queue = new AMQPQueue($channel);
$queue->setName('sms_inbox');
$queue->setFlags(AMQP_DURABLE);
echo $queue->declare() . PHP_EOL; // до 1.2.0


или с версии rabbitmq 1.2.0:

echo $queue->declareQueue() . PHP_EOL; // с 1.2.0


Метод declare() кроме того, что объявляет очередь, возвращает количество сообщений
в очереди.


  Return Values 
  
  Returns the message count.


С версии amqp 1.2.0 AMQPQueue::declare() deprecated и нужно использовать AMQPQueue::declareQueue().

RabbitMQ несколько очередей на одного потребителя

#java #rabbitmq

В мою систему отправляется больше количество сообщений из разных систем. Допустим,
сообщения для автомобилей разных марок.

При каждой отправке новой марки автомобилей из какой-либо системы создается новая
очередь (например, auto.kia, auto.lada и т.п.) в итоге в реббите может быть произвольное
количество очередей.

Но, у меня есть один воркер (Consumer), который должен собирать и обрабатывать сообщения
из всех этих очередей. Как реализовать множества очередей одним воркером?
    

Ответы

Ответ 1

Звучит как классический случай topic exchange.

В них требуется, чтобы routing_key в сообщениях отправителя (producer) состоял из
набора слов, разделённых точками, ваши auto.lada и auto.kia как раз подходят.

Тогда binding_key принимающего (consumer), содержащий спецсимволы * и # играет роль
шаблона вроде регулярного выражения:


* означает "одно любое слово"
# означает "любое число (даже 0) любых слов".


...и ключи auto.lada и auto.kia оба подойдут под шаблон auto.* у принимающего.

Переменной при таком раскладе будет не очередь, а ключ сообщений (routing key для
отправителя). Распределение происходит в рамках одной очереди. Много очередей для этой
цели вам и не нужно. Подробнее в руководстве по AMQP.

Получить список Queue и Exchanges в RabbitMQ

Можно ли получить как-то список существующих очередей? Хочу в дальнейшем использовать их в c#

Ответ

Вот тут упоминают HTTP API - попробовал на своей машине, работает. Читайте JSON, парсите при помощи Json.Net.
http://localhost:15672/api/exchanges http://localhost:15672/api/queues

Сброс настроек Zend_Controller_Front

Здравствуйте! Переписываю проект на SOA и перевожу public/index.php в демона, который умеет принимать сообщения из RabbitMQ, обрабатывать их и отправлять дальше по цепочке в кролика. Столкнулся с проблемой, когда хочу сделать запрос на свой контроллер и экшен, то происходит только 1 вызов, а потом всё зацикливается и не получается установить новый контроллер/экшен... Код коллбека, который дергается при приходе сообщения из кролика: /** @param $msg PhpAmqpLib\Message\AMQPMessage */ $callback = function ($msg) { $dataIn = json_decode($msg->body); $bodyStr = $body = '';
$auth = Zend_Auth::getInstance();
$front = Zend_Controller_Front::getInstance();
if (isset($dataIn->user_id) && !empty($dataIn->user_id)) { $storage = $auth->getStorage(); $data = new stdClass(); $data->user_id = $dataIn->user_id; $storage->write((array)$data);
$request = $front->getRequest(); $request->setParams((array)$dataIn->param); $request->setRequestUri($dataIn->url); $response = $front->getResponse();
$front->dispatch($request, $response);
print_r(array( "getActionKey" => $request->getActionKey(), "getActionName" => $request->getActionName(), "getControllerKey" => $request->getControllerKey(), "getControllerName" => $request->getControllerName(), ));
$body = json_decode($response->getBody()); } $front->clearParams();
$body->client_id = $dataIn->client_id; $bodyStr = json_encode($body);
/** @var $msgChannel PhpAmqpLib\Channel\AMQPChannel */ $msgChannel = $msg->delivery_info['channel']; $msgChannel->basic_ack($msg->delivery_info['delivery_tag']);
//echo "
[x] Обработано " . $msg->delivery_info['routing_key'] . ' : ' . $bodyStr . "
"; $sender = new SenderSysEventToOutox(); $sender->send($bodyStr, $dataIn->client_id); $auth->getStorage()->clear(); }; В сообщении содержится информация о пользователе, который совершает действие $dataIn->user_id В демоне мы устанавливаем ему "авторизацию" на время выполнения запроса, после сбрасываем. Получаем инстанс фронт контроллера $front = Zend_Controller_Front::getInstance(); и после установки параметров делаем вызов на контроллер/экшен, который пришел в сообщении. $request->setParams((array)$dataIn->param); $request->setRequestUri($dataIn->url); $response = $front->getResponse(); $front->dispatch($request, $response); И в конце очищаем все установленные ранее параметры $front->clearParams(); После прихода нового сообщения, по идее, должен происходить все как с чистого листа, так как параметры очищены, но беда бедовая в том, что оно так не происходит, я уже что только не пробовал, и такие комбинации: $front->getResponse()->clearAllHeaders()->clearRawHeaders()->clearBody(); И делал $front->resetInstance(); // тут вообще муть начинается, так как он сбрасывает очень много всего и заново воспользоваться получение инстанса пока не получается. В общем вот... такая лабуда, есть ли у кого мысли, как сбросить все настройки Фронт-контроллера, чтобы при каждом новом вызове все было как будто "в первый раз"? Заранее спасибо.

Ответ

$front = Zend_Controller_Front::getInstance(); $clonedFront = clone $front; работаем с клоном. По окончании убиваем клона. UPD: clone не будет работать, т.к. singleton.. сорри. Попытка номер 2: Вручную создавать обьект Zend_Controller_Request_Http. $request = new Zend_Controller_Request_Http(); //Здесь он сам подхватит новый URI, либо в параметрах указать его Zend_Controller_Front::getInstance()->setRequest($request); ...

RabbitMQ ест память

Запущен сервер на Python с использованием RabbitMQ. Через некоторое время (около месяца) вот процесс съедает всю память и очереди останавливаются. Вот этот процесс:
29431 rabbitmq 20 0 4330792 1,992g 4596 S 12,6 12,7 792:49.90 beam.smp
После убиства процесса все становится хорошо.
Судя по beam.smp, работает виртуальная машина erlang. Но почему она не освобождает память - непонятно.

Ответ

Разобрался с вопросом. Дело было в том, что раббит написан на erlang. В прекрасном языке erlang сборщик мусора в виртуальной машине работает по старым и не особо эффективным алгоритмам. Он начинает чистить память только когда она заканчивается. Однако есть способы указать явно когда он запускается.
Таким образом, через настройки RabbitMQ можно достучаться до виртуальной машины Erlang. Для этого необходимо создать (если его нет) файл /etc/rabbitmq/rabbitmq.config (Debian, Ubuntu). В нем можно указать процент от общей памяти или конкретное количество, сообщающее когда следует запускать сборщик мусора. Следует помнить, что учитывается не оперативная память, а виртуальная(!). Я сделал 20% от общей виртуальной памяти так:
[{rabbit, [{vm_memory_high_watermark, 0.2}]}].
Моя проблема заключалась в том, что значение по умолчанию 0.4, а это слишком много для нагруженного сервера в production...
Убедиться в том, что предел потребляемой виртуальной памяти не превышает норму можно командой sudo rabbitmqctl status. Нужный параметр: vm_memory_limit

Вопрос по поводу очередей в RabbitMQ

Здравствуйте. Я пытаюсь написать очередь на rabbitmq. На всех туториалах которых я смотрел, все работает следующим образом:
Есть код (отправитель), который посылает сообщение в очередь Rabbit'a и сразу завершается. Есть код (получатель), который работает постоянно слушая очередь, чтобы забрать с нее сообщение. После того того как ОН получит сообщение из очереди, есть возможность дать сообщение rabbit'у , что он его получил. И тогда rabbit удалит его из очереди. При этом сам получатель не знает доставлено ли сообщение или нет (оно просто удалилось из очереди).
Но мне этого не достаточно. Мне нужно, чтобы отправитель узнал, что сообщение доставлено.
То есть, попытаюсь привести пример.
Отправитель кладет в очередь два сообщения:
"привет" "пока"
Получатель получил сообщения, и отвечает :
"привет ок" "пока не ок"
Подскажите, можно это реализовать как-то?

Ответ

Вам нужно понять идеологию очередей, как они работают и в первую очередь -- "подтверждение сообщений" (acknowledgement, ack) как описано в этой статье
При выполнении задания - отправлять в другую очередь "задание id=123 выполнено" после успешного выполнения задания.
Плюс опционально:
держать очередь для нотификаций об ошибках периодически проверять "задания, которые зависли более чем на Х минут".
И ещё раз не могу не обратить ваше внимание на то, что у вас ДВЕ очереди сообщений: одна в RabbitMQ, другая - в MSSQL.
Синтетический пример. Допустим, у вас архив страничек в базе MSSQL. И тут же вы добавляете табличку, в которой ставите задания для ботов, которые качают новые статьи. Скачал - добавил новую запись в базу.
Вы сделали избыточную структуру. Можно было обойтись базой в MSSQL и хранить задания ТОЛЬКО в RabbitMQ. И держать сто подписчиков, добавлять и удалять их, чтобы качать быстрее. Обработчик взял задание, обработал, сохранил в базу, пошёл за следующим заданием. Взял другое, ан видит - статья пять минут назад обновлялась, это меньше чем заданный интервал свежести -- удалил из очереди.
Можно было бы не привлекать Rabbit MQ, оставить всё на базе MSSQL.
А у вас две очереди по факту. Такое ну бывает иногда приходится делать, если у вас MSSQL - в одной организации, а Rabbit MQ - это сторонний сервис. Или разные отделы, которые используют разные техологии (одни Linux, другие Windows), бывает и так. Но поскольку я не знаю ваших особенностей -- то просто указываю на излишнее дублировани, вероятно оно вам самому в глаза не бросается.

Таймер в RabbitMQ

Как в RabbitMQ на python сделать отправку сообщений в очередь так, чтобы оно пришло в очередь только через определенное время, например, 4 часа ?

Ответ

RabbitMQ не поддерживает такую "родную" функцию.
Предложение приведенное выше проблематично так как требует базы данных и постоянную обработку сообщений. Такое решение намного более "рискованно". Не говоря уже о том что просто дополнительное обрабатывающее средство + база данных ну просто ни к чему.
Но функцию возможно эмулировать. При этом сообщение будет отослано и будет находиться на сервере до тех пор пока не наступит время сообщение доставить. Следующая версия NServiceBus.RabbitMQ транспорта будет поддерживать Native Delays при помощи Topic Exchanges.
Имплементация не зависит от языка посылающей аппликации.
Идея заключается в создании header exchanges где каждый уровень представляет бит из задержки. Если бит 0, exchange направляет сообщение на следующий уровень. Если бит 1, то в queue с TTL соответствующий задержке на этом уровне. Каждый queue с DLQ binding к следующему exchange.
Схематически, это будет выглядеть так:

Заметка: можно также имплементировать с Header Exchanges
С 32-ух битной задержкой (32 уровня exchanges) задержка получится от 1 до 2^31 секунд.

Как зайти на страницу RabbitMQ?

Установил RabbitMQ. Проверил - все работает. Сообщения отправляются и принимаются. Но вот не могу зайти на страницу сервера RabbitMQ. Ввожу в браузер http://0.0.0.0:15672/ и ничего. Браузер говорит, что такой страницы не существует. ЧТо я делаю не так ?

Ответ

По умолчанию, веб-интерфейс отключен, но его можно включить командой на *nix системах:
rabbitmq-plugins enable rabbitmq_management
На Windows включается командой из папки sbin, где установлен сервер:
rabbitmq-plugins.bat enable rabbitmq_management
После чего необходимо перезапустить RabbitMQ и войти через браузер на http://localhost:15672. По умолчанию, логин и пароль для входа в RabbitMQ Managment Plugin: guest/guest
С версии 3.3 было введено ограничение на доступ по guest/guest с хоста отличного от localhost. Но вы можете создать пользователя test c паролем test и зайти под ним с любого хоста:
rabbitmqctl add_user test test
rabbitmqctl set_user_tags test administrator rabbitmqctl set_permissions -p / test ".*" ".*" ".*"

Сервер очередей под задачу

Пытаюсь выбрать наиболее подходящий сервер очередей для такой задачи: Есть алгоритм, который выполняется для определенного интервала, нужно разбить этот интервал и вынести вычисления в воркеры (количество которых желательно задавать динамически в зависимости от размера интервала), далее дождаться выполнения всех воркеров и выполнить действия над вернувшимися значениями.
Я почитал, что для этого в принципе подходит gearman, но хотелось бы услышать еще и ваши мнения. Также читал про популярный rabbitMq, но не нагуглил в нем возможности задавать несколько воркеров
Язык программирования, в принципе, не важен, главное чтоб многопоточность поддерживал (многоядерную).

Ответ

RabbitMQ позволит тебе подключить к одной очереди несколько consumer-ов (см. http://www.rabbitmq.com/tutorials/tutorial-two-python.html Fair Dispatch). По поводу динамических консьюмеров - тут твое приложение само должно решать когда их запускать и как их выключать. Можешь запустить кол-во консьюмеров равное кол-ву ядер на машине и потом собирать данные.
А твоя задача напоминает реализацию MapReduce. Может тебе имеет смысл взглянуть на Hadoop?
Если все же хочешь написать сам, то по поводу языка можно выбрать Erlang/OTP (на нем написан RabbitMQ). Работа с потоками там построена очень хорошо и объединить несколько машин в кластер, с шарингом потоков, не составит труда. Твое приложение не будет думать о том, где и что выполняется, а просто ждать завершения всех воркеров и вернет результат. Но сломать мозг придется, прежде чем написать функциональный код :)

RabbitMQ несколько очередей на одного потребителя

В мою систему отправляется больше количество сообщений из разных систем. Допустим, сообщения для автомобилей разных марок.
При каждой отправке новой марки автомобилей из какой-либо системы создается новая очередь (например, auto.kia, auto.lada и т.п.) в итоге в реббите может быть произвольное количество очередей.
Но, у меня есть один воркер (Consumer), который должен собирать и обрабатывать сообщения из всех этих очередей. Как реализовать множества очередей одним воркером?

Ответ

Звучит как классический случай topic exchange
В них требуется, чтобы routing_key в сообщениях отправителя (producer) состоял из набора слов, разделённых точками, ваши auto.lada и auto.kia как раз подходят.
Тогда binding_key принимающего (consumer), содержащий спецсимволы * и # играет роль шаблона вроде регулярного выражения:
* означает "одно любое слово" # означает "любое число (даже 0) любых слов".
...и ключи auto.lada и auto.kia оба подойдут под шаблон auto.* у принимающего.
Переменной при таком раскладе будет не очередь, а ключ сообщений (routing key для отправителя). Распределение происходит в рамках одной очереди. Много очередей для этой цели вам и не нужно. Подробнее в руководстве по AMQP