python之celery队列模块

一、celery队列简介

Celery 是一个基于python开发的分布式异步消息任务队列，通过它可以轻松的实现任务的异步处理，如果你的业务场景中需要用到异步任务，就可以考虑使用celery.

1.1使用场景

1.你想对100台机器执行一条批量命令，可能会花很长时间，但你不想让你的程序等着结果返回，而是给你返回一个任务ID,你过一段时间只需要拿着这个任务id就可以拿到任务执行结果，在任务执行ing进行时，你可以继续做其它的事情。

2.你想做一个定时任务，比如每天检测一下你们所有客户的资料，如果发现今天是客户的生日，就给他发个短信祝福

1.2celery原理

Celery 在执行任务时需要通过一个消息中间件来接收和发送任务消息，以及存储任务结果，一般使用rabbitMQ or Redis 或者是数据库来存放消息的中间结果

Celery优点:

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

Celery缺点:

　　 1.目前只能在Linux系统上有较好的支持

Celery工作流程图:

1. Celery Beat：任务调度器，Beat进程会读取配置文件的内容，周期性地将配置中到期需要执行的任务发送给任务队列。

2. Celery Worker：执行任务的消费者，通常会在多台服务器运行多个消费者来提高执行效率。

3. Broker：消息代理，或者叫作消息中间件，接受任务生产者发送过来的任务消息，存进队列再按序分发给任务消费方（通常是消息队列或者数据库）。

4. Producer：调用了Celery提供的API、函数或者装饰器而产生任务并交给任务队列处理的都是任务生产者。

5. Result Backend：任务处理完后保存状态信息和结果，以供查询。Celery默认已支持Redis、RabbitMQ、MongoDB、Django ORM、SQLAlchemy等方式。

Celery的架构图如图所示：

在传统的web应用中,Django的web页面通过url的映射到view,view再执行方法,如果方法需要调用大量的脚本,执行大量的任务,页面就会阻塞,如果在项目中使用Celery队列.首先用户的任务会被celery放到broker中进行中转,然后将任务分为一个个的task来执行,由于celery是异步机制,所以会直接给用户返回task_id,页面拿到task_id就可以执行后续的操作,比如查看任务进度,暂停任务,而无需等待所有任务全部执行完毕,才能看到页面。

二、celery安装

2.1安装:

　　1.在linux(ubuntu)系统上首先安装Celery队列

　　　　pip3 install Celery

2.在linux安装redis

　　　　sudo apt-get install redis-server

3.在linux上安装redis-celery中间件

　　　　pip3 install -U "celery[redis]"

4.启动redis

sudo /etc/init.d/redis-server start

三、celery使用

3.1简单使用

命名为tasks.py

from celery import Celery

app = Celery('tasks',

             broker='redis://localhost',

             backend='redis://localhost')

@app.task

def add(x,y):

    print("running...",x,y)

    return x+y

启动监听并开始执行该服务

celery -A tasks worker -l debug

在开启一个终端进行测试任务

进入python环境

from tasks import add

t = add.delay(3,3) #此时worker会生成一个任务和任务id

t.get() #获取任务执行的结果

t.get(propagate=False) #如果任务执行中出现异常,在client端不会异常退出

t.ready()#查看任务是否执行完毕

t.traceback #打印异常详细信息

3.2项目中创建celery

在当前的目录下创建文件夹celery_pro

mkdir celery_pro

在此目录下创建两个文件

目录结构:

celery_proj

    /__init__.py

    /celery.py

    /tasks.py

celery.py(定义了celery的一些元信息)

from __future__ import absolute_import, unicode_literals

from celery import Celery

app = Celery('proj',

             broker='redis://localhost',   #消息中间接收

             backend='redis://localhost', #消息结果存放

             include=['proj.tasks'])          #执行任务的文件   

# Optional configuration, see the application user guide.

app.conf.update(

    result_expires=3600,

)

if __name__ == '__main__':

    app.start()

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)

启动worker

celery -A celery_pro worker -l debug

再另一个窗口打开python命令模式进行测试

from celery_pro import tasks

t = tasks.add.delay(3,4)

t.get()

Celery的分布式:多启动worker就可以自动实现负载均衡,无需手动管理

Celery永驻后台(开启&重启&关闭)

celery multi start w1 -A celery_pro -l info  #开启后台celery任务

celery  multi restart w1 -A proj -l info #重启该服务

celery multi stop w1 -A proj -l info #关闭该服务

3.3 celery定时任务

在celery_pro文件夹下创建periodic_tasks.py

目录结构:

celery_proj

    /__init__.py

    /celery.py

    /tasks.py

    /periodic_tasks.py

文件内容如下:

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')

    # 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=21, minute=42, day_of_week=5),

        test.s('Happy Mondays!'),

    )

@app.task

def test(arg):

    print(arg)

修改celery.py,加入periodic_task.py

from __future__ import absolute_import, unicode_literals

from celery import Celery

app = Celery('proj',

             broker='redis://localhost',

             backend='redis://localhost',

             include=['celery_pro.tasks','celery_pro.periodic_tasks'])

# Optional configuration, see the application user guide.

app.conf.update(

    result_expires=3600,

)

if __name__ == '__main__':

    app.start()

在服务端启动 celery -A celery_pro worker -l debug

在客户端启动 celery -A celery_pro.periodic_tasks beat -l debug

在服务端如果看到打印的hell ,world说明定时任务配置成功

上面是通过调用函数添加定时任务，也可以像写配置文件一样的形式添加，下面是每30s执行的任务

在celery.py中添加

app.conf.beat_schedule = {

    'add-every-30-seconds': {

        'task': 'cerely_pro.tasks.add', #执行的具体方法

        'schedule': 5.5,  #每秒钟执行

        'args': (16, 16)   #执行的具体动作的参数

    },

}

app.conf.timezone = 'UTC'

更多定制

上面的定时任务比较简单，但如果你想要每周一三五的早上8点给你发邮件怎么办呢?用crontab功能，跟linux自带的crontab功能是一样的，可以个性化定制任务执行时间

rom celery.schedules import crontab

app.conf.beat_schedule = {

    #在每周一早上7:30执行

    'add-every-monday-morning': {

        'task': 'celery_pro.tasks.add',

        'schedule': crontab(hour=7, minute=30, day_of_week=1),

        'args': (16, 16),

    },

还有更多定时配置方式如下：

Example	Meaning
`crontab()`	Execute every minute.
`crontab(minute=0, hour=0)`	Execute daily at midnight.
`crontab(minute=0, hour='*/3')`	Execute every three hours: midnight, 3am, 6am, 9am, noon, 3pm, 6pm, 9pm.
`crontab(minute=0,` `hour='0,3,6,9,12,15,18,21')`	Same as previous.
`crontab(minute='*/15')`	Execute every 15 minutes.
`crontab(day_of_week='sunday')`	Execute every minute (!) at Sundays.
`crontab(minute='',` `hour='',day_of_week='sun')`	Same as previous.
`crontab(minute='*/10',` `hour='3,17,22',day_of_week='thu,fri')`	Execute every ten minutes, but only between 3-4 am, 5-6 pm, and 10-11 pm on Thursdays or Fridays.
`crontab(minute=0,hour='/2,/3')`	Execute every even hour, and every hour divisible by three. This means: at every hour except: 1am, 5am, 7am, 11am, 1pm, 5pm, 7pm, 11pm
`crontab(minute=0, hour='*/5')`	Execute hour divisible by 5. This means that it is triggered at 3pm, not 5pm (since 3pm equals the 24-hour clock value of “15”, which is divisible by 5).
`crontab(minute=0, hour='*/3,8-17')`	Execute every hour divisible by 3, and every hour during office hours (8am-5pm).
`crontab(0, 0,day_of_month='2')`	Execute on the second day of every month.
`crontab(0, 0,` `day_of_month='2-30/3')`	Execute on every even numbered day.
`crontab(0, 0,` `day_of_month='1-7,15-21')`	Execute on the first and third weeks of the month.
`crontab(0, 0,day_of_month='11',` `month_of_year='5')`	Execute on the eleventh of May every year.
`crontab(0, 0,` `month_of_year='*/3')`	Execute on the first month of every quarter.

3.4Celery+Django实现异步任务分发

1.在setting.py的文件同一级别创建celery.py

 from __future__ import absolute_import, unicode_literals

 import os

 from celery import Celery

 # 设置Django的环境变量

 os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'PerfectCRM.settings')

 #设置app的默认处理方式,如果不设置默认是rabbitMQ

 app = Celery('proj',

              broker='redis://localhost',

              backend='redis://localhost'

 )

 #配置前缀

 app.config_from_object('django.conf:settings', namespace='CELERY')

 #自动扫描app下的tasks文件

 app.autodiscover_tasks()

 @app.task(bind=True)

 def debug_task(self):

     print('Request: {0!r}'.format(self.request))

2.修改当前目录下的__init__文件

 from __future__ import absolute_import, unicode_literals

 #启动时检测celery文件

 from .celery import app as celery_app

 __all__ = ['celery_app']

3.在app下新增tasks文件,写要执行的任务

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)

4.在另一个app下新增tasks文件

 from __future__ import absolute_import, unicode_literals

 from celery import shared_task

 import time,random

 @shared_task

 def randnum(start, end):

     time.sleep(3)

     return random.ranint(start,end)

5.在views下增加处理逻辑

 from crm import tasks

 from celery.result import AsyncResult

 import random

 #计算结果

 def celery_call(request):

     randnum =random.randint(0,1000)

     t = tasks.add.delay(randnum,6)

     print('randum',randnum)

     return HttpResponse(t.id)

 #获取结果

 def celery_result(request):

     task_id = request.GET.get('id')

     res = AsyncResult(id=task_id)

     if res.ready():

         return HttpResponse(res.get())

     else:

         return HttpResponse(res.ready())

6.测试

首先启动Django,从web端输入url调用celery_call方法

例:http://192.168.17.133:9000/crm/celery_call,此方法会返回一个task_id(41177118-3647-4830-b8c8-7be76d9819d7)

带着这个task_id 访问http://192.168.17.133:9000/crm/celery_result?id=41177118-3647-4830-b8c8-7be76d9819d7如果可以看到结果说明配置成功

Dnango+Celery实现定时任务

安装Django,Celery中间件

　　　　pip3 install django-celery-beat

在Django的settings文件中,新增app,名称如下

　　 INSTALLED_APPS = (

　　　　.....,

　　　　'django_celery_beat', #新增的app

　　　　)

输入命令

　　　　python manage.py migrate #创建与Django有关定时计划任务的新表

通过celery beat开启定时任务

　　　　celery -A PrefectCRM beat -l info -S django

启动Django服务,进入admin配置页面

　　　　python3 manager.py runserver 0.0.0.0:9000

　　　　并设置settings.py中的

　　　　ALLOW_HOSTS=['*']

可以在原有业务表的基础之上看到新的三张表

后记:经测试，每添加或修改一个任务，celery beat都需要重启一次，要不然新的配置不会被celery beat进程读到

巴特西