目錄

• Celery

• 一、官方

• 二、Celery異步任務(wù)框架Celery架構(gòu)圖消息中間件任務(wù)執(zhí)行單元任務(wù)結(jié)果存儲

• 三、使用場景

• 四、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é)果

• 九、高級使用celery.py 定時任務(wù)配置(循環(huán)的)tasks.pyget_result.py

• 十、django中使用(更新輪播圖案例)redis的配置接口緩存views.py啟動服務(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ù)器，通過自身命令，啟動服務(wù)(內(nèi)部支持socket)
2)celery服務(wù)為為其他項目服務(wù)提供異步解決任務(wù)需求的
注：會有兩個服務(wù)同時運(yùn)行，一個是項目服務(wù)，一個是celery服務(wù)，項目服務(wù)將需要異步處理的任務(wù)交給celery服務(wù)，celery就會在需要時異步完成項目的需求
人是一個獨(dú)立運(yùn)行的服務(wù) | 醫(yī)院也是一個獨(dú)立運(yùn)行的服務(wù)
正常情況下，人可以完成所有健康情況的動作，不需要醫(yī)院的參與；但當(dāng)人生病時，就會被醫(yī)院接收，解決人生病問題
人生病的處理方案交給醫(yī)院來解決，所有人不生病時，醫(yī)院獨(dú)立運(yùn)行，人生病時，醫(yī)院就來解決人生病的需求
"""

Celery架構(gòu)圖

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

消息中間件

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

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

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

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

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

三、使用場景

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

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

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

四、Celery的安裝配置

pip install celery

消息中間件：RabbitMQ/Redis

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

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

# 如果 Celery對象:Celery(...) 是放在一個模塊下的
# 1)終端切換到該模塊所在文件夾位置：scripts
# 2)執(zhí)行啟動worker的命令：celery worker -A 模塊名 -l info -P eventlet
# 注：windows系統(tǒng)需要eventlet支持，Linux與MacOS直接執(zhí)行：celery worker -A 模塊名 -l info
# 注：模塊名隨意


# 如果 Celery對象:Celery(...) 是放在一個包下的
# 1)必須在這個包下建一個celery.py的文件，將Celery(...)產(chǎn)生對象的語句放在該文件中
# 2)執(zhí)行啟動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)啟動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ù)：手動添加，要自定義添加任務(wù)的腳本，右鍵執(zhí)行腳本

# 4)獲取結(jié)果：手動獲取，要自定義獲取任務(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.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 添加立即、延遲任務(wù)

from celery_task import tasks

# delay ：添加立即任務(wù)
# apply_async ：添加延遲任務(wù)
# eta ：執(zhí)行的utc時間


# 添加立即執(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)前時間
utc_ctime = datetime.utcfromtimestamp(ctime.timestamp()) # 當(dāng)前UTC時間
time_delay = timedelta(seconds=second) # 秒
return utc_ctime + time_delay # 當(dāng)前時間+往后延遲的秒
# 天
def eta_days(days):
ctime = datetime.now() # 當(dāng)前時間
utc_ctime = datetime.utcfromtimestamp(ctime.timestamp()) # 當(dāng)前UTC時間
time_delay = timedelta(days=days) # 天
return utc_ctime + time_delay # 當(dāng)前時間+往后延遲的天

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í)行')

九、高級使用

celery.py 定時任務(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)啟動celery(app)服務(wù)：
# 注)：-A 表示相對路徑，所以一定先進(jìn)入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)添加任務(wù)：自動添加任務(wù)，所以要啟動一個添加任務(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'])


# 時區(qū)
app.conf.timezone = 'Asia/Shanghai'
# 是否使用UTC
app.conf.enable_utc = False

# 自動任務(wù)的定時配置
from datetime import timedelta
from celery.schedules import crontab

app.conf.beat_schedule = {
# 定時任務(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：定時時間
'''


'schedule': crontab(hour=8, day_of_week=1), # 每周一早八點(diǎn)
'''
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('任務(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ù)定時任務(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}, # 同時的并發(fā)量
"DECODE_RESPONSES": True,
"PASSWORD": "123",
}
}
}

接口緩存

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

啟動服務(wù)

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

celery.py

"""
celery框架django項目工作流程
1)加載django配置環(huán)境
2)創(chuàng)建Celery框架對象app，配置broker和backend，得到的app就是worker
3)給worker對應(yīng)的app添加可處理的任務(wù)函數(shù)，用include配置給worker的app
4)完成提供的任務(wù)的定時配置app.conf.beat_schedule
5)啟動celery服務(wù)，運(yùn)行worker，執(zhí)行任務(wù)
6)啟動beat服務(wù)，運(yùn)行beat，添加任務(wù)
重點(diǎn)：由于采用了django的反射機(jī)制，使用celery.py所在的celery_task包必須放置項目的根目錄下
"""

# 一、加載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'
# worker
app = Celery(broker=broker, backend=backend, include=['celery_task.tasks']) # 外面的包名和文件名，一般都是固定


# 時區(qū)
app.conf.timezone = 'Asia/Shanghai'
# 是否使用UTC
app.conf.enable_utc = False

# 任務(wù)的定時配置
from datetime import timedelta
from celery.schedules import crontab
app.conf.beat_schedule = {
# 定時任務(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 ：小時
day_of_week ：禮拜
day_of_month:月
month_of_year:年
'''

'''
fall_task：任務(wù)名自定義
task:任務(wù)來源
args:任務(wù)參數(shù)
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，必定終身學(xué)習(xí)

作者：Jeff

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

圖片和內(nèi)容源自網(wǎng)絡(luò)分享，若有侵權(quán)，請聯(lián)系刪除！

上海艾磊科技有限公司專門為企業(yè)提供IT咨詢，IT外包，系統(tǒng)集成，以及各類IT增值服務(wù)。其中增值服務(wù)包括OFFICE 365云服務(wù)，鼎捷企業(yè)ERP管理軟件，云備份，企業(yè)郵箱，無線覆蓋，上網(wǎng)行為管理，VPN架設(shè)，網(wǎng)絡(luò)安全服務(wù)，INTERNET接入，設(shè)備租賃， IP電話服務(wù)

總結(jié)

以上是生活随笔為你收集整理的celery的中文_celery异步任务框架的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。