项目/
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
最佳答案
一句话,不。
但是,您也许可以使用 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': '{}'}]}]
关于python - Celery:轮询运行任务的结果,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/33093579/