Как организовать двойную очередь в Python?

Question

[Fox Loving]

Добрый день.

Изучаю Python, работу очередей. Смотрю в сторону Celery.
Суть в следующем. Есть задача организовать обработку входящих сообщений. Схематично все представлено ниже на Рисунке 1.
Нужно распределять входящие сообщения по Воркерам для обработки и дальнейшей передачи другой программе. Но есть ограничение -- в один момент времени для конкретного devID может выполняться только одна обработка. Т.е. если Воркер еще не обработал прошлое сообщение для какого-то devID, а уже пришло новое сообщение с данным devID, то оно должно ожидать завершения обработки прошлого сообщения. По сути получается двойная очередь: по devID и по Воркерам.

Не пойму, как это осуществить через Celery. Или может вообще не в ту сторону копаю?

Рисунок 1.

Answer 1

[Алексей Черемисин]

Две очереди, одна для всех сообщений, другая - для ожидающих.
Воркер:
1) получает сообщение из общей очереди
2) проверяет блокировку по devID
3) если блокировки нет:
3.1) ставит блокировку на devID
- делает работу
- освобождает блокировку на devID
3.2) блокировка есть:
- перепубликует сообщение в очередь ожидания, ТОЛЬКО если оно из основной очереди
4) проверяет очередь ожидания
- если там сообщение, то возвращается на п2
- иначе на п1

Как-то так... Ну и рассмотреть вопрос на счет дидлока, когда сообщение в очереди ожидания блокируется повисшим воркером...

Comments

[Руслан Роскошный]

Я бы поставил пункт 4 первым, это если в цикле.

[Алексей Черемисин]

Все относительно, это же почти конечный автомат, а там что последнее, что первое, все едино, но можно и первым пунктом. Да и думал всего минут 10, писал дольше, автору топика и проверять.

[Fox Loving]

Все, конечно, правильно, алгоритм то незамысловатый. Но вопрос был немного не о том.
Свой велосипед с блокировками и освобождениеми я всегда смогу изобрести. Но задача то на мой взгляд вполне заурядная и наверняка уже решена в каком-нибудь из имеющихся модулей Python. Вот и хотелось бы просто использовать уже имеющиеся наработки общества Python, и далее сосредоточится на специфичных для моей задачи вещах.
Я просматривал модуль Celery, но не нашел в нем то, что хотел. Может плохо искал?
Или Celery не совсем то, что мне нужно?

[Алексей Черемисин]

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

[Алексей Черемисин]

Ну и посмотрев на celery, там есть подзадания, может имеет смысл как-то вешать подзадания воркеру, который обрабатывает deviceID?

[Алексей Черемисин]

А интересный продукт, пойду читать...