目錄

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

Celery

一、官方

Celery 官網：http://www.celeryproject.org/

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

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

二、Celery異步任務框架

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

Celery架構圖

Celery的架構由三部分組成，消息中間件（message broker）、任務執行單元（worker）和 任務執行結果存儲（task result store）組成。

消息中間件

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對象:Celery(...) 是放在一個包下的
# 1）必須在這個包下建一個celery.py的文件，將Celery(...)產生對象的語句放在該文件中
# 2）執行啟動worker的命令：celery worker -A 包名 -l info -P eventlet
# 注：windows系統需要eventlet支持，Linux與MacOS直接執行：celery worker -A 模塊名 -l info
# 注：包名隨意

放在根目錄下就行：

七、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.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 添加立即、延遲任務

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 = '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'
# 是否使用UTCapp.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.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('任務失敗')
    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)

啟動服務

'''
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")
# 二、加載celery配置環境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'])  # 外面的包名和文件名，一般都是固定
# 時區app.conf.timezone = 'Asia/Shanghai'
# 是否使用UTCapp.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

選擇了IT，必定終身學習

作者：Jeff

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