带有mongodb结果后端的示例celeniv4.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似乎都忽略了配置设置，在没有任何结果后端的情况下启动。anywon有一个使用最新版本的Celery的工作示例吗？我能找到的唯一其他例子是v3 Celery设置，这对我来说也不起作用，因为我在生产中使用的是Mongo副本集群，而该版本似乎不支持它。

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

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

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')))