首页 / 问题 / 正文

Celery "入门"无法获取结果；总是待处理

10 投票
3 回答
12823 浏览
提问于 2025-04-18 18:30

我一直在尝试按照 Celery 的入门指南和下一步指南进行设置。我的环境是 Windows 7 64 位，Anaconda Python 2.7（32 位），安装了 Erlang 32 位的二进制文件，还有 RabbitMQ 服务器，以及通过 pip install celery 安装的 celery。

按照指南，我创建了一个 proj 文件夹，里面有 init.py、tasks.py 和 celery.py。我的 init.py 是空的。下面是 celery.py 的内容：

from __future__ import absolute_import

from celery import Celery

app = Celery('proj',
             broker='amqp://',
             backend='amqp://',
             include=['proj.tasks'])

#Optional configuration, see the application user guide
app.conf.update(
    CELERY_TASK_RESULT_EXPIRES=3600,
    CELERY_TASK_SERIALIZER='json',
    CELERY_ACCEPT_CONTENT=['json'],  # Ignore other content
    CELERY_RESULT_SERIALIZER='json',
)

if __name__ == '__main__':
    app.start()

这是 tasks.py 的内容：

from __future__ import absolute_import

from .celery import app

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


@app.task
def mul(x, y):
    return x * y


@app.task
def xsum(numbers):
    return sum(numbers)

首先，我知道我需要确保 RabbitMQ 服务正在运行。任务管理器的服务标签显示 RabbitMQ 确实在运行。为了启动 celery 服务器并加载我的任务，我打开 cmd.exe，导航到 proj 的上级目录（我称这个文件夹为 celery_demo），然后运行这个命令：

celery -A proj.celery worker -l debug

运行后得到这个输出：

C:\Users\bnables\Documents\Python\celery_demo>celery -A proj.celery worker -l debug
[2014-08-25 17:00:09,308: DEBUG/MainProcess] | Worker: Preparing bootsteps.
[2014-08-25 17:00:09,313: DEBUG/MainProcess] | Worker: Building graph...
[2014-08-25 17:00:09,315: DEBUG/MainProcess] | Worker: New boot order: {Timer, Hub, Queues (intra), Pool, Autoreloader, Autoscaler, StateDB, Beat, Con
sumer}
[2014-08-25 17:00:09,322: DEBUG/MainProcess] | Consumer: Preparing bootsteps.
[2014-08-25 17:00:09,322: DEBUG/MainProcess] | Consumer: Building graph...
[2014-08-25 17:00:09,332: DEBUG/MainProcess] | Consumer: New boot order: {Connection, Events, Heart, Mingle, Gossip, Tasks, Control, Agent, event loop
}

 -------------- celery@MSSLW40013047 v3.1.13 (Cipater)
---- **** -----
--- * ***  * -- Windows-7-6.1.7601-SP1
-- * - **** ---
- ** ---------- [config]
- ** ---------- .> app:         proj:0x3290370
- ** ---------- .> transport:   amqp://guest:**@localhost:5672//
- ** ---------- .> results:     amqp
- *** --- * --- .> concurrency: 8 (prefork)
-- ******* ----
--- ***** ----- [queues]
 -------------- .> celery           exchange=celery(direct) key=celery


[tasks]
  . celery.backend_cleanup
  . celery.chain
  . celery.chord
  . celery.chord_unlock
  . celery.chunks
  . celery.group
  . celery.map
  . celery.starmap
  . proj.tasks.add
  . proj.tasks.mul
  . proj.tasks.xsum

[2014-08-25 17:00:09,345: DEBUG/MainProcess] | Worker: Starting Pool
[2014-08-25 17:00:09,417: DEBUG/MainProcess] ^-- substep ok
[2014-08-25 17:00:09,420: DEBUG/MainProcess] | Worker: Starting Consumer
[2014-08-25 17:00:09,421: DEBUG/MainProcess] | Consumer: Starting Connection
[2014-08-25 17:00:09,457: DEBUG/MainProcess] Start from server, version: 0.9, properties: {u'information': u'Licensed under the MPL.  See http://www.r
abbitmq.com/', u'product': u'RabbitMQ', u'copyright': u'Copyright (C) 2007-2014 GoPivotal, Inc.', u'capabilities': {u'exchange_exchange_bindings': Tru
e, u'connection.blocked': True, u'authentication_failure_close': True, u'basic.nack': True, u'per_consumer_qos': True, u'consumer_priorities': True, u
'consumer_cancel_notify': True, u'publisher_confirms': True}, u'cluster_name': u'rabbit@MSSLW40013047.ndc.nasa.gov', u'platform': u'Erlang/OTP', u'ver
sion': u'3.3.5'}, mechanisms: [u'AMQPLAIN', u'PLAIN'], locales: [u'en_US']
[2014-08-25 17:00:09,460: DEBUG/MainProcess] Open OK!
[2014-08-25 17:00:09,460: INFO/MainProcess] Connected to amqp://guest:**@127.0.0.1:5672//
[2014-08-25 17:00:09,461: DEBUG/MainProcess] ^-- substep ok
[2014-08-25 17:00:09,461: DEBUG/MainProcess] | Consumer: Starting Events
[2014-08-25 17:00:09,516: DEBUG/MainProcess] Start from server, version: 0.9, properties: {u'information': u'Licensed under the MPL.  See http://www.r
abbitmq.com/', u'product': u'RabbitMQ', u'copyright': u'Copyright (C) 2007-2014 GoPivotal, Inc.', u'capabilities': {u'exchange_exchange_bindings': Tru
e, u'connection.blocked': True, u'authentication_failure_close': True, u'basic.nack': True, u'per_consumer_qos': True, u'consumer_priorities': True, u
'consumer_cancel_notify': True, u'publisher_confirms': True}, u'cluster_name': u'rabbit@MSSLW40013047.ndc.nasa.gov', u'platform': u'Erlang/OTP', u'ver
sion': u'3.3.5'}, mechanisms: [u'AMQPLAIN', u'PLAIN'], locales: [u'en_US']
[2014-08-25 17:00:09,519: DEBUG/MainProcess] Open OK!
[2014-08-25 17:00:09,520: DEBUG/MainProcess] using channel_id: 1
[2014-08-25 17:00:09,522: DEBUG/MainProcess] Channel open
[2014-08-25 17:00:09,523: DEBUG/MainProcess] ^-- substep ok
[2014-08-25 17:00:09,523: DEBUG/MainProcess] | Consumer: Starting Heart
[2014-08-25 17:00:09,530: DEBUG/MainProcess] ^-- substep ok
[2014-08-25 17:00:09,533: DEBUG/MainProcess] | Consumer: Starting Mingle
[2014-08-25 17:00:09,538: INFO/MainProcess] mingle: searching for neighbors
[2014-08-25 17:00:09,539: DEBUG/MainProcess] using channel_id: 1
[2014-08-25 17:00:09,540: DEBUG/MainProcess] Channel open
[2014-08-25 17:00:10,552: INFO/MainProcess] mingle: all alone
[2014-08-25 17:00:10,552: DEBUG/MainProcess] ^-- substep ok
[2014-08-25 17:00:10,552: DEBUG/MainProcess] | Consumer: Starting Gossip
[2014-08-25 17:00:10,553: DEBUG/MainProcess] using channel_id: 2
[2014-08-25 17:00:10,555: DEBUG/MainProcess] Channel open
[2014-08-25 17:00:10,559: DEBUG/MainProcess] ^-- substep ok
[2014-08-25 17:00:10,559: DEBUG/MainProcess] | Consumer: Starting Tasks
[2014-08-25 17:00:10,566: DEBUG/MainProcess] ^-- substep ok
[2014-08-25 17:00:10,566: DEBUG/MainProcess] | Consumer: Starting Control
[2014-08-25 17:00:10,568: DEBUG/MainProcess] using channel_id: 3
[2014-08-25 17:00:10,569: DEBUG/MainProcess] Channel open
[2014-08-25 17:00:10,572: DEBUG/MainProcess] ^-- substep ok
[2014-08-25 17:00:10,573: DEBUG/MainProcess] | Consumer: Starting event loop
[2014-08-25 17:00:10,575: WARNING/MainProcess] celery@MSSLW40013047 ready.
[2014-08-25 17:00:10,575: DEBUG/MainProcess] basic.qos: prefetch_count->32

这里的 -A 是告诉 celery 去哪里找我的 celery 应用实例。用 proj 也可以，但那样它只会搜索 proj.celery，所以这里更明确一点是没问题的。worker 是给 celery 的命令，告诉它要启动一些工作进程来执行从 proj.celery 加载的任务。最后，-l debug 是告诉 celery 设置日志级别为调试，这样我能看到很多信息。通常情况下，这里会是 -l info。

为了测试我的任务服务器，我打开一个 IPython Qt 控制台，导航到 celery_demo 文件夹（里面有 proj）。然后我输入 from proj.tasks import add。简单调用 add(1, 2) 会返回 3，这没有使用服务器，正如预期的那样。当我调用 add.delay 时，发生了以下情况：

add.delay(2, 3)

返回结果是：

<AsyncResult: 42123ff3-e94e-4673-808a-ec6c847679d8>

在我的 cmd.exe 窗口中，我得到了：

[2014-08-25 17:20:38,109: INFO/MainProcess] Received task: proj.tasks.add[42123ff3-e94e-4673-808a-ec6c847679d8]
[2014-08-25 17:20:38,109: DEBUG/MainProcess] TaskPool: Apply <function _fast_trace_task at 0x033CD6F0> (args:(u'proj.tasks.add', u'42123ff3-e94e-4673-
808a-ec6c847679d8', [2, 3], {}, {u'timelimit': [None, None], u'utc': True, u'is_eager': False, u'chord': None, u'group': None, u'args': [2, 3], u'retr
ies': 0, u'delivery_info': {u'priority': None, u'redelivered': False, u'routing_key': u'celery', u'exchange': u'celery'}, u'expires': None, u'hostname
': 'celery@MSSLW40013047', u'task': u'proj.tasks.add', u'callbacks': None, u'correlation_id': u'42123ff3-e94e-4673-808a-ec6c847679d8', u'errbacks': No
ne, u'reply_to': u'70ed001d-193c-319c-9447-8d77c231dc10', u'taskset': None, u'kwargs': {}, u'eta': None, u'id': u'42123ff3-e94e-4673-808a-ec6c847679d8
', u'headers': {}}) kwargs:{})
[2014-08-25 17:20:38,124: DEBUG/MainProcess] Task accepted: proj.tasks.add[42123ff3-e94e-4673-808a-ec6c847679d8] pid:4052
[2014-08-25 17:20:38,125: INFO/MainProcess] Task proj.tasks.add[42123ff3-e94e-4673-808a-ec6c847679d8] succeeded in 0.0130000114441s: 5

所以最后一行显示，结果 5 正在被计算。接下来，我想存储 AsyncResult 对象，检查它的状态并获取结果值：

result = add.delay(3, 4)

但是 result.state 和 result.get(timeout=1) 的表现并不如预期：

In:  result.state
Out: 'Pending'
In:  result.status
Out: 'Pending'

In: result.get(timeout=1)
---------------------------------------------------------------------------
TimeoutError                              Traceback (most recent call last)
<ipython-input-17-375f2d3530cb> in <module>()
----> 1 result.get(timeout=1)

C:\Anaconda32\lib\site-packages\celery\result.pyc in get(self, timeout, propagate, interval, no_ack, follow_parents)
    167                 interval=interval,
    168                 on_interval=on_interval,
--> 169                 no_ack=no_ack,
    170             )
    171         finally:

C:\Anaconda32\lib\site-packages\celery\backends\amqp.pyc in wait_for(self, task_id, timeout, cache, propagate, no_ack, on_interval, READY_STATES, PROPAGATE_STATES, **kwargs)
    155                                     on_interval=on_interval)
    156             except socket.timeout:
--> 157                 raise TimeoutError('The operation timed out.')
    158 
    159         if meta['status'] in PROPAGATE_STATES and propagate:

TimeoutError: The operation timed out.

我期望的 result.state 或 result.status 应该是 'SUCCESSFUL'，而 result.get(timeout=1) 的结果应该是 5。

看起来结果存储或消息传递没有正常工作。教程只是说在调用 Celery() 时设置名为 backend 的参数，或者在 CELERY_RESULT_BACKEND 配置中设置。在“入门”中，它有 backend='amqp'，而“下一步”中则有 backend='amqp://'，这也是在github 示例中使用的。

我在这个问题上纠结了很久，完全不知道该怎么继续。有没有什么建议可以尝试的？谢谢！

debugging asynchronous programming celery rabbitmq message broker erlang task management result backend

3 个回答

2

.delay 方法会把你的任务放入一个队列中。“result.state” 显示为 pending，这意味着你的任务还没有执行。可能队列里有很多任务，所以你的任务被延迟了。

检查一下是否有其他任务正在运行。

>>> from celery.task.control import inspect
>>> i = inspect()
>>> i.scheduled()
>>> i.active()
回答于 2025-04-18 由 Python大师
分享 举报
5

你必须加上 track_started=True。这个选项可能不太好知道。

默认情况下，这个值是 False，所以当任务没有完成时，你总是会看到 PENDING（待处理）。另外，你可能在后端或消息中介的配置上出错了。请再检查一下。

@app.task(bind=True, track_started=True)
def add(self, x, y):
    return x + y
回答于 2025-04-18 由 Python大师
分享 举报
22
  • W8 x64
  • Python 2.7.3 (ActivePython)
  • Erlang 17.1 x64
  • RabbitMQ服务器 3.3.5
  • Celery 3.1.13

我这边也随机出现停止工作的情况。问题和之前的一样——一直在等待中。重新安装Erlang或者RabbitMQ都没用。

我还在Debian Linux 7 x86上测试过，这里一切正常，没有问题。

另外：https://github.com/celery/celery/issues/2146

这可能是和Windows有关的问题，设置工作标志--pool=solo目前对我解决了这个问题。

回答于 2025-04-18 由 Python大师
分享 举报

撰写回答