一、celery的作用

celery 是一個簡單、靈活且可靠的分布式系統(tǒng)，用于處理大量消息，同時為操作提供一致的接口。它專注于實時操作，但支持任務(wù)調(diào)度。

Celery 主要用于異步任務(wù)處理，特別是在Web應(yīng)用環(huán)境中，用于執(zhí)行后臺任務(wù)，如發(fā)送電子郵件、處理圖片、視頻轉(zhuǎn)碼、運行復(fù)雜的計算等，而不必阻塞Web請求/響應(yīng)周期。

二、安裝celery

命令：

pip install celery -i https://pypi.tuna.tsinghua.edu.cn/simple

三、使用celery 異步執(zhí)行任務(wù)

官方文檔：First steps with Django — Celery 5.4.0 documentation (celeryq.dev)

1、配置

• 1）給celery設(shè)置環(huán)境，為django項目的環(huán)境
• 2）創(chuàng)建celery實例
• 3）設(shè)置broker,也就是我們存放任務(wù)的隊列,這里從配置文件中加載broker
• 4）設(shè)置自動檢測任務(wù)，autodiscover_tasks方法

具體實現(xiàn)如下:

broker配置文件如下，這里用redis做隊列來存任務(wù)：

2、創(chuàng)建好celery實例和配置broker之后，開始創(chuàng)建任務(wù)

比如我要創(chuàng)建一個發(fā)送郵件的任務(wù)：

• 1）創(chuàng)建文件夾，創(chuàng)建tasks.py文件

• 2）在tasks.py文件中編寫任務(wù)(函數(shù))， 一定要用task裝飾器修飾起來

• 3）在app.autodiscover_tasks([ ])中添加你的任務(wù)

參數(shù)是列表，列表中的元素是task的路徑(就是要執(zhí)行的任務(wù)所在的路徑)

• 4）在你想要異步執(zhí)行該函數(shù)的地方導(dǎo)入，用delay方法將任務(wù)推到隊列中

四、使用celery定期執(zhí)行任務(wù)

官方文檔：Periodic Tasks — Celery 5.4.0 documentation (celeryq.dev)

1、創(chuàng)建一個文件，專門用來放定時任務(wù)的配置(也可以不用)

• task： 就是要執(zhí)行的定時任務(wù)， 注意位置要找對，要找到對應(yīng)的函數(shù)
• schedule：執(zhí)行的周期

2、用 app.conf.beat_schedule 來指定我們的定時任務(wù)

3、在app.autodiscover_tasks([ ])中添加你的任務(wù)

參數(shù)是列表，列表中的元素是task的路徑(就是要執(zhí)行的任務(wù)所在的路徑)

五、啟動命令

• 命令：Celery -A celery實例腳本文件 worker --loglevel=INFO
• linux命令：celery -A celery_tasks.main worker --loglevel=INFO
• windows命令：celery -A celery_tasks.main worker --concurrency=4 --loglevel=INFO -P threads

定時任務(wù)，需要開兩個終端分別執(zhí)行兩條命令，celery_tasks.main替換成你自己的腳本文件

• 1、celery -A celery_tasks.main worker --concurrency=4- -loglevel=INFO -P threads
• 2、celery -A celery_tasks.main beat --loglevel=INFO

六、 效果展示

任務(wù)的執(zhí)行

定期任務(wù)的執(zhí)行

七、具體文件目錄

• main.py

'''
1、生產(chǎn)者
2、消費者
使用命令(linux)：celery -A celery實例腳本文件 worker -l info  啟動進程，消費者拿任務(wù)， 終端執(zhí)行
windows命令：celery -A celery_tasks.main worker --concurrency=4 --loglevel=INFO -P threads
3、消息隊列
'''
 
import os
from celery import Celery
 
# 為celery設(shè)置環(huán)境
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "mall_test.settings")
 
# 創(chuàng)建celery實例
app = Celery('celery_tasks')
 
# 設(shè)置broker,消息隊列, 從配置信息中加載
app.config_from_object('celery_tasks.config')
 
# 讓celery自動檢測指定包的任務(wù)
# 參數(shù)是列表，列表中的元素是task的路徑(就是要執(zhí)行的任務(wù)所在的路徑)
app.autodiscover_tasks(['celery_tasks.sms', 'celery_tasks.email', 'celery_tasks.generate_static'])
 
# 定時任務(wù)
from celery_tasks.celery_beat_schedule import CELERY_BEAT_SCHEDULE
app.conf.beat_schedule = CELERY_BEAT_SCHEDULE  # 加入定時任務(wù)

• config.py

# 設(shè)置消息隊列，用redis實現(xiàn)
broker_url = 'redis://127.0.0.1:6379/15'

• celery_beat_schedule.py

from celery.schedules import crontab
 
# 定時任務(wù), 可以添加多個，啟動命令
CELERY_BEAT_SCHEDULE = {
    'update-html': {
        'task': 'celery_tasks.generate_static.tasks.generate_static_index_html',  # 找到對應(yīng)函數(shù)即可
        'schedule': crontab(minute='*/1'),  # 周期，這里是30分鐘分鐘
    },
}

總結(jié)

以上為個人經(jīng)驗，希望能給大家一個參考，也希望大家多多支持腳本之家。

最新評論