在tornado中调用celery，用tornado-celery库，但是异步调用完后，没有回调

发布于 2022-09-02 00:04:56 字数 4203 浏览 17 评论 0

hi，all

背景

最近在一个tornado项目里面，用了celery做队列，因为底层存储是mysql，所以想到用celery来把耗时的请求交给它做。celery配置的broker是rabbitmq，然后搭配tornado-celery库来进行调用。

问题

我看了下网上关于tornado-celery的教程，在ubuntu和mac os上面跑了下面这个例子，两个地方用的库的版本基本是一致的，但是osx那边的调用一直不成功，调用下面例子的curl http://localhost/sleep命令，请求就卡死在那里了。

例子如下

task.py

#!/usr/bin/env python
# encoding: utf-8

import time
from celery import Celery

C_FORCE_ROOT=True

celery = Celery("tasks", broker="amqp://guest:guest@localhost:5672")
celery.conf.CELERY_RESULT_BACKEND = "amqp"


@celery.task
def sleep(seconds):
    time.sleep(float(seconds))
    return seconds

if __name__ == "__main__":
    celery.start()

server.py

#!/usr/bin/env python
# encoding: utf-8

import tornado.httpserver
import tornado.ioloop
import tornado.options
import tornado.web
import tornado.gen
import tornado.httpclient

import time
from celery import Celery
import tcelery, task

# celery = Celery("tasks", broker="amqp://guest:guest@localhost:5672")
# celery.conf.CELERY_RESULT_BACKEND = "amqp"

from tornado.options import define, options
define("port", default=8000, help="run on the given port", type=int)

tcelery.setup_nonblocking_producer()

class SleepHandler(tornado.web.RequestHandler):
    # @tornado.web.asynchronous
    @tornado.gen.coroutine
    def get(self):
        # tornado.gen.Task的参数是:要执行的函数, 参数
        yield tornado.gen.Task(task.sleep.apply_async, args=[1])
        self.write("when i sleep 5s")
        self.finish()

class JustNowHandler(tornado.web.RequestHandler):
    def get(self):
        self.write("i hope just now see you")

if __name__ == "__main__":
    tornado.options.parse_command_line()
    app = tornado.web.Application(handlers=[
            (r"/sleep", SleepHandler), (r"/justnow", JustNowHandler)])
    http_server = tornado.httpserver.HTTPServer(app)
    http_server.listen(options.port)
    tornado.ioloop.IOLoop.instance().start()

运行命令如下：

对于task.py

celery -A task worker --loglevel=info

对于server.py

python server.py

测试命令如下：

curl http://localhost:8000/sleep

其他

我的python环境信息

amqp (1.4.6)
anyjson (0.3.3)
backports.ssl-match-hostname (3.4.0.2)
billiard (3.3.0.20)
celery (3.1.18)
certifi (2015.9.6.2)
ecdsa (0.13)
kombu (3.0.26)
MySQL-python (1.2.5)
paramiko (1.15.2)
pika (0.10.0)
pip (7.1.2)
pycrypto (2.6.1)
pytz (2015.4)
redis (2.10.3)
redisco (0.1.4)
requests (2.7.0)
setuptools (18.2)
SQLAlchemy (1.0.8)
tornado (4.1)
tornado-celery (0.3.4)
wheel (0.24.0)

我的系统信息 mac os 10.11
rabbitmq版本信息如下

(zizcloud)➜  test git:(develop) ✗ rabbitmqctl status
Status of node rabbit@localhost ...
[{pid,14825},
 {running_applications,[{rabbit,"RabbitMQ","3.2.4"},
                        {mnesia,"MNESIA  CXC 138 12","4.10"},
                        {os_mon,"CPO  CXC 138 46","2.2.13"},
                        {xmerl,"XML parser","1.3.4"},
                        {sasl,"SASL  CXC 138 11","2.3.3"},
                        {stdlib,"ERTS  CXC 138 10","1.19.3"},
                        {kernel,"ERTS  CXC 138 10","2.16.3"}]},
 {os,{unix,darwin}},
 {erlang_version,"Erlang R16B02 (erts-5.10.3) [source] [smp:2:2] [async-threads:30] [hipe] [kernel-poll:true]\n"},
 {memory,[{total,21551368},
          {connection_procs,1444},
          {queue_procs,2888},
          {plugins,0},
          {other_proc,8883464},
          {mnesia,31760},
          {mgmt_db,0},
          {msg_index,12148},
          {other_ets,429988},
          {binary,8384},
          {code,8784583},
          {atom,461133},
          {other_system,2935576}]},
 {vm_memory_high_watermark,0.4},
 {vm_memory_limit,858993459},
 {disk_free_limit,50000000},
 {disk_free,137763581952},
 {file_descriptors,[{total_limit,2460},
                    {total_used,3},
                    {sockets_limit,2212},
                    {sockets_used,1}]},
 {processes,[{limit,1048576},{used,123}]},
 {run_queue,0},
 {uptime,7}]
...done.

分享到QQ

分享到微博