一.Celery介绍和基本使用
1.什么是Celery：
是一个基于python开发的分布式异步消息任务队列，通过它可以轻松的实现任务的异步处理，如果你的业务场景中需要用到异步任务，就可以考虑使用celery，Celery在执行任务时需要通过一个消息中间件来接收和发送任务消息，以及存储任务结果， 一般使用rabbitMQ or Redis
2.Celery有以下优点：
(1)简单：一单熟悉了celery的工作流程后，配置和使用还是比较简单的
(2)高可用：当任务执行失败或执行过程中发生连接中断，celery 会自动尝试重新执行任务
(3)快速：一个单进程的celery每分钟可处理上百万个任务
(4)灵活： 几乎celery的各个组件都可以被扩展及自定制
3.Celery基本工作流程
(1)请求把任务给---celerv组件，他把任务给放到rabbitmq(broker中间商)同时生成一个任务id
(2)worker(执行任务的节点)从rabbitmq(broker中间商)里取数据把执行好的任务放到rabbitmq(broker中间商)里面
(3)请求再通过celerv组件去rabbitmq(broker中间商)里取结果
4.Celery的基本使用
(1)创建一个celery application用来定义你的任务列表
#!/bin/env python3 #_*_coding:utf-8_*_ from celery import Celery app = Celery('tasks', #给app取得名字 broker='redis://:123456@192.168.1.232', #连接rabbitmq(broker中间商) backend='redis://:123456@192.168.1.232' #把结果写到redis里 ) #加装饰器代表这是worker可以执行的一个任务 @app.task #定义加法任务 def add(x,y): print("running...",x,y) return x+y
(2)启动Celery Worker来开始监听并执行任务
命令：celery -A celery_test worker -l debug
(3)开一个终端，调用任务
>>> from celery_test import add >>> t = add.delay(18,18)
Celery Worker日志打印：
[2019-06-17 12:34:09,503: WARNING/ForkPoolWorker-1] running...
[2019-06-17 12:34:09,503: WARNING/ForkPoolWorker-1] 18
[2019-06-17 12:34:09,503: WARNING/ForkPoolWorker-1] 18
(4)在执行
>>> t.get()
返回：36
Celery Worker日志打印：
[2019-06-17 12:34:09,509: INFO/ForkPoolWorker-1] Task celery_test.add[4b9e24ef-b148-4e81-a019-d9b28309b93d] succeeded in 0.007139017805457115s: 36
二.在项目中如何使用celery
1.可以把celery配置成一个应用目录格式如下：
proj/__init__.py proj/celery.py #celery的配置 proj/tasks.py #任务1 proj/tasks2.py #任务2
(1)celery配置：
from __future__ import absolute_import, unicode_literals #from __future__ import absolute_import是从python包的绝对路径里取import我们安装的包而不是当前目录 from celery import Celery #默认import当前目录的Celery，当前目录没有 app = Celery('proj', broker='redis://:123456@192.168.1.232', #连上rabbitMQ的参数 backend='redis://:123456@192.168.1.232',#连上rabbitMQ的参数 include=['proj.tasks','proj.tasks2']) #定义proj目录下的tasks脚本和proj目录下的tasks2脚本 #可以给app设置参数 app.conf.update( result_expires=3600, #所有任务结果一个小时之内结果没被取走就没了 ) if __name__ == '__main__': app.start()
(2)任务1：
from __future__ import absolute_import, unicode_literals from .celery import app #导入当前目录下celery.py里的app，用装饰器 #相加 @app.task def add(x, y): return x + y
(3)任务2：
from __future__ import absolute_import, unicode_literals from .celery import app #导入当前目录下celery.py里的app，用装饰器 #求和 @app.task def xsum(numbers): return sum(numbers)
(3)退出上层目录启动：celery -A proj worker -l debug
(4)开一个终端，调用任务
>>> from proj import tasks2,tasks >>> t2 = tasks2.xsum.delay([3,45,5,6,7,4,88,21,5]) >>> t1 = tasks.add.delay(2,5) >>> t1.get()
返回：7
>>> t2.get()
返回：184
2.后台启动worker
(1)启动1个w1任务：celery multi start w1 -A celery_django -l info
返回：
celery multi v4.3.0 (rhubarb)
> Starting nodes...
> w1@localhost: OK
(2)启动1个w2任务：celery multi start w2 -A celery_django -l info
返回：
celery multi v4.3.0 (rhubarb)
> Starting nodes...
> w2@localhost: OK
(3)停止w2任务：celery multi stop w2 -A proj -l info
返回：
celery multi v4.3.0 (rhubarb)
> Stopping nodes...
> w2@localhost: TERM -> 11221
三.Celery定时任务
celery支持定时任务，设定好任务的执行时间，celery就会定时自动帮你执行， 这个定时任务模块叫celery beat
1.通过计划任务实现每隔10秒，30秒，一周执行任务
(1)celery配置：
from __future__ import absolute_import, unicode_literals #from __future__ import absolute_import是从python包的绝对路径里取import我们安装的包而不是当前目录 from celery import Celery #默认import当前目录的Celery，当前目录没有 app = Celery('proj', broker='redis://:123456@192.168.1.232', #连上rabbitMQ的参数 backend='redis://:123456@192.168.1.232',#连上rabbitMQ的参数 include=['proj.periodic_task']) #定义proj目录下的periodic_task脚本 #可以给app设置参数 app.conf.update( result_expires=3600, #所有任务结果一个小时之内结果没被取走就没了 ) if __name__ == '__main__': app.start()
(2)定时任务：periodic_task.py
from __future__ import absolute_import, unicode_literals from celery.schedules import crontab from .celery import app #只要脚本一启动立刻执行这个函数,这个函数自动有两个参数sender(添加任务), **kwargs @app.on_after_configure.connect def setup_periodic_tasks(sender, **kwargs): #每隔十秒钟执行test这个函数，.s是给test这个函数传的参数 sender.add_periodic_task(10.0, test.s('hello'), name='add every 10') #name='add every 10'任务名 #每三十秒执行test这个函数，.s是给test这个函数传的参数 sender.add_periodic_task(30.0, test.s('xixi'), expires=10) #expires=10任务结果保存10秒钟 #每周一早上七点半执行test这个函数，.s是给test这个函数传的参数 sender.add_periodic_task( crontab(hour=7, minute=30, day_of_week=1), test.s('Happy Mondays!'), ) @app.task def test(arg): print(arg)
(3)启动worker：celery -A proj worker -l debug
(4)启动任务调度器：celery -A proj.periodic_task beat -l debug
十秒返回：
[2019-06-18 14:01:54,236: WARNING/ForkPoolWorker-3] hello
[2019-06-18 14:01:54,243: INFO/ForkPoolWorker-3] Task proj.periodic_task.test[89b349b8-3c19-4968-a09b-214a72164cee] succeeded in 0.007367603946477175s: None
[2019-06-18 14:02:14,192: INFO/MainProcess] Received task: proj.periodic_task.test[1803e491-5a41-4fe1-bfe2-18b8335ae7d5]
三十秒返回：
[2019-06-18 14:03:14,192: WARNING/ForkPoolWorker-3] xixi
[2019-06-18 14:03:14,192: DEBUG/MainProcess] Task accepted: proj.periodic_task.test[8fe2e9f2-e07e-4f87-bd80-daa4d3c64605] pid:27865
[2019-06-18 14:03:14,194: INFO/ForkPoolWorker-3] Task proj.periodic_task.test[8fe2e9f2-e07e-4f87-bd80-daa4d3c64605] succeeded in 0.0014079520478844643s: None
[2019-06-18 14:03:34,196: INFO/MainProcess] Received task: proj.periodic_task.test[80b7fcfa-b1e6-4c11-8390-3e3a0cc68f4e]
2.通过计划任务实现每五秒做一次相加操作
(1)celery配置：celery.py
from __future__ import absolute_import, unicode_literals #from __future__ import absolute_import是从python包的绝对路径里取import我们安装的包而不是当前目录 from celery import Celery #默认import当前目录的Celery，当前目录没有 app = Celery('proj', broker='redis://:123456@192.168.1.232', #连上rabbitMQ的参数 backend='redis://:123456@192.168.1.232', #连上rabbitMQ的参数 include=['proj.tasks','proj.periodic_task']) #定义proj目录下的periodic_task脚本 #可以给app设置参数 app.conf.update( result_expires=3600, #所有任务结果一个小时之内结果没被取走就没了 ) if __name__ == '__main__': app.start()
(2)相加任务：tasks.py
from __future__ import absolute_import, unicode_literals from .celery import app #导入当前目录下celery.py里的app，用装饰器 #相加 @app.task def add(x, y): return x + y
(3)计划任务：
from __future__ import absolute_import, unicode_literals from celery.schedules import crontab from .celery import app app.conf.beat_schedule = { '每隔五秒执行相加': { 'task': 'proj.tasks.add', 'schedule': 5.0, 'args': (10, 10) }, } app.conf.timezone = 'UTC' @app.task def test(arg): print(arg)
(4)启动worker：celery -A proj worker -l debug
(5)启动任务调度器：celery -A proj.periodic_task beat -l debug
worker输出：
[2019-06-18 14:29:02,377: DEBUG/MainProcess] Task accepted: proj.tasks.add[03c85e0e-973c-4607-882f-44bb1951cd7a] pid:28339
[2019-06-18 14:29:02,378: INFO/ForkPoolWorker-3] Task proj.tasks.add[03c85e0e-973c-4607-882f-44bb1951cd7a] succeeded in 0.0011271699331700802s: 20
任务调度器输出：
[2019-06-18 14:28:22,392: INFO/MainProcess] Scheduler: Sending due task 每隔五秒执行相加 (proj.tasks.add)
[2019-06-18 14:28:22,415: DEBUG/MainProcess] proj.tasks.add sent. id->083a5d92-f6ff-47ef-8bb8-be800a7edfa8
四.Celery与django结合
django可以轻松跟celery结合实现异步任务
(1)在项目文件夹里创建celery_djangocelery_djangocelery.py
#celery配置
from __future__ import absolute_import, unicode_literals import os from celery import Celery os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'celery_django.settings') #celery_django项目名字 app = Celery('celery_task') app.config_from_object('django.conf:settings', namespace='CELERY') #必须以CELERY大写开头 app.autodiscover_tasks() #可以自动刷新 @app.task(bind=True) def debug_task(self): # print('Request: {0!r}'.format(self.request))
(2)在项目文件夹celery_djangocelery_django\__init__.py里写入
from __future__ import absolute_import, unicode_literals from .celery import app as celery_app __all__ = ['celery_app']
(3)配置文件写入：celery_djangocelery_djangosettings.py
#连接redis CELERY_BROKER_URL = 'redis://:123456@192.168.1.232' CELERY_RESULT_BACKEND = 'redis://:123456@192.168.1.232'
(4)总url写入：celery_djangocelery_djangourls.py
from django.conf.urls import url,include from django.contrib import admin urlpatterns = [ url(r'^admin/', admin.site.urls), url(r'^celery/', include('celery_task.urls')),
(5)项目url写入：celery_djangocelery_taskurls.py
from django.conf.urls import url from celery_task import views urlpatterns = [ url(r'^celery_test/', views.celery_test ), #获取任务id url(r'^celery_res/', views.celery_res ), #通过任务id获取到执行结果 ]
(6)创建一个celery application用来定义你的任务列表：celery_djangocelery_task asks.py
from __future__ import absolute_import, unicode_literals from celery import shared_task # import time @shared_task def add(x, y): print("running task add",x,y ) time.sleep(10) return x + y
(7)进入项目目录里启动Celery Worker来开始监听并执行任务
命令：celery -A celery_django worker -l debug
(8)调用任务:celery_djangocelery_taskviews.py
from __future__ import absolute_import, unicode_literals from celery import shared_task # import time @shared_task def add(x, y): print("running task add",x,y ) time.sleep(10) return x + y
(9)执行任务获取id和通过id取得执行的结果函数
/celery/celery_django/celery_task/views.py
from django.shortcuts import HttpResponse from celery_task.tasks import add from celery.result import AsyncResult #执行任务获取id号 def celery_test(request): task = add.delay(4,22) return HttpResponse(task.id) #返回id号 #通过id好获取执行结果 def celery_res(request): #获取id号 task_id = '2edbcc88-12ad-4709-b770-d38179c67ef5' res = AsyncResult(id=task_id) #通过id号获取任务结果 return HttpResponse(res.get()) #把结果返回给前端
访问：http://192.168.1.232:8080/celery/celery_test/
得到任务id:2edbcc88-12ad-4709-b770-d38179c67ef5
访问：http://192.168.1.232:8080/celery/celery_res/
通过任务id得到结果：26
五.在django中使用计划任务功能
settings.py配置文件添加
INSTALLED_APPS = [ "django_celery_beat", ]
创建表存定时任务：python3 manage.py migrate