celery+rabbitmq基本使用

Posted 2022-01-26 lianyeah

版本：

　　(celery==4.3，rabbitmq==3.7)

一.使用rabbitmq

# 安装rabbitMQ(mac)
brew install rabbitmq
# 配置环境变量(.bash_profile或.profile,注意路径)
export RABBIT_HOME=/usr/local/Cellar/rabbitmq/3.7.15
export PATH=$PATH:$RABBIT_HOME/sbin

# 设置RabbitMQ:创建一个用户，一个虚拟主机并设置权限,(需要先启动服务)
sudo rabbitmqctl add_user myuser mypassword
sudo rabbitmqctl add_vhost myvhost  # 虚拟主机
sudo rabbitmqctl set_user_tags myuser mytag  # administrator
sudo rabbitmqctl set_permissions -p myvhost myuser ".*" ".*" ".*"

#启动服务
sudo rabbitmq-server
# 后台运行
sudo rabbitmq-server -detached
# 不要kill(1)来停止服务器
sudo rabbitmqctl stop

二.celery使用

# 安装
pip install celery

　　应用

# tasks.py
from celery import Celery

app = Celery(
    ‘tasks‘,  # 当前模块的名字
    broker=‘amqp://[email protected]:port//‘  # 消息队列的url
)

@app.task
def add(x, y):
    return x + y

　　运行worker

celery -A tasks worker --loglevel=info

　　调用任务

# delay
from tasks import add
add.delay(arg1,arg2,kwarg1=‘x‘,kwarg2=‘y‘)
add.delay(*args, **kwargs).apply_async(args, kwargs)

# apply_async
task.apply_async(args=[arg1,arg2], kwargs=‘kwargs‘:‘x‘,‘kwargs‘:‘y‘)
tasks.apply_async((arg,), ‘kwarg‘: value)
# 从现在起10秒内执行
tasks.apply_async(countdown=10) 
# 从现在起10秒内执行,使用指定eta
tasks.apply_async(eta=now + timedelta(seconds=10))
# 从现在起一分钟后执行，但在2分钟后过期
tasks.apply_async(countdown=60, expires=120)
# 在2天后到期，设置使用datetime对象
T.apply_async(expires=now + timedelta(days=2))

# send_task：任务未在当前进程中注册
app.send_task(‘任务‘, args=[arg,], queue=‘default‘)

# signature用于传递任务调用签名的对象(例如通过网络发送),并且它们也支持calling api
task.s(arg1,arg2,kwarg1=‘x‘,kwargs2=‘y‘).apply_async()