同时在进程中使用 Celery 并在任务中使用 gevent

Question

我想将 Celery 用作我的任务队列，这样我的网络应用程序就可以将任务排入队列，返回响应，并且该任务将同时处理/某天/...我构建了一种 API，所以我提前不知道会有什么样的任务——未来可能会有处理HTTP请求的任务，另一个IO，也有CPU消耗的任务。在这方面，我想在进程上运行 Celery 的工作人员，因为这些是 Python 中通用的并行性。

但是，我也想在我的任务中使用 gevent，所以我可以让一个任务产生许多 HTTP 请求等。问题是，当我这样做时：

from gevent import monkey
monkey.patch_all()

芹菜停止工作。它开始了，但没有任务可以有效地排队——它们似乎去代理，但芹菜工人没有收集和处理它们。只能启动和等待。如果我删除这些行并在没有任何 gevent 和并行化的情况下执行任务，那么一切正常。

我认为这可能是因为 gevent 补丁也线程化。所以我尝试了

from gevent import monkey
monkey.patch_all(thread=False)

...但是 Celery 甚至没有启动，它在没有给出原因的情况下崩溃（打开了调试级别的日志记录）。

是否可以使用 Celery 将任务排入队列并使用 gevent 在单个任务中做一些事情？如何？我做错了什么？

Answer 1

我认为启动任务的推荐方式如下。

python manage.py celery worker -P gevent --loglevel=INFO

Gevent 需要尽早修补。

Answer 2

您可以使用包含多个 greenlet 的多个线程运行 celery，如下所示：

$ celery multi start 4 -P gevent -l info -c:1-4 1000

Answer 3

根据我奇怪的经验，Celery Beat 无法与具有 gevent 池的工作人员正常工作（计划任务被阻止并永远等待），除非您为 Beat 进程激活 gevent Monkey 修补。

但是，celery beat 不支持 --pool=gevent 或 -P gevent 选项。注入 gevent 猴子补丁的正确方法是使用自定义的celery 二进制文件，例如：

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from gevent import monkey
monkey.patch_all()

import re
import sys

from celery.__main__ import main

if __name__ == '__main__':
    sys.argv[0] = re.sub(r'(-script\.pyw|\.exe)?$', '', sys.argv[0])
    sys.exit(main())

保存为celery-gevent，运行Beat服务如下：

celery-gevent beat --app=proj.celery:app --loader=djcelery.loaders.DjangoLoader -f /var/log/celery/beat.log -l INFO --workdir=/my/proj --pidfile=/var/run/celery/beat.pid

在proj.celery 中，您还应该修补Django 连接以避免DatabaseError：

from __future__ import absolute_import

import os
# Set the Django settings module for the 'celery' program
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'proj.settings')

import django
# Load Django model definitions, etc
django.setup()

from django.db import connection
# Allow thread sharing to ensure that Django database connection
# works properly with gevent.
connection.allow_thread_sharing = True

from django.conf import settings
from celery import Celery

app = Celery('proj')

# Using a string here means the worker will not have to
# pickle the object when using Windows.
app.config_from_object('django.conf:settings')
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)

（以上示例适用于 Python 2.7.10、Celery 3.1.18、Django 1.8.2 和 gevent 1.0.2）

Answer 4

据我所知，这不可能。如果有人找到更好的答案，我会接受它而不是这个我的。

唯一的选择是将 gevent 也用作 Celery 工作人员的后端。为了完成这样的事情，需要做的是在配置文件中添加以下内容：

CELERYD_POOL = 'gevent'

可以在here 找到有关此选项的更多详细信息。有关 gevent 池的更多信息是on this page。请注意 gevent 池仍被标记为实验性的事实。我发现没有可用于比较不同任务（面向 IO 的任务、面向 CPU 的任务）上的进程和异步 gevent 池的基准，但最后我意识到即使是我的 CPU-bound 任务实际上也会更多 IO比CPU，因为我使用数据库来保存结果，数据库连接将是一个瓶颈，而不是计算部分。我不会有真正会占用 CPU 的科学任务。