Часто задаваемые вопросы¶

Для каких целей следует использовать сельдерей?¶

Ответ: Queue everything and delight everyone - хорошая статья, описывающая, зачем использовать очередь в веб-контексте.

Вот некоторые распространенные случаи использования:

• Выполнение чего-либо в фоновом режиме. Например, завершить веб-запрос как можно быстрее, а затем инкрементально обновить страницу пользователя. Это создает у пользователя впечатление хорошей производительности и «быстродействия», хотя на самом деле работа может занять некоторое время.

• Выполнение чего-либо после завершения веб-запроса.

• Убедитесь, что что-то сделано, выполнив это асинхронно и используя повторные попытки.

• Планирование периодических работ.

И в какой-то степени:

• Распределенные вычисления.

• Параллельное выполнение.

Действительно ли Celery состоит из 50.000 строк кода?¶

Ответ: Нет, об этом и подобных больших количествах сообщалось в различных местах.

На момент написания статьи цифры таковы:

• ядро: 7 141 строка кода.

• тесты: 14 209 строк.

• backends, contrib, compat утилиты: 9 032 строки.

Количество строк кода не является полезной метрикой, поэтому даже если бы Celery состоял из 50 тысяч строк кода, вы не смогли бы сделать никаких выводов на основе такого числа.

Много ли зависимостей у Celery?¶

Распространенная критика заключается в том, что Celery использует слишком много зависимостей. Обоснование такого опасения трудно представить, особенно учитывая повторное использование кода как устоявшийся способ борьбы со сложностью в современной разработке программного обеспечения, а также то, что стоимость добавления зависимостей очень низка сейчас, когда менеджеры пакетов, такие как pip и PyPI, делают установку и поддержку зависимостей делом прошлого.

В процессе работы Celery заменил несколько зависимостей, и в настоящее время список зависимостей выглядит следующим образом:

Сельдерей имеет большой вес?¶

Celery создает очень незначительные накладные расходы как по занимаемой памяти, так и по производительности.

Но обратите внимание, что конфигурация по умолчанию не оптимизирована ни по времени, ни по пространству, для получения дополнительной информации смотрите руководство Оптимизация.

Зависит ли сельдерей от огурца?¶

Ответ: Нет, Celery может поддерживать любую схему сериализации.

У нас есть встроенная поддержка JSON, YAML, Pickle и msgpack. Каждая задача связана с типом содержимого, поэтому вы можете даже отправить одну задачу с помощью pickle, а другую - с помощью JSON.

Раньше по умолчанию поддерживалась сериализация pickle, но начиная с версии 4.0 по умолчанию используется JSON. Если вам требуется отправлять сложные объекты Python в качестве аргументов задачи, вы можете использовать pickle в качестве формата сериализации, но смотрите примечания в Сериализаторы.

Если вам нужно взаимодействовать с другими языками, вы должны использовать формат сериализации, подходящий для этой задачи, что в значительной степени означает любой сериализатор, который не является pickle.

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

Является ли Celery только для Django?¶

Ответ: Нет, вы можете использовать Celery с любым фреймворком, веб или другим.

Обязательно ли использовать AMQP/RabbitMQ?¶

Ответ: Нет, хотя рекомендуется использовать RabbitMQ, вы также можете использовать Redis, SQS или Qpid.

Дополнительную информацию см. в разделе Бэкенды и брокеры.

Redis в качестве брокера не будет работать так же хорошо, как AMQP-брокер, но комбинация RabbitMQ в качестве брокера и Redis в качестве хранилища результатов широко используется. Если у вас есть строгие требования к надежности, вам рекомендуется использовать RabbitMQ или другой AMQP-брокер. Некоторые транспорты также используют опрос, поэтому они, скорее всего, будут потреблять больше ресурсов. Однако, если вы по каким-то причинам не можете использовать AMQP, смело используйте эти альтернативы. Они, вероятно, будут хорошо работать для большинства случаев использования, и обратите внимание, что приведенные выше пункты не являются специфическими для Celery; если использование Redis/базы данных в качестве очереди хорошо работало для вас раньше, то, вероятно, оно будет работать и сейчас. Вы всегда сможете обновить систему позже, если потребуется.

Является ли Celery многоязычным?¶

Ответ: Да.

worker - это реализация Celery в Python. Если язык имеет AMQP-клиент, то не должно быть много работы по созданию рабочего в вашем языке. Рабочий Celery - это просто программа, подключающаяся к брокеру для обработки сообщений.

Кроме того, есть еще один способ быть независимым от языка, и это использование задач REST, вместо того, чтобы ваши задачи были функциями, они являются URL. С помощью этой информации вы можете даже создавать простые веб-серверы, которые позволяют предварительно загружать код. Просто предоставьте конечную точку, которая выполняет операцию, и создайте задачу, которая просто выполняет HTTP-запрос к этой конечной точке.

Вы также можете использовать Flower’s REST API для вызова задач.

MySQL выдает ошибку deadlock, что я могу сделать?¶

Ответ: В MySQL уровень изоляции по умолчанию установлен на REPEATABLE-READ, если вам это не нужно, установите его на READ-COMMITTED. Вы можете сделать это, добавив следующее в my.cnf:

[mysqld]
transaction-isolation = READ-COMMITTED

Более подробная информация о InnoDB<<<0 >>> в руководстве пользователя MySQL.

(Спасибо Гонзе Кралю и Антону Цигулярову за это решение)

Рабочий ничего не делает, просто висит.¶

Ответ: См. MySQL is throwing deadlock errors, what can I do?, или Why is Task.delay/apply*/the worker just hanging?.

Результаты заданий возвращаются ненадежно¶

Ответ: Если вы используете бэкэнд базы данных для получения результатов, и в частности используете MySQL, смотрите MySQL is throwing deadlock errors, what can I do?.

Почему Task.delay/apply*/the worker просто висит?¶

Ответ: В некоторых AMQP-клиентах есть ошибка, которая приводит к зависанию, если не удается аутентифицировать текущего пользователя, пароль не подходит или у пользователя нет доступа к указанному виртуальному хосту. Обязательно проверьте логи вашего брокера (для RabbitMQ это /var/log/rabbitmq/rabbit.log на большинстве систем), там обычно содержится сообщение, описывающее причину.

Работает ли он на FreeBSD?¶

Ответ: Зависит;

При использовании транспорта RabbitMQ (AMQP) и Redis он должен работать из коробки.

Для других транспортов используется пул prefork совместимости и требуется работающая реализация POSIX семафоров, которая включена во FreeBSD по умолчанию начиная с FreeBSD 8.x. Для более старых версий FreeBSD необходимо включить POSIX семафоры в ядре и вручную перекомпилировать billiard.

К счастью, Виктор Петерссон написал учебник, который поможет вам начать работу с Celery на FreeBSD здесь: http://www.playingwithwire.com/2009/10/how-to-get-celeryd-to-work-on-freebsd/.

У меня возникают IntegrityError: Duplicate Key ошибки. Почему?¶

Ответ: См. MySQL is throwing deadlock errors, what can I do?. Спасибо @@howsthedotcom.

Почему мои задания не обрабатываются?¶

Ответ: В RabbitMQ вы можете узнать, сколько потребителей в настоящее время получают задания, выполнив следующую команду:

$ rabbitmqctl list_queues -p <myvhost> name messages consumers
Listing queues ...
celery     2891    2

Это показывает, что в очереди задач ожидает обработки 2891 сообщение, и их обрабатывают два потребителя.

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

Когда сообщение получено рабочим, брокер ожидает подтверждения, прежде чем пометить сообщение как обработанное. Брокер не будет повторно отправлять это сообщение другому потребителю, пока потребитель не будет закрыт должным образом.

Если вы столкнулись с этой проблемой, вам придется вручную убить всех рабочих и перезапустить их:

$ pkill 'celery worker'

$ # - If you don't have pkill use:
$ # ps auxww | awk '/celery worker/ {print $2}' | xargs kill

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

$ pkill -9 'celery worker'

$ # - If you don't have pkill use:
$ # ps auxww | awk '/celery worker/ {print $2}' | xargs kill -9

Почему моя задача не выполняется?¶

Ответ: Возможно, существуют синтаксические ошибки, препятствующие импорту модуля задач.

Вы можете узнать, может ли Celery выполнить задание, выполнив его вручную:

>>> from myapp.tasks import MyPeriodicTask
>>> MyPeriodicTask.delay()

Просмотрите файл журнала рабочих, чтобы узнать, удалось ли ему найти задание, или произошла какая-то другая ошибка.

Почему моя периодическая задача не выполняется?¶

Ответ: См. Why won’t my Task run?.

Как очистить все ожидающие задания?¶

Ответ: Вы можете использовать команду celery purge для очистки всех настроенных очередей задач:

$ celery -A proj purge

или программно:

>>> from proj.celery import app
>>> app.control.purge()
1753

Если вы хотите очистить сообщения только из определенной очереди, вам нужно использовать AMQP API или утилиту celery amqp:

$ celery -A proj amqp queue.purge <queue name>

Число 1753 - это количество удаленных сообщений.

Вы также можете запустить рабочий с включенной опцией --purge, чтобы очистить сообщения при запуске рабочего.

Я очистил сообщения, но в очереди все еще остались сообщения?¶

Ответ: Задания квитируются (удаляются из очереди), как только они фактически выполнены. После того, как рабочий получил задание, пройдет некоторое время, пока оно будет действительно выполнено, особенно если много заданий уже ожидают выполнения. Сообщения, которые не были подтверждены, сохраняются рабочим до тех пор, пока он не закроет соединение с брокером (сервером AMQP). Когда это соединение закрывается (например, потому что рабочий был остановлен), задания будут повторно отправлены брокером следующему доступному рабочему (или тому же рабочему, если он был перезапущен), поэтому для правильной очистки очереди от ожидающих заданий необходимо остановить всех рабочих, а затем очистить задания с помощью celery.control.purge().

Как получить результат задачи, если у меня есть ID, указывающий на нее?¶

Ответ: Используйте task.AsyncResult:

>>> result = my_task.AsyncResult(task_id)
>>> result.get()

Это даст вам экземпляр AsyncResult, использующий бэкенд текущего результата задачи.

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

>>> result = app.AsyncResult(task_id)
>>> result.get()

Разве использование pickle не является проблемой безопасности?¶

Ответ: Действительно, начиная с Celery 4.0 сериализатором по умолчанию теперь является JSON, чтобы убедиться, что люди выбирают сериализаторы осознанно и знают об этой проблеме.

Необходимо защититься от несанкционированного доступа к брокеру, базам данных и другим сервисам, передающим травленые данные.

Обратите внимание, что это не только то, что вы должны знать о Celery, например, Django также использует pickle для своего клиента кэша.

Для сообщений задачи вы можете установить параметр task_serializer на «json» или «yaml» вместо pickle.

Аналогично для результатов задания можно установить result_serializer.

Более подробно об используемых форматах и порядке поиска при проверке того, какой формат использовать для задания, смотрите Сериализаторы.

Можно ли шифровать сообщения?¶

Ответ: Некоторые AMQP-брокеры поддерживают использование SSL (в том числе RabbitMQ). Вы можете включить его с помощью параметра broker_use_ssl.

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

Безопасно ли запускать celery worker от имени root?¶

Ответ: Нет!

В настоящее время мы не знаем о каких-либо проблемах безопасности, но было бы невероятно наивно полагать, что их не существует, поэтому рекомендуется запускать службы Celery (celery worker, celery beat, celeryev и т.д.) от имени непривилегированного пользователя.

Почему RabbitMQ дает сбой?¶

Ответ: RabbitMQ аварийно завершает работу, если у него заканчивается память. Это будет исправлено в будущем выпуске RabbitMQ. пожалуйста, обратитесь к FAQ по RabbitMQ: https://www.rabbitmq.com/faq.html#node-runs-out-of-memory.

Неправильная конфигурация Celery может привести к сбою на старых версиях RabbitMQ. Даже если это не приведет к сбою, это все равно может потребовать много ресурсов, поэтому важно, чтобы вы знали о распространенных подводных камнях.

• События.

Выполнение worker с опцией -E будет отправлять сообщения о событиях, происходящих внутри рабочего.

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

• Результаты работы бэкенда AMQP.

При работе с бэкендом результатов AMQP каждый результат задачи будет отправлен в виде сообщения. Если вы не будете собирать эти результаты, они будут накапливаться, и RabbitMQ в конце концов исчерпает память.

Этот бэкенд результатов теперь устарел, поэтому вам не следует его использовать. Используйте либо бэкенд RPC для вызовов в стиле rpc, либо постоянный бэкенд, если вам нужен многопотребительский доступ к результатам.

По умолчанию результаты истекают через 1 день. Возможно, будет полезно снизить это значение, настроив параметр result_expires.

Если вы не используете результаты для задания, убедитесь, что вы установили опцию ignore_result:

@app.task(ignore_result=True)
def mytask():
    pass

class MyTask(Task):
    ignore_result = True

Могу ли я использовать Celery с ActiveMQ/STOMP?¶

Ответ: Нет. Раньше он поддерживался в Carrot (наша старая библиотека обмена сообщениями), но в настоящее время не поддерживается в Kombu (наша новая библиотека обмена сообщениями).

Какие функции не поддерживаются, если не используется брокер AMQP?¶

Это неполный список функций, недоступных при использовании виртуальных транспортов:

• Команды удаленного управления (поддерживаются только Redis).

• Мониторинг с помощью событий может работать не во всех виртуальных транспортах.

• Типы обмена header и fanout

  (fanout поддерживается Redis).

Как можно повторно использовать одно и то же соединение при вызове задач?¶

Ответ: См. параметр broker_pool_limit. Пул соединений включен по умолчанию начиная с версии 2.5.

sudo в subprocess возвращает None¶

Существует опция конфигурации sudo, которая делает незаконным выполнение sudo для процесса без tty:

Defaults requiretty

Если у вас есть такая конфигурация в файле /etc/sudoers, то задачи не смогут вызывать sudo, когда worker работает как демон. Если вы хотите разрешить это, то вам нужно удалить строку из /etc/sudoers.

См.: http://timelordz.com/wiki/Apache_Sudo_Commands

Почему работники удаляют задания из очереди, если они не могут их обработать?¶

Ответ:

Рабочий отклоняет неизвестные задания, сообщения с ошибками кодировки и сообщения, не содержащие нужных полей (в соответствии с протоколом сообщений заданий).

Если он не отклонял их, они могли доставляться снова и снова, вызывая зацикливание.

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

Могу ли я назвать задачу по имени?¶

Ответ: Да, используйте app.send_task().

Вы также можете вызвать задачу по имени, из любого языка, используя AMQP-клиент:

>>> app.send_task('tasks.add', args=[2, 2], kwargs={})
<AsyncResult: 373550e8-b9a0-4666-bc61-ace01fa4f91d>

Чтобы использовать chain, chord или group с задачами, вызываемыми по имени, используйте метод Celery.signature():

>>> chain(
...     app.signature('tasks.add', args=[2, 2], kwargs={}),
...     app.signature('tasks.add', args=[1, 1], kwargs={})
... ).apply_async()
<AsyncResult: e9d52312-c161-46f0-9013-2713e6df812d>

Могу ли я получить идентификатор текущей задачи?¶

Ответ: Да, текущий идентификатор и многое другое доступно в запросе задачи:

@app.task(bind=True)
def mytask(self):
    cache.set(self.request.id, "Running")

Для получения дополнительной информации см. раздел Запрос задания.

Если у вас нет ссылки на экземпляр задачи, вы можете использовать app.current_task:

>>> app.current_task.request.id

Но обратите внимание, что это будет любая задача, будь то задача, выполняемая рабочим, или задача, вызываемая непосредственно этой задачей, или задача, вызываемая с нетерпением.

Чтобы конкретно получить текущую задачу, над которой ведется работа, используйте current_worker_task:

>>> app.current_worker_task.request.id

Могу ли я указать пользовательский идентификатор задачи?¶

Ответ: Да, используйте аргумент task_id для Task.apply_async():

>>> task.apply_async(args, kwargs, task_id='…')

Можно ли использовать декораторы с задачами?¶

Ответ: Да, но см. примечание в боковой панели по адресу Основы.

Можно ли использовать естественные идентификаторы задач?¶

Ответ: Да, но убедитесь, что он уникален, так как поведение двух задач с одинаковым id не определено.

Возможно, мир не взорвется, но они определенно могут переписать результаты друг друга.

Можно ли запустить задачу после завершения другой задачи?¶

Ответ: Да, вы можете безопасно запускать задачу внутри задачи.

Общим шаблоном является добавление обратных вызовов к задачам:

from celery.utils.log import get_task_logger

logger = get_task_logger(__name__)

@app.task
def add(x, y):
    return x + y

@app.task(ignore_result=True)
def log_result(result):
    logger.info("log_result got: %r", result)

Приглашение:

>>> (add.s(2, 2) | log_result.s()).delay()

Дополнительную информацию см. в разделе Холст: Проектирование рабочих потоков.

Можно ли отменить выполнение задания?¶

Ответ: Да, используйте result.revoke():

>>> result = add.apply_async(args=[2, 2], countdown=120)
>>> result.revoke()

или если у вас есть только идентификатор задачи:

>>> from proj.celery import app
>>> app.control.revoke(task_id)

Последние также поддерживают передачу списка идентификаторов задач в качестве аргумента.

Почему мои команды дистанционного управления не принимаются всеми работниками?¶

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

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

Чтобы обойти это, вы можете явно установить имя узла для каждого рабочего, используя аргумент -n на worker:

$ celery -A proj worker -n worker1@%h
$ celery -A proj worker -n worker2@%h

где %h расширяется до текущего имени хоста.

Можно ли отправлять некоторые задания только на некоторые серверы?¶

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

Дополнительную информацию см. в разделе Задачи маршрутизации.

Можно ли отключить предварительную выборку заданий?¶

Ответ: Возможно! Термин AMQP «prefetch» сбивает с толку, поскольку он используется только для описания лимита предварительной выборки задач. На самом деле никакой предварительной выборки не происходит.

Отключение лимитов предварительной выборки возможно, но это означает, что рабочий будет потреблять столько заданий, сколько сможет, настолько быстро, насколько это возможно.

Обсуждение пределов предварительной выборки и параметров конфигурации для рабочего, который резервирует только одну задачу за раз, можно найти здесь: Пределы предварительной выборки.

Можно ли изменить интервал периодической задачи во время выполнения?¶

Ответ: Да, вы можете использовать планировщик базы данных Django, или вы можете создать новый подкласс расписания и переопределить is_due():

from celery.schedules import schedule

class my_schedule(schedule):

    def is_due(self, last_run_at):
        return run_now, next_time_to_check

Поддерживает ли Celery приоритеты задач?¶

Ответ: Да, RabbitMQ поддерживает приоритеты начиная с версии 3.5.0, а транспорт Redis эмулирует поддержку приоритетов.

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

Следует ли мне использовать retry или acks_late?¶

Ответ: Зависит от ситуации. Не обязательно одно или другое, вы можете захотеть использовать и то, и другое.

Task.retry используется для повторного выполнения задач, в частности, для ожидаемых ошибок, которые можно поймать с помощью блока try. Транзакция AMQP не используется для этих ошибок: если задача вызывает исключение, оно все равно подтверждается!.

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

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

@app.task
def process_upload(filename, tmpfile):
    # Increment a file count stored in a database
    increment_file_counter()
    add_file_metadata_to_db(filename, tmpfile)
    copy_file_to_destination(filename, tmpfile)

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

Кроме того, Task.retry имеет возможности, недоступные в транзакциях AMQP: задержка между повторными попытками, максимальное количество повторных попыток и т.д.

Поэтому используйте retry для ошибок Python, и если ваша задача идемпотентна, сочетайте это с acks_late, если требуется такой уровень надежности.

Можно ли запланировать выполнение задач на определенное время?¶

Ответ: Да. Вы можете использовать аргумент eta из Task.apply_async().

См. также Периодические задачи.

Могу ли я безопасно отключить работника?¶

Ответ: Да, используйте сигнал TERM.

Это даст команду рабочему завершить все текущие задания и выключиться как можно скорее. Ни одно задание не должно быть потеряно даже при экспериментальных транспортах, пока завершается выключение.

Вы никогда не должны останавливать worker сигналом KILL (kill -9), если только вы не пробовали TERM несколько раз и не ждали несколько минут, чтобы дать ему возможность отключиться.

Также убедитесь, что вы убиваете только главный рабочий процесс, а не любой из его дочерних процессов. Вы можете направить сигнал kill на определенный дочерний процесс, если знаете, что этот процесс в данный момент выполняет задачу, от которой зависит выключение рабочего, но это также означает, что для задачи будет установлено состояние WorkerLostError, так что задача больше не будет выполняться.

Определить тип процесса проще, если у вас установлен модуль setproctitle:

$ pip install setproctitle

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

Могу ли я запустить рабочий в фоновом режиме на [платформе]?¶

Ответ: Да, пожалуйста, смотрите Демонизация.

Какое назначение имеют таблицы базы данных, созданные командой django-celery-beat?¶

При использовании расписания с опорой на базу данных периодическое расписание задач берется из модели PeriodicTask, есть также несколько других вспомогательных таблиц (IntervalSchedule, CrontabSchedule, PeriodicTasks).

Какое назначение имеют таблицы базы данных, созданные командой django-celery-results?¶

Расширение бэкенда результатов базы данных Django требует двух дополнительных моделей: TaskResult и GroupResult.

Поддерживает ли Celery операционную систему Windows?¶

Ответ: Нет.

Начиная с Celery 4.x, Windows больше не поддерживается из-за нехватки ресурсов.

Но он может продолжать работать, и мы с радостью примем исправления.