python - Celery:轮询正在运行的任务以获取结果

Question

项目/

celery .py

from __future__ import absolute_import

from kombu import Exchange, Queue
from celery import Celery

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

a_exchange = Exchange('a_ex', type='topic')

# Optional configuration, see the application user guide.
app.conf.update(
    CELERY_TASK_RESULT_EXPIRES=3600,
    CELERY_DISABLE_RATE_LIMITS=True,
    CELERY_ROUTES = {"app.tasks.timeme": "a"}
)

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

任务.py

from __future__ import absolute_import

from proj.celery import app
import time

@app.task
def timeme(ts):
    print 'hi'
    lat = time.time() - float(ts)
    return (lat, time.time())

do_tasks.py

import proj.tasks
import time
import sys

stime = time.time()
running = []
while time.time() < stime + 15:
    res = proj.tasks.timeme.apply_async(args=[time.time()], link=proj.tasks.timeme.s())
    running.append(res)

    for res in running:      #<------------ this gets extremely slow if running gets big!
          if res.ready()
              print res.get()

在上面的代码中，随着 running 越来越大，循环 running 并查看它是否准备好结果需要很长时间。

在运行 celery 任务时是否有类似 select.select 或 poll/epoll 的东西？

所以我可以做类似下面的事情:

While running:
    read, w, e = select.select([running], [], [])
    print read.get()
    running.remove(read)
    break

Answer 1

一句话，没有。

但是您可以使用 celery.app.control.inspect 获得您想要的东西

i = app.control.inspect()
i.active()
[{'worker1.example.com':
    [{'name': 'tasks.sleeptask',
      'id': '32666e9b-809c-41fa-8e93-5ae0c80afbbf',
      'args': '(8,)',
      'kwargs': '{}'}]}]