如何在龙卷风中使用芹菜，红色作为经纪人和后端?

Question

当我尝试在龙卷风 4.2 中使用芹菜 3.1.11、龙卷风-芹菜 0.3.5 时，它会引发异常。它在没有产量的情况下工作，但无法异步获得结果......当我使用 rabbitmq 作为经纪人时，我也发现它可以工作，而 redis 会引发以下错误......

这是我的代码。

from mycelery import celery_task
import tcelery
tcelery.setup_nonblocking_producer()
token = yield tornado.gen.Task(celery_task.get_rongcloud_token.apply_async,args=[3])
print token

我的芹菜任务：

from celery import Celery, platforms
from celery.schedules import crontab
from celery.utils.log import get_task_logger
from celery.exceptions import SoftTimeLimitExceeded
platforms.C_FORCE_ROOT = True  # linux 下要root用户才不报错
broker = 'redis://:'+settings.REDIS_PASS+'@127.0.0.1:6379/5'
backend = 'redis://:'+settings.REDIS_PASS+'@127.0.0.1:6379/6'
app = Celery('tasks', broker=broker, backend=backend)
@app.task(name='mycelery.celery_task.get_rongcloud_token')
def get_rongcloud_token(user_id):
print 'xxxxx'
a = 'xxx'
return a

这是错误：

TypeError: <function wrapper at 0x5bd2c80> is not JSON serializable

呵呵，找个问题：龙卷风芹菜不能用根。任务或回调

Answer 1

关于tcelery的有限文档解释不了太多，您的示例也没有显示太多。从有限的文档中，您似乎确实在按预期使用它(除非我错过了某些内容(，所以我不确定为什么您的代码不起作用。我已经成功地使用以下方法：

class CeleryTasks(web.RequestHandler):
@gen.coroutine
def get(self):
future = concurrent.Future()
celery_task = tasks.get_rongcloud_token.delay(3)
check_status(celery_task, future)
yield future
self.write(future.result())
#self.write(celery_task.result)

def check_status(celery_task, future):
"""
Check the status of the celery task and set the result in the future
"""
if not celery_task.ready():
ioloop.IOLoop.current().call_later(
1,
check_status,
celery_task,
future)
else:
future.set_result(celery_task.result)

首先创建一个裸Future，这将产生，直到芹菜任务的结果可用。接下来执行芹菜任务，就像通常异步执行一样(例如。task_fn.delay(*args)或task_fn.apply_async(*args)(。将celery_task和future传递到一个函数(check_status(中，该函数将检查任务是否"就绪"，如果没有，则在稍后递归调用自身并再次检查。然后yieldFuture，直到设置结果。任务完成并且结果可用后，使用结果设置Future，并对结果执行任何需要执行的操作。

Answer 2

基于notorious.no的回答：

def check_status(celery_task, future):
"""
celery状态检查辅助函数...
Check the status of the celery task and set the result in the future
"""
import tornado.ioloop
if not celery_task.ready():
tornado.ioloop.IOLoop.current().call_later(
0.1,  # 100ms
check_status,
celery_task,
future)
else:
future.set_result(celery_task.result)
@tornado.gen.coroutine
def celery_get_rongcloud_token(user_id, name='', portraitUri=''):
from mycelery import celery_task
import tornado.concurrent
future = tornado.concurrent.Future()
get_token_task = celery_task.get_rongcloud_token.delay(user_id, name, portraitUri)
check_status(get_token_task, future)
yield future
token = future.result()
raise tornado.gen.Return(token)

然后在您的代码中：

token = yield celery_get_rongcloud_token(user_id)