celery mysql 异步_celery异步框架

不念不忘少年蓝@ 2022-10-27 13:48 227阅读 0赞

Celery

一、官方

二、Celery异步任务框架

"""

1)可以不依赖任何服务器，通过自身命令，启动服务(内部支持socket)

2)celery服务为为其他项目服务提供异步解决任务需求的

注：会有两个服务同时运行，一个是项目服务，一个是celery服务，项目服务将需要异步处理的任务交给celery服务，celery就会在需要时异步完成项目的需求

人是一个独立运行的服务 | 医院也是一个独立运行的服务

正常情况下，人可以完成所有健康情况的动作，不需要医院的参与；但当人生病时，就会被医院接收，解决人生病问题

人生病的处理方案交给医院来解决，所有人不生病时，医院独立运行，人生病时，医院就来解决人生病的需求

"""

Celery架构图

Celery的架构由三部分组成，消息中间件(message broker)、任务执行单元(worker)和 任务执行结果存储(task result store)组成。

![9480acf0761dc092b2216531e0e182c4.png][]

消息中间件

Celery本身不提供消息服务，但是可以方便的和第三方提供的消息中间件集成。包括，RabbitMQ, Redis等等

任务执行单元

Worker是Celery提供的任务执行的单元，worker并发的运行在分布式的系统节点中。

任务结果存储

Task result store用来存储Worker执行的任务的结果，Celery支持以不同方式存储任务的结果，包括AMQP, redis等

三、使用场景

异步执行：解决耗时任务

延迟执行：解决延迟任务

定时执行：解决周期(周期)任务

四、Celery的安装配置

pip install celery

消息中间件：RabbitMQ/Redis

app=Celery('任务名', broker='xxx', backend='xxx')

五、两种celery任务结构：提倡用包管理，结构更清晰

\# 如果 Celery对象:Celery(...) 是放在一个模块下的

\# 1)终端切换到该模块所在文件夹位置：scripts

\# 2)执行启动worker的命令：celery worker -A 模块名 -l info -P eventlet

\# 注：windows系统需要eventlet支持，Linux与MacOS直接执行：celery worker -A 模块名 -l info，启动后会在该模块内找这个celery产生的对象app

对象

\# 注：模块名随意

![3d20c853648ad7987e35ba714ade8deb.png][]

\# 如果 Celery对象:Celery(...) 是放在一个包下的

\# 1)必须在这个包下建一个celery.py的文件，将Celery(...)产生对象的语句放在该文件中

\# 2)执行启动worker的命令：celery worker -A 包名 -l info -P eventlet

\# 注：windows系统需要eventlet支持，Linux与MacOS直接执行：celery worker -A 模块名 -l info

\# 注：包名随意

放在根目录下就行：

![ef209ae1a23844174e5fb5064eb5a912.png][]

七、Celery执行异步任务

包架构封装

project

├── celery\_task \# celery包

│ ├── \_\_init\_\_.py \# 包文件

│ ├── celery.py \# celery连接和配置相关文件，且名字必须交celery.py

│ └── tasks.py \# 所有任务函数

├── add\_task.py \# 添加任务

└── get\_result.py \# 获取结果

八、基本使用

celery.py 基本配置

\# 1)创建app + 任务

\# 2)启动celery(app)服务：

\# 非windows

\# 命令：celery worker -A celery\_task -l info

\# windows：

\# pip3 install eventlet

\# celery worker -A celery\_task -l info -P eventlet

\# 3)添加任务：手动添加，要自定义添加任务的脚本，右键执行脚本

\# 4)获取结果：手动获取，要自定义获取任务的脚本，右键执行脚本

from celery import Celery

\# 无密码

broker = 'redis://127.0.0.1:6379/1'

backend = 'redis://127.0.0.1:6379/2'

\# 有密码：

broker = 'redis://:123@127.0.0.1:6379/1'

backend = 'redis://:123@127.0.0.1:6379/2'

app = Celery(broker=broker, backend=backend, include=\['celery\_task.tasks'\])

'''

broker : 任务仓库

backend ： 任务结果仓库

include ：任务(函数)所在文件

'''

tasks.py 添加任务

from .celery import app

@app.task

def add(n1,n2):

res = n1+n2

print('n1+n2 = %s' % res)

return res

@app.task

def low(n1,n2):

res = n1-n2

print('n1-n2 = %s' % res)

return res

add\_task.py 添加立即、延迟任务

from celery\_task import tasks

\# delay ：添加立即任务

\# apply\_async ： 添加延迟任务

\# eta ： 执行的utc时间

\# 添加立即执行任务

t1 = tasks.add.delay(10, 20)

t2 = tasks.low.delay(100, 50)

print(t1.id)

\# 添加延迟任务

from celery\_package.tasks import jump

from datetime import datetime,timedelta

\# 秒

def eta\_second(second):

ctime = datetime.now() \# 当前时间

utc\_ctime = datetime.utcfromtimestamp(ctime.timestamp()) \# 当前UTC时间

time\_delay = timedelta(seconds=second) \# 秒

return utc\_ctime + time\_delay \# 当前时间+往后延迟的秒

\# 天

def eta\_days(days):

ctime = datetime.now() \# 当前时间

utc\_ctime = datetime.utcfromtimestamp(ctime.timestamp()) \# 当前UTC时间

time\_delay = timedelta(days=days) \# 天

return utc\_ctime + time\_delay \# 当前时间+往后延迟的天

jump.apply\_async(args=(20,5), eta=eta\_second(10)) \# 10秒后执行

jump.apply\_async(args=(20,5), eta=eta\_days(1)) \# 1天后执行

get\_result.py 获取结果

from celery\_task.celery import app

from celery.result import AsyncResult

\#通过获取任务id进行比得到任务的执行结果

id = '21325a40-9d32-44b5-a701-9a31cc3c74b5'

if \_\_name\_\_ == '\_\_main\_\_':

async = AsyncResult(id=id, app=app)

if async.successful():

result = async.get()

print(result)

elif async.failed():

print('任务失败')

elif async.status == 'PENDING':

print('任务等待中被执行')

elif async.status == 'RETRY':

print('任务异常后正在重试')

elif async.status == 'STARTED':

print('任务已经开始被执行')

九、高级使用

celery.py 定时任务配置(循环的)

特点：

添加任务的终端关闭之后，停止添加

celery服务端关闭后，把关闭之后未执行的任务都执行一遍，然后继续接收任务

相当于生产者消费者模型

\# 1)创建app + 任务

\# 2)启动celery(app)服务：

\# 注)：-A 表示相对路径，所以一定先进入celery\_task所在包

\-l 表示打印到日志 info 级别

\# 非windows

\# 命令：celery worker -A celery\_task -l info

\# windows：

\# pip3 install eventlet

\# celery worker -A celery\_task -l info -P eventlet

\# 3)添加任务：自动添加任务，所以要启动一个添加任务的服务

\# 命令：celery beat -A celery\_task -l info

\# 4)获取结果

from celery import Celery

\# 无密码

broker = 'redis://127.0.0.1:6379/1'

backend = 'redis://127.0.0.1:6379/2'

\# 有密码：

broker = 'redis://:123@127.0.0.1:6379/1'

backend = 'redis://:123@127.0.0.1:6379/2'

app = Celery(broker=broker, backend=backend, include=\['celery\_task.tasks'\])

\# 时区

app.conf.timezone = 'Asia/Shanghai'

\# 是否使用UTC

app.conf.enable\_utc = False

\# 自动任务的定时配置

from datetime import timedelta

from celery.schedules import crontab

app.conf.beat\_schedule = \{

\# 定时任务名字

'fall\_task': \{

'task': 'celery\_task.tasks.fall',

'args':(30,20),

'schedule': timedelta(seconds=3), \# 3秒后执行

\# 'schedule': crontab(hour=8, day\_of\_week=1), \# 每周一早八点

'''

fall\_task：任务名自定义

task:任务来源

args:任务参数

schedule：定时时间

'''

'schedule': crontab(hour=8, day\_of\_week=1), \# 每周一早八点

'''

minute : 分钟

hour ：小时

day\_of\_week ：礼拜

day\_of\_month:月

month\_of\_year:年

'''

tasks.py

from .celery import app

@app.task

def fall(n1,n2):

res = n1/n2

print('n1 /n2 = %s' % res)

return res

get\_result.py

from celery\_task.celery import app

from celery.result import AsyncResult

id = '21325a40-9d32-44b5-a701-9a31cc3c74b5'

if \_\_name\_\_ == '\_\_main\_\_':

async = AsyncResult(id=id, app=app)

if async.successful():

result = async.get()

print(result)

elif async.failed():

print('任务失败')

elif async.status == 'PENDING':

print('任务等待中被执行')

elif async.status == 'RETRY':

print('任务异常后正在重试')

elif async.status == 'STARTED':

print('任务已经开始被执行')

十、django中使用(更新轮播图案例)

最终达到的效果：根据定时任务来更新redis中的缓存。用户获取资源都是从redis缓存中获取。避免了数据库的压力

redis的配置

dev.py

\# 缓存redis数据库配置

CACHES = \{

"default": \{

"BACKEND": "django\_redis.cache.RedisCache",

"LOCATION": "redis://127.0.0.1:6379/10",

"OPTIONS": \{

"CLIENT\_CLASS": "django\_redis.client.DefaultClient",

"CONNECTION\_POOL\_KWARGS": \{"max\_connections": 100\}, \# 同时的并发量

"DECODE\_RESPONSES": True,

"PASSWORD": "123",

接口缓存

"""

1)什么是接口的后台缓存

前台访问后台接口，后台会优先从缓存(内存)中查找接口数据

如果有数据，直接对前台响应缓存数据

如果没有数据，与(mysql)数据库交互，得到数据，对前台响应，同时将数据进行缓存，以备下次使用

了解：前台缓存 - 前台在请求到接口数据后，在前台建立缓存，再发送同样请求时，发现前台缓存有数据，就不再对后台做请求了

2)什么的接口会进行接口缓存

i)接口会被大量访问：比如主页中的接口，几乎所有人都会访问，而且会重复访问

ii)在一定时间内数据不会变化(或数据不变化)的接口

iii)接口数据的时效性不是特别强(数据库数据发生变化了，不是立即同步给前台，验后时间同步给前台也没事)

注：理论上所有接口都可以建立缓存，只要数据库与缓存数据同步及时

3)如何实现接口缓存：主页轮播图接口

"""

views.py

from rest\_framework.viewsets import ModelViewSet

from rest\_framework import mixins

from . import models, serializers

from django.conf import settings

from rest\_framework.response import Response

from django.core.cache import cache

class BannerViewSet(ModelViewSet, mixins.ListModelMixin):

queryset = models.Banner.objects.filter(is\_delete=False, is\_show=True).order\_by('-orders')\[:settings.BANNER\_COUNT\]

serializer\_class = serializers.BannerSerializer

\# 有缓存走缓存，没有缓存走数据库，然后同步给缓存。接口自己实现

def list(self, request, \*args, \*\*kwargs):

banner\_list = cache.get('banner\_list')

if not banner\_list:

print('走了数据库')

response = self.list(request, \*args, \*\*kwargs)

banner\_list = response.data

cache.set('banner\_list', banner\_list, 86400) \# 存进缓存中，缓存配置了redis数据库

return Response(banner\_list)

![1e90465f5c56a857b07aedd04e0c1114.png][]

启动服务

'''

1):先切换到celery\_task所在的同级目录(一般为根目录下)

2):开一个终端(启动服务)：celery worker -A celery\_task -l info -P eventlet

3):再开一个终端(添加任务):celery beat -A celery\_task -l info

'''

\# 注)：-A 表示相对路径，所以一定先进入celery\_task所在包

\-l 表示打印到日志 info 级别

celery.py

"""

celery框架django项目工作流程

1)加载django配置环境

2)创建Celery框架对象app，配置broker和backend，得到的app就是worker

3)给worker对应的app添加可处理的任务函数，用include配置给worker的app

4)完成提供的任务的定时配置app.conf.beat\_schedule

5)启动celery服务，运行worker，执行任务

6)启动beat服务，运行beat，添加任务

重点：由于采用了django的反射机制，使用celery.py所在的celery\_task包必须放置项目的根目录下

"""

\# 一、加载django配置环境

import os

os.environ.setdefault("DJANGO\_SETTINGS\_MODULE", "luffyapi.settings.dev")

![bb9f9d0dab6f78531f08833c410fbe9c.png][]

\# 二、加载celery配置环境

from celery import Celery

broker = 'redis://:123@127.0.0.1:6379/1'

backend = 'redis://:123@127.0.0.1:6379/2'

\# worker

app = Celery(broker=broker, backend=backend, include=\['celery\_task.tasks'\]) \# 外面的包名和文件名，一般都是固定

\# 时区

app.conf.timezone = 'Asia/Shanghai'

\# 是否使用UTC

app.conf.enable\_utc = False

\# 任务的定时配置

from datetime import timedelta

from celery.schedules import crontab

app.conf.beat\_schedule = \{

\# 定时任务名字

'update\_banner\_cache': \{

'task': 'celery\_task.tasks.update\_banner\_list',

'args': (),

'schedule': timedelta(seconds=10), \# 3秒一次

\# 'schedule': crontab(hour=8, day\_of\_week=1), \# 每周一早八点

\# 'schedule': crontab(minute=0, day\_of\_week=1), \# 每周一早八点

'''

minute : 分钟

hour ：小时

day\_of\_week ：礼拜

day\_of\_month:月

month\_of\_year:年

'''

fall\_task：任务名自定义

task:任务来源

args:任务参数

schedule：定时时间(秒)

'''

tasks.py

from .celery import app

from django.core.cache import cache

from home import models, serializers

from django.conf import settings

@app.task

def update\_banner\_list():

queryset = models.Banner.objects.filter(is\_delete=False, is\_show=True).order\_by('-orders')\[:settings.BANNER\_COUNT\]

banner\_list = serializers.BannerSerializer(queryset, many=True).data

\# 拿不到request对象，所以头像的连接base\_url要自己组装

for banner in banner\_list:

banner\['image'\] = 'http://127.0.0.1:8000%s' % banner\['image'\]

cache.set('banner\_list', banner\_list, 86400)

return True

[9480acf0761dc092b2216531e0e182c4.png]: /images/20221024/77cc6c36184c4fa1aaa374bebbf5acf0.png
[3d20c853648ad7987e35ba714ade8deb.png]: /images/20221024/769af6bc78cf445da053d6cbd48f6220.png
[ef209ae1a23844174e5fb5064eb5a912.png]: /images/20221024/b98c3de2a37f435ca3a6c32dff16f6e2.png
[1e90465f5c56a857b07aedd04e0c1114.png]: /images/20221024/64d02f059ab74138bd754eeb02629815.png
[bb9f9d0dab6f78531f08833c410fbe9c.png]: /images/20221024/5ed6e5f9fcab4174b56694aca5b88c47.png