Celery throttling — настраивам rate limit для очередей

[warner]

Спасибо, полезно. Я как-то делал через rate_limit но это работало только потому что я заранее знал сколько точно будет воркеров :) Здесь подход намного лучше.

[baldr]

Да, спасибо за статью. Я по заголовку сначала подумал что будет что-то простое вроде того же rate_limit, но подход, действительно, интересный.
Я, не зная о таком подходе, использовал бы какой-нибудь колхоз на Redis, но этот способ гораздо удобнее.

[glader]

Я правильно понял, что если задач долго нет, то токены скапливаются в очереди, и по приходу большой пачки задач они все будут запущены сразу, без лимитов (потому что токенов много)?

[30mb1]

С своих примерах я ограничиваю максимальное число токенов до 2, чтобы они не скапливались именно во избежание такого кейса. Вот кусок кода из статьи:

# 1 очередь под сами таски и 1 очередь под токены для них
app.conf.task_queues = [
    Queue('github'),
    # я ограничил длину очереди до 2ух, чтобы токены не скапливались
    # иначе это может привести к пробою нашего rate limit'a
    Queue('github_tokens', max_length=2)
]

Но, как я и сказал в конце, принцип скапливаемости может наоборот стать фишкой ;)

[kilokanat]

 ... Добавим эти строчки к main.py:

@app.task(bind=True) 
def get_github_api1(self, max_retries=None):     
    rate_limit(self, 'github')         
    print ('Called Api 1')

Опытным путем установлено, что max_retries=None пишется в декораторе. В противном случае тыщакрасныхстрочкав ошибки при достижения лимита retry.

Также опытным путем установлено, что в сelery версии 4+ и все ваше великолепие в Win10 почти нельзя...

С глубоким уважением прошу обозначить этот момент в статье.

[kilokanat]

+ один важный вещь. В итоговом причесанном один тонкий момент:

@app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
    # автоматически настроим выпуск токенов с нужной скоростью
    for name, limit in rate_limits.items():
        sender.add_periodic_task(
          60 / limit, 
          token.signature(queue=name+'_tokens'), 
          name=name + '_tokens')

Добавлено в создание периодической задачи "уникальное имя задачи" (name=name + '_tokens').

Без уникального имени для periodic_task, будет создаваться и наполняться токенами только одна!, последней указанная очередь, и фсё_ваще_не_работает.

Также при создании воркеров тоже нужны уникальные имена:

celery -A main worker -Q github -n worker1@%h
celery -A main worker -Q google -n worker2@%h