сельдерей, одна пока одна очередь и много параллельных во времени

Question

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

Есть ли способ указать, что загрузка всегдабудет один за другим.Только одна задача будет выполнена вовремя, а другие работники будут работать в других очередях?

Answer 1

Пока что я реализовал такое решение, работает довольно хорошо.Но я не совсем уверен, что max_retries = None утверждает, что количество повторных попыток будет неограниченным.Это решение работает на Redis, но может работать на любом другом движке, который поддерживает атомарную операцию приращения.

@task(max_retries=None,default_retry_delay=3)
def sleepTask():
    if r.incr('sleep_working')>1:
        r.incr('sleep_working',-1)
        sleepTask.retry()
    else:
        try:
            r.expire('sleep_working',3600)
            sleep(30)
        finally:
            r.incr('sleep_working',-1)
        return True

Ключевым моментом здесь является то, что incr атомарно, поэтому никогда не случится, что два клиента получат счетчик== 1.

Кроме того, истечение срока очень важно, все может произойти, и мы получим наш счетчик> 1 навсегда, поэтому expire гарантирует, что, несмотря ни на что, после определенного времени счетчик будет сброшен.Это значение может быть скорректировано с учетом потребностей.У меня загрузка больших файлов, так что 3600 звучит нормально.

Я думаю, что это хорошая отправная точка, чтобы создать собственный объект Task, который будет делать все это автоматически, получая значения redis_key и expire_time.Я буду обновлять этот пост, если я сделаю такую ​​задачу.

В качестве бонуса это решение также можно легко настроить на 2/3 / любое другое число параллельных лимитов, изменив> 1 на> любое число

Answer 2

Эффективным способом сериализации выполнения задач является использование мьютекса (Mutual Exclusion).

Модуль Python's threading имеет a Lock объект , который может быть привык к этому эффекту :

# ...
module_lock = threading.Lock() # or make this an attribute in an object with sufficiently-large scope
# ...
def do_interesting_task():
     with module_lock.acquire():
         interesting_task()

«Оставьте всякую надежду, входите сюда».

Мьютексы и семафоры являются мощными инструментами, но используютсянеразумно они будут приводить к тупикам и изредка кушать ваш обед.