基于以前的一篇文章,celery+Rabbit MQ的安装和使用,
本文更加详细的介绍如何安装和使用celey, Rabbit MQ。
并记录在使用celery时遇到的一些问题。
1.安装 Rabbit MQ
在 OS X上,直接执行如下命令:
$ brew install rabbitmq
其他操作系统下的安装可以参考安装 RabbitMQ
启动RabbitMQ
$ sudo rabbitmq-server
你也可以添加 -detached 属性来让它在后台运行(注意:只有一个破折号):
$ sudo rabbitmq-server -detached
查看RabbitMQ状态
$ sudo rabbitmqctl status
停止RabbitMQ
永远不要用 kill 停止 RabbitMQ 服务器,而是应该用 rabbitmqctl 命令:
$ sudo rabbitmqctl stop
添加用户
默认用户guest,密码guest,只允许本地访问,如需远程访问,需要设置.
$ sudo rabbitmqctl add_user test 123456
Adding user "test" ...
添加虚拟主机,并赋予用户test权限
$ sudo rabbitmqctl add_vhost myvhost
Adding vhost "myvhost" ...
$ sudo rabbitmqctl set_permissions -p myvhost test ".*" ".*" ".*"
Setting permissions for user "test" in vhost "myvhost" ...
2.安装celery
2.1 创建虚拟环境,并安装celery
$ mkdir celery_demo
$ cd celery_demo
$ virtualenv -p python3 venv3
$ ./venv3/bin/pip install celery
2.2 配置celery
创建配置文件 celeryconfig.py,里面包含BROKER_URL
、CELERYD_LOG_FORMAT
、CELERY_ROUTES
.
# celeryconfig.py
RABBIT_MQ = {
'HOST': '127.0.0.1',
'PORT': 5672,
'USER': 'test',
'PASSWORD': '123456'
}
# broker
BROKER_URL = 'amqp://%s:%s@%s:%s/myvhost' % (RABBIT_MQ['USER'], RABBIT_MQ['PASSWORD'], RABBIT_MQ['HOST'], RABBIT_MQ['PORT'])
# celery日志格式
CELERYD_LOG_FORMAT = '[%(asctime)s] [%(levelname)s] %(message)s'
CELERY_ROUTES = {
'demo_task.add': {'queue': 'sunday'},
}
其中,参数定义如下:
BROKER_URL
指定了broker信息,即消息队列的地址。CELERYD_LOG_FORMAT
指定了日志格式。CELERY_ROUTES
指定了路由信息,即调用demo_task.add
后,消息具体放入哪个队列,这里是队列名称为sunday
。
2.3 启动消费者
消费者
消费者代码如下:
# demo_task.py
from celery import Celery
app = Celery("orange", backend='amqp')
app.config_from_object("celeryconfig")
@app.task
def add(x, y):
return x + y
首先,创建Celery
实例,从文件中读取配置。
接着,定义task。
其中,在创建Celery
实例时,参数backend
指定了结果存储后端,用于追踪task执行状态和结果。这里使用amqp
,即使用RabbitMQ保存结果。默认情况下,backend
参数是关闭的。
使用celery worker
启动消费者
./venv3/bin/celery worker -A demo_task -Q sunday --loglevel=info -f app.log
[email protected] v4.3.0 (rhubarb)
Darwin-18.2.0-x86_64-i386-64bit 2019-04-20 17:49:40
[config]
.> app: orange:0x109ca0400
.> transport: amqp://test:**@127.0.0.1:5672/myvhost
.> results: disabled://
.> concurrency: 8 (prefork)
.> task events: OFF (enable -E to monitor tasks in this worker)
[queues]
.> sunday exchange=sunday(direct) key=sunday
[tasks]
. demo_task.add
其中,参数定义如下:
参数
-A
是app name,即定义celery的文件。参数
-Q
指定了队列的名称,如果不指定,默认为celery
。参数
-f
指定了日志打印文件。
可以通过以下命令查看更多帮助信息:
celery help
查看celery的选项celery worker --help
查看worker的选项
2.4 启动生产者
生产者
代码如下api.py所示:
# api.py
from demo_task import add
print("start...")
result = add.apply_async((1, 2))
print("result:", result)
print(result.ready())
print("end...")
代码中,将1
、2
参数放入消息队列。
celery配置backend
参数后, 调用任务时,会返回 AsyncResult
实例。
AsyncResult
的ready()
方法可以查看任务是否完成处理。
执行api.py
执行结果分两种情况:worker启动和没有启动。
当celery worker启动的时候,结果如下:
./venv3/bin/python api.py
start...
result: 60dd0ab6-5fa2-4190-954e-584eb519384f
True
end...
可以看到,task很快得到执行,结果状态为True。
当celery worker没有启动的时候,结果如下:
./venv3/bin/python api.py
start...
result: 7280466f-73cd-44ec-85e7-5ad4f079a797
False
end...
可以看到,结果状态为False。
直接获取结果
可以使用get()
函数等待任务完成,但这很少使用,因为它把异步调用变成了同步调用:
... ...
# sync
print(result.get(timeout=1))
print("end...")
并且,如果worker没有启动,会报超时错误:
start...
result: 190ad871-aae2-48ac-a18e-962cd2d537b7
False
Traceback (most recent call last):
File "/workspace/celery_demo/venv3/lib/python3.6/site-packages/celery/backends/amqp.py", line 146, in wait_for
on_interval=on_interval)
File "/workspace/celery_demo/venv3/lib/python3.6/site-packages/celery/backends/amqp.py", line 223, in consume
conn, consumer, timeout, on_interval)[task_id]
File "/workspace/celery_demo/venv3/lib/python3.6/site-packages/celery/backends/amqp.py", line 202, in drain_events
raise socket.timeout()
socket.timeout
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "api.py", line 14, in <module>
print(result.get(timeout=1))
File "/workspace//celery_demo/venv3/lib/python3.6/site-packages/celery/result.py", line 226, in get
on_message=on_message,
File "/workspace/celery_demo/venv3/lib/python3.6/site-packages/celery/backends/base.py", line 496, in wait_for_pending
no_ack=no_ack,
File "/workspace/celery_demo/venv3/lib/python3.6/site-packages/celery/backends/amqp.py", line 148, in wait_for
raise TimeoutError('The operation timed out.')
celery.exceptions.TimeoutError: The operation timed out.
2.5 查看结果
查看app.log中的日志
[2019-04-20 20:53:59,327] [INFO] Connected to amqp://test:**@127.0.0.1:5672/myvhost
[2019-04-20 20:53:59,345] [INFO] mingle: searching for neighbors
[2019-04-20 20:54:00,379] [INFO] mingle: all alone
[2019-04-20 20:54:00,441] [INFO] [email protected] ready.
[2019-04-20 20:55:46,498] [INFO] Received task: demo_task.add[1c8d47bd-449d-4bd9-b4db-819777081d23]
[2019-04-20 20:55:46,561] [INFO] Task demo_task.add[1c8d47bd-449d-4bd9-b4db-819777081d23] succeeded in 0.060540573904290795s: 3
3.任务过期
调用apply_async
时使用参数expires
,则表示任务有超时时间,超过这个时间后,task不会得到执行。
result = add.apply_async((1, 2), expires=10)
当worker没有启动或者其他异常情况下,会出现任务超时,不被执行。
查看日志,可以看到task过期,不会被执行。
[2019-04-20 21:32:45,828] [INFO] Received task: demo_task.add[12691c27-3f2f-4c96-9b4e-54636e20d0eb] expires:[2019-04-20 13:31:51.030013+00:00]
[2019-04-20 21:32:45,829] [INFO] Discarding revoked task: demo_task.add[12691c27-3f2f-4c96-9b4e-54636e20d0eb]