带有 mongodb 结果后端的示例 celery v4.2

Question

我正在尝试在 Flask 下设置一个 Celery 应用程序以接受 API 请求，然后将 Celery 工作人员分开以执行长时间运行的任务。我的问题是我的 Flask 和我环境中的其他所有东西都使用 MongoDB，所以我不想为 Celery 结果设置一个单独的 SQL 数据库。我找不到任何关于如何以 MongoDB 集群作为后端正确配置 Celery 的好例子。

这是我试图让它接受的设置：

CELERY_RESULT_BACKEND = "mongodb"
CELERY_MONGODB_BACKEND_SETTINGS = {"host": "mongodb://mongodev:27017",
                               "database": "celery",
                               "taskmeta_collection": "celery_taskmeta"}

无论我做什么，Celery 似乎都忽略了配置设置并且在没有任何结果后端的情况下启动。有没有使用最新版本的芹菜的工作示例？我能找到的唯一其他示例是 v3 Celery 设置，这对我也不起作用，因为我在生产中使用 Mongo 副本集群，该版本似乎不受支持。

[编辑]以复杂的方式添加更多信息，我将配置设置为与应用程序的其余部分一起使用。

配置值首先通过 docker-compose 文件作为环境变量传递，如下所示：

environment:
  - PYTHONPATH=/usr/src/
  - APP_SETTINGS=config.DevelopmentConfig
  - FLASK_ENV=development
  - CELERY_BROKER_URL=amqp://guest:guest@rabbit1:5672
  - CELERY_BROKER_DEV=amqp://guest:guest@rabbit1:5672
  - CELERY_RESULT_SERIALIZER=json
  - CELERY_RESULT_BACKEND=mongodb
  - CELERY_MONGODB_BACKEND_SETTINGS={"host":"mongodb://mongodev:27017","database":"celery","taskmeta_collection":"celery_taskmeta"}

然后，在 config.py 文件中加载它们：

class DevelopmentConfig(BaseConfig):
    """Development configuration"""
    CELERY_BROKER_URL = os.getenv('CELERY_BROKER_DEV')
    CELERY_RESULT_SERIALIZER = os.getenv('CELERY_RESULT_SERIALIZER')
    CELERY_RESULT_BACKEND = os.getenv('CELERY_RESULT_BACKEND')
    CELERY_MONGODB_BACKEND_SETTINGS = ast.literal_eval(os.getenv('CELERY_MONGODB_BACKEND_SETTINGS'))

然后，当 Celery 启动时，会加载配置：

app = Celery('celeryworker', broker=os.getenv('CELERY_BROKER_URL'),
             include=['celeryworker.tasks'])
print('app initiated')
app.config_from_object(app_settings)
app.conf.update(accept_content=['json'])
print("CELERY_MONGODB_BACKEND_SETTINGS",
       os.getenv('CELERY_MONGODB_BACKEND_SETTINGS'))
print("celery config",app.conf)

当应用程序出现在此处时，我看到的所有故障排除打印件都是如此。我已经编辑了很多配置输出，只是为了显示我在这里的内容是通过 config.py 到 app.config 但被 Celery 忽略。您可以看到该值将其放入 celery.py 文件中，我确信 Celery 会对其进行处理，因为在我在 config.py 中添加 ast.literal_eval 之前，Celery 会抛出一个错误，指出 MongoDB 后端设置需要是dict 而不是一个字符串。不幸的是，现在它作为正确的字典被传递，Celery 忽略了它。

app_settings SGSDevOps.config.DevelopmentConfig
app initiated
CELERY_MONGODB_BACKEND_SETTINGS {"host":"mongodb://mongodev:27017","database":"celery","taskmeta_collection":"celery_taskmeta"}
celery config Settings(Settings({'BROKER_URL': 'amqp://guest:guest@rabbit1:5672', 'CELERY_INCLUDE': ['celeryworker.tasks'], 'CELERY_ACCEPT_CONTENT': ['json']}, 'BROKER_URL': 'amqp://guest:guest@rabbit1:5672', 'CELERY_MONGODB_BACKEND_SETTINGS': None, 'CELERY_RESULT_BACKEND': None}))
APP_SETTINGS config.DevelopmentConfig
app.config <Config {'ENV': 'development', 'CELERY_BROKER_URL': 'amqp://guest:guest@rabbit1:5672', 'CELERY_MONGODB_BACKEND_SETTINGS': {'host': 'mongodb://mongodev:27017', 'database': 'celery', 'taskmeta_collection': 'celery_taskmeta'}, 'CELERY_RESULT_BACKEND': 'mongodb', 'CELERY_RESULT_SERIALIZER': 'json', }>

-------------- celery@a5ea76b91f77 v4.2.1 (windowlicker)
---- **** -----
--- * ***  * -- Linux-4.9.93-linuxkit-aufs-x86_64-with-debian-9.4 2018-10-29 17:25:27
-- * - **** ---
- ** ---------- [config]
- ** ---------- .> app:         celeryworker:0x7f28e828f668
- ** ---------- .> transport:   amqp://guest:**@rabbit1:5672//
- ** ---------- .> results:     mongodb://
- *** --- * --- .> concurrency: 2 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** -----
 -------------- [queues]
                .> celery           exchange=celery(direct) key=celery

[tasks]
  . celeryworker.tasks.longtime_add

Answer 1

我仍然不知道为什么上述配置不起作用，但我找到了一种解决方法，可以在应用加载后使用新的配置值名称更新配置：

app = Celery('celeryworker', broker=os.getenv('CELERY_BROKER_URL'),
              backend=os.getenv('CELERY_RESULT_BACKEND'),
              include=['SGSDevOps.celeryworker.tasks'])
print('app initiated')
app.config_from_object(app_settings)
app.conf.update(accept_content=['json'])
app.conf.update(mongodb_backend_settings=ast.literal_eval(os.getenv('CELERY_MONGODB_BACKEND_SETTINGS')))