介绍
Celery 是一个基于python开发的分布式异步消息任务队列，通过它可以轻松的实现任务的异步处理， 如果你的业务场景中需要用到异步任务，就可以考虑使用celery。
异步任务介绍
在写项目过程中经常会遇到一些耗时的任务, 比如:发送邮件、发送短信等等~。这些操作如果都同步执行耗时长对用户体验不友好,在这种情况下就可以把任务放在后台异步执行
celery就是用于处理异步任务的框架,celery能完成的功能远不止异步任务,还有一个很常用的功能定时任务.
架构
Celery包含如下组件：
Celery Beat：任务调度器，Beat进程会读取配置文件的内容，周期性地将配置中到期需要执行的任务发送给任务队列。
Celery Worker：执行任务的消费者，通常会在多台服务器运行多个消费者来提高执行效率。
Broker：消息代理，或者叫作消息中间件，接受任务生产者发送过来的任务消息，存进队列再按序分发给任务消费方（通常是消息队列或者数据库）。
Producer：调用了Celery提供的API、函数或者装饰器而产生任务并交给任务队列处理的都是任务生产者。
Result Backend：任务处理完后保存状态信息和结果，以供查询。Celery默认已支持Redis、RabbitMQ、MongoDB、Django ORM、SQLAlchemy等方式。
特点
- 简单：一单熟悉了celery的工作流程后，配置和使用还是比较简单的
- 高可用：当任务执行失败或执行过程中发生连接中断，celery 会自动尝试重新执行任务
- 快速：一个单进程的celery每分钟可处理上百万个任务
- 灵活： 几乎celery的各个组件都可以被扩展及自定制
多任务执行方式
后台多任务方式：
启动：celery multi start w1 -A proj -l info
重启：celery multi restart w1 -A proj -l info
停止：celery multi stop w1 -A proj -l info
任务执行完毕后才退出：celery multi stopwait w1 -A proj -l info
定时任务
celery支持定时任务，设定好任务的执行时间，celery就会定时自动帮你执行， 这个定时任务模块叫celery beat
方式一：函数方式
from celery import Celery
from celery.schedules import crontab
app = Celery()
@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')
# Calls test('world') every 30 seconds
sender.add_periodic_task(30.0, test.s('world'), expires=10)
# Executes every Monday morning at 7:30 a.m.
sender.add_periodic_task(
crontab(hour=7, minute=30, day_of_week=1),
test.s('Happy Mondays!'),
)
@app.task
def test(arg):
print(arg)
方式二：配置文件形式
app.conf.beat_schedule = {
'add-every-30-seconds': {
'task': 'tasks.add',
'schedule': 30.0,
'args': (16, 16)
},
}
app.conf.timezone = 'UTC'
任务添加好了，需要让celery单独启动一个进程来定时发起这些任务， 注意， 这里是发起任务，不是执行，这个进程只会不断的去检查你的任务计划， 每发现有任务需要执行了，就发起一个任务调用消息，交给celery worker去执行
启动任务调度器celery beat：celery -A periodic_task beat
此时还差一步，就是还需要启动一个worker,负责执行celery beat发起的任务
启动celery worker来执行任务：celery -A periodic_task worker
常用的定时任务配置
| Example | Meaning |
|---|---|
| crontab() | 默认是每分钟 |
| crontab(minute=0, hour=0) | 每天0点执行 |
| crontab(minute=0, hour='*/3') | 每隔3个小时执行 |
| crontab(minute=0,hour='0,3,6') | 0点、3点、6点执行 |
| crontab(minute='*/15') | 每隔15分钟 |
| crontab(day_of_week='sunday') | 每逢周日的每一分钟执行 |
| crontab(minute='',hour='',day_of_week='sun') | 与上一个意义一致 |
| crontab(minute='*/10',hour='3,17,22',day_of_week='fri') | 在周五3-4点、17-18点、22-23点之间每隔10分钟执行 |
| crontab(minute=0,hour='/2,/3') | 能被2，3整除的小时执行 |
| crontab(minute=0, hour='*/5') | 能被5整除的小时执行 |
| crontab(minute=0, hour='*/3,8-17') | 能被3整除或者8-17之间的小时执行 |
| crontab(0, 0,day_of_month='2') | 每个月的第二天 |
| crontab(0, 0,day_of_month='2-30/3') | 偶数天执行 |
| crontab(0, 0,day_of_month='1-7,15-21') | 每个月的第一周和第三周 |
| crontab(0, 0,day_of_month='11',month_of_year='5') | 每年的5月11日执行 |
| crontab(0, 0,month_of_year='*/3') | 在每个季度的第一个月执行 |
上面能满足你绝大多数定时任务需求了，甚至还能根据潮起潮落来配置定时任务
最佳实践之与django结合
django 可以轻松跟celery结合实现异步任务，只需简单配置即可
项目目录
- proj /
- proj / __init__ . py
- proj / settings . py
- proj / urls . py
- manage . py
proj/proj/celery.py
from __future__ import absolute_import, unicode_literals
import os
from celery import Celery
# 为celery程序设置默认的Django设置模块
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'proj.settings')
app = Celery('proj')
# 从配置文件中引入celery相关参数，每个参数使用CELERY_开头
app.config_from_object('django.conf:settings', namespace='CELERY')
# 从每个app中的tasks.py文件加载异步任务
app.autodiscover_tasks()
@app.task(bind=True)
def debug_task(self):
print('Request: {0!r}'.format(self.request))
proj/proj/__init__.py
from __future__ import absolute_import, unicode_literals
# 确保celery文件一定被引用
from .celery import app as celery_app
__all__ = ['celery_app']
proj/xxx/tasks.py
from __future__ import absolute_import, unicode_literals
from celery import shared_task
@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)
proj/xxx/views.py
from django.shortcuts import render,HttpResponse
from bernard import tasks
def task_test(request):
res = tasks.add.delay(228,24)
print("start running task")
print("async task res",res.get() )
return HttpResponse('res %s'%res.get())
在django中使用计划任务功能
安装：pip install django-celery-beat
加载：在settings.py中INSTALLED_APPS加入django_celery_beat
生成相关数据库：python manage.py migrate
开启定时任务：celery -A proj beat -l info -S django
结果：
此时启动你的celery beat 和worker，会发现每隔2分钟，beat会发起一个任务消息让worker执行scp_task任务
注意，经测试，每添加或修改一个任务，celery beat都需要重启一次，要不然新的配置不会被celery beat进程读到
celery flower
(1).查看任务历史,任务具体参数,开始时间等信息。
(2).提供图表和统计数据。
(3).实现全面的远程控制功能, 包括但不限于 撤销/终止任务, 关闭重启 worker, 查看正在运行任务。
(4).提供一个 HTTP API , 方便集成。
终端执行：celery flower --broker=redis://localhost:6379/6