Celery

Celery

官方

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

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

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

 

Celery架構

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

消息中間件

Celery自己不提供消息服務，可是能夠方便的和第三方提供的消息中間件集成。包括，RabbitMQ, Redis等等

任務執行單元

Worker是Celery提供的任務執行的單元，worker併發的運行在分佈式的系統節點中。

任務結果存儲

Task result store用來存儲Worker執行的任務的結果，Celery支持以不一樣方式存儲任務的結果，包括AMQP, redis等

 

使用場景

異步任務：將耗時操做任務提交給Celery去異步執行，好比發送短信/郵件、消息推送、音視頻處理等等

定時任務：定時執行某件事情，好比天天數據統計

 

Celery的安裝配置

pip install celery

消息中間件：RabbitMQ/Redis

app=celery.Celery('任務名', broker='xxx', backend='xxx', include=['xxx', 'xxx'])

 

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'
app = Celery(broker=broker, backend=backend, include=['celery_task.tasks'])

tasks.py

from .celery import app
import time
@app.task
def add(n, m):
    print(n)
    print(m)
    time.sleep(10)
    print('n+m的結果：%s' % (n + m))
    return n + m

@app.task
def low(n, m):
    print(n)
    print(m)
    print('n-m的結果：%s' % (n - m))
    return n - m

add_task.py

from celery_task import tasks

# 添加當即執行任務
t1 = tasks.add.delay(10, 20)
t2 = tasks.low.delay(100, 50)
print(t1.id)


# 添加延遲任務
from datetime import datetime, timedelta
def eta_second(second):
    ctime = datetime.now()
    utc_ctime = datetime.utcfromtimestamp(ctime.timestamp())
    time_delay = timedelta(seconds=second)
    return utc_ctime + time_delay

tasks.low.apply_async(args=(200, 50), eta=eta_second(10))

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

# 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）添加任務：自動添加任務，因此要啓動一個添加任務的服務
# 命令：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'
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 = {
    'low-task': {
        'task': 'celery_task.tasks.low',
        'schedule': timedelta(seconds=3),
        # 'schedule': crontab(hour=8, day_of_week=1),  # 每週一早八點
        'args': (300, 150),
    }
}

tasks.py

from .celery import app

import time
@app.task
def add(n, m):
    print(n)
    print(m)
    time.sleep(10)
    print('n+m的結果：%s' % (n + m))
    return n + m


@app.task
def low(n, m):
    print(n)
    print(m)
    print('n-m的結果：%s' % (n - m))
    return n - m

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中使用

celery.py

# 重點：要將 項目名.settings 所佔的文件夾添加到環境變量
# import sys
# sys.path.append(r'項目絕對路徑')

# 開啓django支持
import os
os.environ.setdefault('DJANGO_SETTINGS_MODULE', '項目名.settings')
import django
django.setup()



# 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）添加任務：自動添加任務，因此要啓動一個添加任務的服務
# 命令：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'
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 = {
    'django-task': {
        'task': 'celery_task.tasks.test_django_celery',
        'schedule': timedelta(seconds=3),
        'args': (),
    }
}

tasks.py

from .celery import app

from home.models import Banner
from settings.const import BANNER_COUNT  # 輪播圖最大顯示條數
from home.serializers import BannerModelSerializer
from django.core.cache import cache
@app.task
def update_banner_list():
    # 獲取最新內容
    banner_query = Banner.objects.filter(is_delete=False, is_show=True).order_by('-orders')[:BANNER_COUNT]
    # 序列化
    banner_data = BannerModelSerializer(banner_query, many=True).data
    for banner in banner_data:
        banner['image'] = 'http://127.0.0.1:8000' + banner['image']
    # 更新緩存
    cache.set('banner_list', banner_data)
    return True

 

celery的三種使用的應用場景 - 重點

"""
url: 發送短信的接口/email/, 發送者地址 與 發送信息 (發送的時間)
視圖類: EmailAPIView - post
post邏輯: 
    1. 交給celery異步當即執行: 拿到celery_task包中的任務, 調用delay就能夠將任務交給celery異步執行
    2. 定時發送短信(延遲到指定時間發送): 拿到celery_task包中的任務, 根據發送時間肯定延遲執行的時間, 調用apply_async延遲執行
    
定時任務: 定時爬蟲, 定時更新(接口緩存更新)
"""