python - Flask with Celery - worker 以 exitcode1 退出

Question

我有一个装有 celery 的 flask 应用程序。
当我按如下方式运行工作人员时:

celery -A app.celery worker

我得到以下输出

 -------------- celery@local-pc v3.1.22 (Cipater)
---- **** -----
--- * ***  * -- Windows-7-6.1.7601-SP1
-- * - **** ---
- ** ---------- [config]
- ** ---------- .> app:         app:0x483e668
- ** ---------- .> transport:   mongodb://localhost:27017/app
- ** ---------- .> results:     mongodb://localhost:27017/app
- *** --- * --- .> concurrency: 8 (prefork)
-- ******* ----
--- ***** ----- [queues]
 -------------- .> celery           exchange=celery(direct) key=celery


[2016-03-08 15:52:05,587: WARNING/MainProcess] celery@local-pc ready.
[2016-03-08 15:52:08,855: ERROR/MainProcess] Process 'Worker-8' pid:9720 exited with 'exitcode 1'
[2016-03-08 15:52:08,855: ERROR/MainProcess] Process 'Worker-7' pid:11940 exited with 'exitcode 1'
[2016-03-08 15:52:08,856: ERROR/MainProcess] Process 'Worker-6' pid:13120 exited with 'exitcode 1'
...

它无休止地运行，CPU 提高到 100%。

相关配置为:

CELERY_BROKER_URL = 'mongodb://localhost:27017/app'
CELERY_RESULT_BACKEND = 'mongodb://localhost:27017/'
CELERY_MONGODB_BACKEND_SETTINGS = {
    'database': 'app',
    'taskmeta_collection': 'my_taskmeta_collection',
}
CELERY_IMPORTS = ('app.tasks', )
CELERYD_FORCE_EXEC = True
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'

我的项目结构是:

proj/
    config.py
    app/
        __init__.py
        tasks.py
        views.py

这就是我在 ___init___.py 中配置 celery 的方式”:

app = Flask(__name__)
app.config.from_object('config')
db = SQLAlchemy(app)


def make_celery(app):
    celery = Celery(app.import_name, broker=app.config['CELERY_BROKER_URL'])
    celery.conf.update(app.config)
    TaskBase = celery.Task

    class ContextTask(TaskBase):
        abstract = True

        def __call__(self, *args, **kwargs):
            with app.app_context():
                return TaskBase.__call__(self, *args, **kwargs)
    celery.Task = ContextTask
    return celery

celery = make_celery(app)

这就是我在 tasks.py 中的内容

from app import celery


@celery.task()
def add_together(a, b):
    return a + b

更新

当我从配置文件中删除以下行时，worker 不会退出

CELERY_IMPORTS = ('app.tasks', )

但我得到以下错误

Traceback (most recent call last):
  File "d:\python34\lib\site-packages\celery\worker\consumer.py", line 456, in on_task_received
    strategies[name](message, body,
KeyError: 'app.tasks.add_together'

Answer 1

在官方 Celery 网站上 http://docs.celeryproject.org/en/3.1/getting-started/brokers/mongodb.html

说明当使用 mongoDB 作为代理时，您需要自行承担风险。

Using MongoDB

Experimental Status

The MongoDB transport is in need of improvements in many areas and there are several open bugs. Unfortunately we don’t have the resources or funds required to improve the situation, so we’re looking for contributors and partners willing to help.

要测试这是否不是产生问题的原因，请尝试使用 redis 或 rabbitMQ 作为代理。