Celery分布式任务

Posted 2020-12-30 czlong

Celery分布式任务

celery call把任务给一个组件，组件交给rabiitmq放到队列broker，队列返回任务id给celery组件再给call，
任务完成时call拿着id通过celery去rabbitmq取。broker发任务给worker

1.Celery有以下优点：

简单：一单熟悉了celery的工作流程后，配置和使用还是比较简单的
高可用：当任务执行失败或执行过程中发生连接中断，celery 会自动尝试重新执行任务
快速：一个单进程的celery每分钟可处理上百万个任务
灵活： 几乎celery的各个组件都可以被扩展及自定制

2.Celery安装使用
Celery的默认broker是RabbitMQ, 仅需配置一行就可以
broker_url = ‘amqp://guest:[email protected]:5672//‘
使用Redis做broker也可以
安装redis组件
pip install -U "celery[redis]"
app.conf.broker_url = ‘redis://localhost:6379/0‘ redis://:[email protected]:port/db_number
配置一下把任务结果存在哪:app.conf.result_backend = ‘redis://localhost:6379/0‘

3.开始使用Celery啦　　
pip install celery
软链：ln -s /usr/local/python3/bin/celery /usr/bin/celery
移除软链：rm + name

4.简单任务
创建一个任务文件
from celery import Celery

app = Celery(‘tasks‘, # app的名字
broker=‘redis://:[email protected]:6379/0‘)
# backend=‘redis://localhost‘)

@app.task
def add(x, y): # 加装饰器，这是worker可以执行的一个任务
print("running...", x, y)
return x + y

启动Celery Worker来开始监听并执行任务
celery -A tasks worker --loglevel=info

注：window上报错，pip install eventlet celery -A <mymodule> worker -l info -P eventlet
调用任务
再打开一个终端， 进行命令行模式，调用任务　　
from tasks import add
add.delay(4, 4)

result = add.delay(4, 4) #赋值变量后能接收值

发布任务后，多个worker时只能由其中一个抢到并执行，linux和window的worker一样

任务状态：
result.ready()
取结果时，设置超时时间
result.get(timeout=1) #如果还没处理完，会报错
propagate 参数覆盖get的报错，而是返回任务的报错
result.get(propagate=False)
回撤
result.traceback
二、在项目中使用celery
1.目录格式
proj/__init__.py
/celery.py #配置连接
/tasks.py

2. proj/celery.py内容
from __future__ import absolute_import, unicode_literals
# 默认导入当前同名文件，这里导入后从python包的绝对路径导入，支持unicode兼容
from celery import Celery

app = Celery(‘proj‘,
broker=‘redis://:[email protected]:6379/0‘,
backend=‘redis://:[email protected]:6379/0‘,
include=[‘celery_project.tasks‘]) # 要执行的任务文件模板位置

# Optional configuration, see the application user guide.
app.conf.update(
result_expires=3600,
)

if __name__ == ‘__main__‘:
app.start()

3. proj/tasks.py中的内容
from __future__ import absolute_import, unicode_literals
from .celery import app # .表示不绝对导入
@app.task
def add(x, y):
return x + y

@app.task
def mul(x, y):
return x * y

@app.task
def xsum(numbers):
return sum(numbers)

4.启动celery的worker
elery -A celery_project worker -l info -P eventlet
5.发布任务
进入项目所在文件夹，cmd导入运行
# 这时不能从文件导入函数，只能在上一层文件夹导入文件，相对路径.xxx不能作为函数主入口
from celery_project import tasks
t = tasks.add.delay(2,3)
t.get()

6.后台进程（仅linux）
启动
celery multi start w1 -A proj -l info
停止
celery multi stop w1 -A proj -l info
celery multi stopwait w1 -A proj -l info # 等待任务结束才停止
查看进程
ps -ef |grep celery

三、celery定时任务
celery支持定时任务，设定好任务的执行时间，celery就会定时自动帮你执行， 这个定时任务模块叫celery beat
celery beat会循环监听定时任务，负责交给worker
1.编写定时任务，加入app的include
from __future__ import absolute_import, unicode_literals
from .celery import app # .表示不绝对导入
from celery.schedules import crontab

@app.on_after_configure.connect # 连接后启动任务
def setup_periodic_tasks(sender, **kwargs):
# Calls test(‘hello‘) every 10 seconds.
sender.add_periodic_task(10.0, test.s(‘hello‘), name=‘add every 10‘) # 每隔10秒执行，.s相当于delay传参

# Calls test(‘world‘) every 30 seconds
sender.add_periodic_task(30.0, test.s(‘world‘), expires=10) # 结果保存10秒

# Executes every Monday morning at 7:30 a.m.
sender.add_periodic_task(
crontab(hour=7, minute=30, day_of_week=1), # crontab是linux的定时模块 crontab -e 编写定时文件 分钟 小时 日 月 星期几 command
test.s(‘Happy Mondays!‘),
)
@app.task
def test(arg):
print(‘run fun‘,arg)
2.启动celery beat
celery -A celery_project.periodic beat -l info # 在根目录执行，或者理解为任务的命名
# beat就会根据定时时间wake up来把任务交给worker
# 指定beat最后运行时间要存储在哪个位置，默认当前目录
celery -A celery_project.periodic beat -s /home/celery/var/run/celerybeat-schedule


配置文件添加定时任务
app.conf.beat_schedule = {
‘add-every-30-seconds‘: {
‘task‘: ‘celery_project.tasks.add‘, # 一个app下该任务的命名
‘schedule‘: 30.0,
‘args‘: (16, 16)
},
}
app.conf.timezone = ‘UTC‘

更复杂的定时配置　　
用crontab功能，跟linux自带的crontab功能是一样的，可以个性化定制任务执行时间
from celery.schedules import crontab

app.conf.beat_schedule = {
# Executes every Monday morning at 7:30 a.m.
‘add-every-monday-morning‘: {
‘task‘: ‘celery_project.tasks.add‘,
‘schedule‘: crontab(hour=7, minute=30, day_of_week=1),
‘args‘: (16, 16),
},
}

四、与django结合使用
django 可以轻松跟celery结合实现异步任务，只需简单配置即可

1.定义celery项目实例
from __future__ import absolute_import, unicode_literals
import os
from celery import Celery

# set the default Django settings module for the ‘celery‘ program.
os.environ.setdefault(‘DJANGO_SETTINGS_MODULE‘, ‘MyBlog.settings‘)

app = Celery(‘task‘) # 项目名

# Using a string here means the worker don‘t have to serialize
# the configuration object to child processes.
# - namespace=‘CELERY‘ means all celery-related configuration keys
# should have a `CELERY_` prefix.
app.config_from_object(‘django.conf:settings‘, namespace=‘CELERY‘) # 使用django的session

# Load task modules from all registered Django app configs.
app.autodiscover_tasks() # 项目下所有app的celery任务发现

@app.task(bind=True)
def debug_task(self):
print(‘Request: {0!r}‘.format(self.request))
2.在django的settings配置celery的连接
#for celery
CELERY_BROKER_URL = ‘redis://:[email protected]:6379/0‘
CELERY_RESULT_BACKEND = ‘redis://:[email protected]:6379/0‘

3.设置celery随django项目一起启动
from __future__ import absolute_import, unicode_literals

# This will make sure the app is always imported when
# Django starts so that shared_task will use this app.
from .celery import app as celery_app

__all__ = [‘celery_app‘]

4.具体的app里在tasks.py写要执行的任务，以被worker执行
# Create your tasks here
from __future__ import absolute_import, unicode_literals
from celery import shared_task # 多个app共享worker

@shared_task
def add(x,y):
return x + y

@shared_task
def mul(x, y):
return x * y

@shared_task
def xsum(numbers):
return sum(numbers)
5.在app里写django的views，然后调用上面写好的celery task，这样访问页面时自动发布任务
from django.http import HttpResponse
from celery.result import AsyncResult
from .celery_task import add

task_id=None
def celery_pub(request):
task = add.delay(22,23)
global task_id
task_id = task.id
# 拿到任务id即可返回，不用等待get获取值，以后再调用id拿值
return HttpResponse(task.id)

def celery_get(request):
global task_id
result = AsyncResult(id=task_id)
return HttpResponse(result.get())

6.进入Django根目录，通过cmd启动worker以接收并执行任务
celery -A 项目名 worker -l info

# 访问页面拿到id，再通过另一个请求拿到结果。

五、Django定时任务

1.添加django_celery_beat模块到django的INSTALLED_APPS

INSTALLED_APPS = (
...,
‘django_celery_beat‘,
)
2.创建数据库中的表格
python manage.py migrate
3.在admin中添加定时任务（已注册的任务）到数据库

4.启动beat去数据库取任务
celery -A MyBlog beat -l info -S django

5.启动worker即可