目錄

• Celery
• 一、官方
• 二、Celery異步任務(wù)框架Celery架構(gòu)圖消息中間件任務(wù)執(zhí)行單元任務(wù)結(jié)果存儲(chǔ)
• 三、使用場(chǎng)景
• 四、Celery的安裝配置
• 五、兩種celery任務(wù)結(jié)構(gòu)：提倡用包管理，結(jié)構(gòu)更清晰
• 七、Celery執(zhí)行異步任務(wù)包架構(gòu)封裝
• 八、基本使用celery.py 基本配置tasks.py 添加任務(wù)add_task.py 添加立即、延遲任務(wù)get_result.py 獲取結(jié)果
• 九、高級(jí)使用celery.py 定時(shí)任務(wù)配置(循環(huán)的)tasks.pyget_result.py
• 十、django中使用(更新輪播圖案例)redis的配置接口緩存views.py啟動(dòng)服務(wù)celery.pytasks.py

Celery

一、官方

Celery 官網(wǎng)：http://www.celeryproject.org/

Celery 官方文檔英文版：http://docs.celeryproject.org/en/latest/index.html

Celery 官方文檔中文版：http://docs.jinkan.org/docs/celery/

二、Celery異步任務(wù)框架

"""
1）可以不依賴任何服務(wù)器，通過自身命令，啟動(dòng)服務(wù)(內(nèi)部支持socket)
2）celery服務(wù)為為其他項(xiàng)目服務(wù)提供異步解決任務(wù)需求的
注：會(huì)有兩個(gè)服務(wù)同時(shí)運(yùn)行，一個(gè)是項(xiàng)目服務(wù)，一個(gè)是celery服務(wù)，項(xiàng)目服務(wù)將需要異步處理的任務(wù)交給celery服務(wù)，celery就會(huì)在需要時(shí)異步完成項(xiàng)目的需求
人是一個(gè)獨(dú)立運(yùn)行的服務(wù) | 醫(yī)院也是一個(gè)獨(dú)立運(yùn)行的服務(wù)
	正常情況下，人可以完成所有健康情況的動(dòng)作，不需要醫(yī)院的參與；但當(dāng)人生病時(shí)，就會(huì)被醫(yī)院接收，解決人生病問題
	人生病的處理方案交給醫(yī)院來解決，所有人不生病時(shí)，醫(yī)院獨(dú)立運(yùn)行，人生病時(shí)，醫(yī)院就來解決人生病的需求
"""

Celery架構(gòu)圖

Celery的架構(gòu)由三部分組成，消息中間件（message broker）、任務(wù)執(zhí)行單元（worker）和 任務(wù)執(zhí)行結(jié)果存儲(chǔ)（task result store）組成。

消息中間件

Celery本身不提供消息服務(wù)，但是可以方便的和第三方提供的消息中間件集成。包括，RabbitMQ, Redis等等

任務(wù)執(zhí)行單元

Worker是Celery提供的任務(wù)執(zhí)行的單元，worker并發(fā)的運(yùn)行在分布式的系統(tǒng)節(jié)點(diǎn)中。

任務(wù)結(jié)果存儲(chǔ)

Task result store用來存儲(chǔ)Worker執(zhí)行的任務(wù)的結(jié)果，Celery支持以不同方式存儲(chǔ)任務(wù)的結(jié)果，包括AMQP, redis等

三、使用場(chǎng)景

異步執(zhí)行：解決耗時(shí)任務(wù)

延遲執(zhí)行：解決延遲任務(wù)

定時(shí)執(zhí)行：解決周期(周期)任務(wù)

四、Celery的安裝配置

pip install celery

消息中間件：RabbitMQ/Redis

App=Celery('任務(wù)名', broker='xxx', backend='xxx')

五、兩種celery任務(wù)結(jié)構(gòu)：提倡用包管理，結(jié)構(gòu)更清晰

# 如果 Celery對(duì)象:Celery(...) 是放在一個(gè)模塊下的
# 1）終端切換到該模塊所在文件夾位置：scripts
# 2）執(zhí)行啟動(dòng)worker的命令：celery worker -A 模塊名 -l info -P eventlet
# 注：windows系統(tǒng)需要eventlet支持，linux與macOS直接執(zhí)行：celery worker -A 模塊名 -l info
# 注：模塊名隨意
# 如果 Celery對(duì)象:Celery(...) 是放在一個(gè)包下的
# 1）必須在這個(gè)包下建一個(gè)celery.py的文件，將Celery(...)產(chǎn)生對(duì)象的語句放在該文件中
# 2）執(zhí)行啟動(dòng)worker的命令：celery worker -A 包名 -l info -P eventlet
# 注：windows系統(tǒng)需要eventlet支持，Linux與MacOS直接執(zhí)行：celery worker -A 模塊名 -l info
# 注：包名隨意

放在根目錄下就行：

七、Celery執(zhí)行異步任務(wù)

包架構(gòu)封裝

project
    ├── celery_task  	# celery包
    │   ├── __init__.py # 包文件
    │   ├── celery.py   # celery連接和配置相關(guān)文件，且名字必須交celery.py
    │   └── tasks.py    # 所有任務(wù)函數(shù)
    ├── add_task.py  	# 添加任務(wù)
    └── get_result.py   # 獲取結(jié)果

八、基本使用

celery.py 基本配置

# 1）創(chuàng)建app + 任務(wù)
# 2）啟動(dòng)celery(app)服務(wù)：
# 非windows# 命令：celery worker -A celery_task -l info# windows：# pip3 install eventlet# celery worker -A celery_task -l info -P eventlet# 3）添加任務(wù)：手動(dòng)添加，要自定義添加任務(wù)的腳本，右鍵執(zhí)行腳本
# 4）獲取結(jié)果：手動(dòng)獲取，要自定義獲取任務(wù)的腳本，右鍵執(zhí)行腳本
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 : 任務(wù)倉庫
backend ： 任務(wù)結(jié)果倉庫
include ：任務(wù)(函數(shù))所在文件
'''

tasks.py 添加任務(wù)

from .celery import app
@app.taskdef add(n1,n2):    res = n1+n2    print('n1+n2 = %s' % res)
    return res
@app.taskdef low(n1,n2):    res = n1-n2    print('n1-n2 = %s' % res)
    return res

add_task.py 添加立即、延遲任務(wù)

from celery_task import tasks
# delay  ：添加立即任務(wù)# apply_async ： 添加延遲任務(wù)# eta ： 執(zhí)行的utc時(shí)間# 添加立即執(zhí)行任務(wù)t1 = tasks.add.delay(10, 20)
t2 = tasks.low.delay(100, 50)
print(t1.id)# 添加延遲任務(wù)from celery_package.tasks import jump
from datetime import datetime,timedelta
# 秒def eta_second(second):    ctime = datetime.now()  # 當(dāng)前時(shí)間    utc_ctime = datetime.utcfromtimestamp(ctime.timestamp())  # 當(dāng)前UTC時(shí)間    time_delay = timedelta(seconds=second)  # 秒    return utc_ctime + time_delay  # 當(dāng)前時(shí)間+往后延遲的秒
# 天def eta_days(days):    ctime = datetime.now()  # 當(dāng)前時(shí)間    utc_ctime = datetime.utcfromtimestamp(ctime.timestamp())  # 當(dāng)前UTC時(shí)間    time_delay = timedelta(days=days)  # 天    return utc_ctime + time_delay  # 當(dāng)前時(shí)間+往后延遲的天
jump.apply_async(args=(20,5), eta=eta_second(10))  # 10秒后執(zhí)行
jump.apply_async(args=(20,5), eta=eta_days(1))  # 1天后執(zhí)行

get_result.py 獲取結(jié)果

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('任務(wù)失敗')
    elif async.status == 'PENDING':
        print('任務(wù)等待中被執(zhí)行')
    elif async.status == 'RETRY':
        print('任務(wù)異常后正在重試')
    elif async.status == 'STARTED':
        print('任務(wù)已經(jīng)開始被執(zhí)行')

九、高級(jí)使用

celery.py 定時(shí)任務(wù)配置(循環(huán)的)

特點(diǎn)：

添加任務(wù)的終端關(guān)閉之后，停止添加

celery服務(wù)端關(guān)閉后，把關(guān)閉之后未執(zhí)行的任務(wù)都執(zhí)行一遍，然后繼續(xù)接收任務(wù)

# 1）創(chuàng)建app + 任務(wù)
# 2）啟動(dòng)celery(app)服務(wù)：
# 注）：-A  表示相對(duì)路徑，所以一定先進(jìn)入celery_task所在包	   -l 表示打印到日志 info 級(jí)別# 非windows# 命令：celery worker -A celery_task -l info# windows：# pip3 install eventlet# celery worker -A celery_task -l info -P eventlet# 3）添加任務(wù)：自動(dòng)添加任務(wù)，所以要啟動(dòng)一個(gè)添加任務(wù)的服務(wù)
# 命令：celery beat -A celery_task -l info# 4）獲取結(jié)果
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'])
# 時(shí)區(qū)app.conf.timezone = 'Asia/Shanghai'
# 是否使用UTCapp.conf.enable_utc = False# 自動(dòng)任務(wù)的定時(shí)配置from datetime import timedelta
from celery.schedules import crontab
app.conf.beat_schedule = {    # 定時(shí)任務(wù)名字    'fall_task': {
        'task': 'celery_task.tasks.fall',
        'args':(30,20),
        'schedule': timedelta(seconds=3),  # 3秒后執(zhí)行
        # 'schedule': crontab(hour=8, day_of_week=1),  # 每周一早八點(diǎn)
    }}'''
fall_task：任務(wù)名自定義
task:任務(wù)來源
args:任務(wù)參數(shù)
schedule：定時(shí)時(shí)間
'''
'schedule': crontab(hour=8, day_of_week=1),  # 每周一早八點(diǎn)
'''
minute : 分鐘
hour ：小時(shí)
day_of_week ：禮拜
day_of_month:月
month_of_year:年
'''

tasks.py

from .celery import app
@app.taskdef 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('任務(wù)失敗')
    elif async.status == 'PENDING':
        print('任務(wù)等待中被執(zhí)行')
    elif async.status == 'RETRY':
        print('任務(wù)異常后正在重試')
    elif async.status == 'STARTED':
        print('任務(wù)已經(jīng)開始被執(zhí)行')

十、django中使用(更新輪播圖案例)

最終達(dá)到的效果：根據(jù)定時(shí)任務(wù)來更新redis中的緩存。用戶獲取資源都是從redis緩存中獲取。避免了數(shù)據(jù)庫的壓力

redis的配置

dev.py

# 緩存redis數(shù)據(jù)庫配置
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}, # 同時(shí)的并發(fā)量
            "DECODE_RESPONSES": True,
            "PASSword": "123",
        }    }}

接口緩存

"""
1）什么是接口的后臺(tái)緩存
	前臺(tái)訪問后臺(tái)接口，后臺(tái)會(huì)優(yōu)先從緩存(內(nèi)存)中查找接口數(shù)據(jù)
		如果有數(shù)據(jù)，直接對(duì)前臺(tái)響應(yīng)緩存數(shù)據(jù)
		如果沒有數(shù)據(jù)，與(MySQL)數(shù)據(jù)庫交互，得到數(shù)據(jù)，對(duì)前臺(tái)響應(yīng)，同時(shí)將數(shù)據(jù)進(jìn)行緩存，以備下次使用
	
	了解：前臺(tái)緩存 - 前臺(tái)在請(qǐng)求到接口數(shù)據(jù)后，在前臺(tái)建立緩存，再發(fā)送同樣請(qǐng)求時(shí)，發(fā)現(xiàn)前臺(tái)緩存有數(shù)據(jù)，就不再對(duì)后臺(tái)做請(qǐng)求了
	
2）什么的接口會(huì)進(jìn)行接口緩存
	i）接口會(huì)被大量訪問：比如主頁中的接口，幾乎所有人都會(huì)訪問，而且會(huì)重復(fù)訪問
	ii）在一定時(shí)間內(nèi)數(shù)據(jù)不會(huì)變化(或數(shù)據(jù)不變化)的接口
	iii）接口數(shù)據(jù)的時(shí)效性不是特別強(qiáng)（數(shù)據(jù)庫數(shù)據(jù)發(fā)生變化了，不是立即同步給前臺(tái)，驗(yàn)后時(shí)間同步給前臺(tái)也沒事）
	注：理論上所有接口都可以建立緩存，只要數(shù)據(jù)庫與緩存數(shù)據(jù)同步及時(shí)
	
3）如何實(shí)現(xiàn)接口緩存：主頁輪播圖接口
"""

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    # 有緩存走緩存，沒有緩存走數(shù)據(jù)庫，然后同步給緩存。接口自己實(shí)現(xiàn)    def list(self, request, *args, **kwargs):        banner_list = cache.get('banner_list')
        if not banner_list:
            print('走了數(shù)據(jù)庫')
            response = self.list(request, *args, **kwargs)            banner_list = response.data
            cache.set('banner_list', banner_list, 86400)  # 存進(jìn)緩存中，緩存配置了redis數(shù)據(jù)庫
        return Response(banner_list)

啟動(dòng)服務(wù)

'''
1):先切換到celery_task所在的同級(jí)目錄(一般為根目錄下)
2):開一個(gè)終端(啟動(dòng)服務(wù))：		celery worker -A celery_task -l info -P eventlet
3):再開一個(gè)終端(添加任務(wù)):	celery beat -A celery_task -l info
'''
# 注）：-A  表示相對(duì)路徑，所以一定先進(jìn)入celery_task所在包
	   -l 表示打印到日志 info 級(jí)別

celery.py

"""
celery框架django項(xiàng)目工作流程
1）加載django配置環(huán)境
2）創(chuàng)建Celery框架對(duì)象app，配置broker和backend，得到的app就是worker
3）給worker對(duì)應(yīng)的app添加可處理的任務(wù)函數(shù)，用include配置給worker的app
4）完成提供的任務(wù)的定時(shí)配置app.conf.beat_schedule
5）啟動(dòng)celery服務(wù)，運(yùn)行worker，執(zhí)行任務(wù)
6）啟動(dòng)beat服務(wù)，運(yùn)行beat，添加任務(wù)
重點(diǎn)：由于采用了django的反射機(jī)制，使用celery.py所在的celery_task包必須放置項(xiàng)目的根目錄下
"""
# 一、加載django配置環(huán)境import os
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "luffyapi.settings.dev")
# 二、加載celery配置環(huán)境from celery import Celery
broker = 'redis://:123@127.0.0.1:6379/1'
backend = 'redis://:123@127.0.0.1:6379/2'
# workerapp = Celery(broker=broker, backend=backend, include=['celery_task.tasks'])  # 外面的包名和文件名，一般都是固定
# 時(shí)區(qū)app.conf.timezone = 'Asia/Shanghai'
# 是否使用UTCapp.conf.enable_utc = False# 任務(wù)的定時(shí)配置from datetime import timedelta
from celery.schedules import crontab
app.conf.beat_schedule = {    # 定時(shí)任務(wù)名字    'update_banner_cache': {
        'task': 'celery_task.tasks.update_banner_list',
        'args': (),
        'schedule': timedelta(seconds=10),  # 3秒一次
        # 'schedule': crontab(hour=8, day_of_week=1),  # 每周一早八點(diǎn)
        # 'schedule': crontab(minute=0, day_of_week=1),  # 每周一早八點(diǎn)
    }}'''
minute : 分鐘
hour ：小時(shí)
day_of_week ：禮拜
day_of_month:月
month_of_year:年
'''
'''
fall_task：任務(wù)名自定義
task:任務(wù)來源
args:任務(wù)參數(shù)
schedule：定時(shí)時(shí)間(秒)
'''

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對(duì)象，所以頭像的連接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

選擇了IT，必定終身學(xué)習(xí)

作者：Jeff

出處：http://dwz.date/aNfM