第一部分：Celery 是什么

Celery 是一個開源的、分布式的任務(wù)隊(duì)列系統(tǒng)，使用 Python 編寫。它的核心目的是處理異步任務(wù)和定時任務(wù)

核心概念類比：餐廳

想象一個繁忙的餐廳：

• 顧客/服務(wù)員 (Web 服務(wù)器)：接收點(diǎn)單（HTTP 請求）。
• 訂單 (Task)：需要完成的具體工作，比如“做一份披薩”。
• 廚房 (Celery Workers)：專門負(fù)責(zé)處理訂單的后臺團(tuán)隊(duì)。
• 訂單隊(duì)列 (Message Broker)：服務(wù)員和廚房之間的傳菜口，上面掛著待處理的訂單。

服務(wù)員將訂單放到傳菜口后，就可以立刻回去服務(wù)下一個顧客，而不必在廚房等待披薩做完。廚房的工作人員會從傳菜口按順序取出訂單并制作披薩。這就是 Celery 的核心理念：將耗時的操作放到后臺執(zhí)行，讓主程序（如 Web 服務(wù)器）能夠快速響應(yīng)客戶端。

第二部分：為什么需要使用 Celery？

在 Web 開發(fā)中，有些操作執(zhí)行起來很慢，如果讓用戶在網(wǎng)頁上一直等待直到操作完成，體驗(yàn)會非常差，甚至導(dǎo)致請求超時。

典型使用場景：

1.耗時任務(wù)：

• 發(fā)送電子郵件（用戶注冊后發(fā)送驗(yàn)證郵件）。
• 圖像/視頻處理（生成縮略圖、添加水印、轉(zhuǎn)碼）。
• 數(shù)據(jù)分析和報告生成。

2.定時任務(wù)：

• 每天凌晨備份數(shù)據(jù)庫。
• 每周一早上給所有用戶發(fā)送周報。
• 每隔 5 分鐘檢查一次系統(tǒng)狀態(tài)。

3.大規(guī)模處理：

同時處理成千上萬個 API 調(diào)用或數(shù)據(jù)清洗任務(wù)。

第三部分：Celery 的核心組件

一個完整的 Celery 系統(tǒng)通常包含四個部分：

Celery Client (客戶端)：

• 這是你的 Python 應(yīng)用程序（如 Django, Flask 項(xiàng)目）。
• 它負(fù)責(zé)定義任務(wù)和發(fā)送（調(diào)用）任務(wù)到消息代理。

Message Broker (消息代理)：

• 這是 Celery 的中樞神經(jīng)系統(tǒng)，負(fù)責(zé)傳遞消息。
• Client 將任務(wù)發(fā)送到 Broker，Worker 從 Broker 接收任務(wù)。
• 常用選擇：
  • Redis：非常流行，速度快，同時可作為結(jié)果后端。
  • RabbitMQ：功能最全、最可靠的選擇，是 Celery 的官方推薦。
  • Amazon SQS / 其他…

Celery Worker (工作者)：

• 這是一個（或多個）獨(dú)立的進(jìn)程，持續(xù)監(jiān)視消息代理。
• 它從 Broker 中獲取任務(wù)并執(zhí)行它們。
• 你可以根據(jù)任務(wù)量啟動多個 Worker 來并行處理，實(shí)現(xiàn)分布式。

Result Backend (結(jié)果后端) (可選)：

• 用于存儲任務(wù)執(zhí)行后的結(jié)果和狀態(tài)（如“成功”、“失敗”、“進(jìn)度”）。
• 如果你的應(yīng)用需要知道任務(wù)是否完成并獲取返回值，就需要配置它。
• 常用選擇：Redis, Django 數(shù)據(jù)庫, Memcached 等。

數(shù)據(jù)流總結(jié)：Client -(發(fā)送任務(wù))-> Broker -(分配任務(wù))-> Worker -(存儲結(jié)果)-> Result Backend

第四部分：快速入門與實(shí)踐

我們將使用 Redis 作為消息代理和結(jié)果后端，因?yàn)樗惭b簡單且功能強(qiáng)大。

步驟 1：安裝必要的庫

pip install celery redis

確保你已安裝并啟動了 Redis 服務(wù)。

步驟 2：創(chuàng)建 Celery 應(yīng)用和任務(wù) (tasks.py)

# tasks.py
from celery import Celery

# 創(chuàng)建 Celery 實(shí)例，'my_app' 是當(dāng)前項(xiàng)目的名稱
app = Celery('my_app',
             broker='redis://localhost:6379/0',      # 使用 Redis 作為 Broker
             backend='redis://localhost:6379/0')     # 使用 Redis 作為 Backend

# 定義一個任務(wù)
@app.task
def add(x, y):
    print(f"正在計(jì)算 {x} + {y}...")
    # 模擬一個耗時操作
    import time
    time.sleep(5)
    result = x + y
    print(f"計(jì)算完成！結(jié)果是 {result}")
    return result

@app.task
def send_email(to, subject, body):
    # 模擬發(fā)送郵件的邏輯
    print(f"正在向 {to} 發(fā)送郵件，主題：{subject}")
    # ... 實(shí)際的發(fā)信代碼 ...
    return f"郵件已發(fā)送至 {to}"

步驟 3：啟動 Worker

在終端中，進(jìn)入 tasks.py 所在的目錄，運(yùn)行以下命令啟動 Worker：

celery -A tasks worker --loglevel=info

• -A tasks：指定 Celery 應(yīng)用的位置（tasks.py 文件中的 app）。
• --loglevel=info：指定日志級別，方便查看運(yùn)行信息。

如果成功，你會看到類似下面的輸出，表示 Worker 已啟動并在等待任務(wù)：

 -------------- celery@YourComputer v5.3.0 (emerald-rush)
--- ***** -----
-- ******* ----- [Configuration]
- *** --- * --- . broker:      redis://localhost:6379/0
- ** ---------- . result:      redis://localhost:6379/0
- ** ---------- . app:         my_app:0x...
- ** ---------- . concurrency: 8 (prefork)
- ** ---------- . tasks:       ['tasks.add', 'tasks.send_email']
--- ***** ----- [Queues]
 -------------- . celery           exchange=celery(direct) key=celery

[tasks]
  . tasks.add
  . tasks.send_email

[2024-...] [INFO] Task tasks.add[1234-...] received

步驟 4：調(diào)用任務(wù)

現(xiàn)在，打開一個 Python Shell（在另一個終端中），來調(diào)用我們定義的任務(wù)。

# 在 Python Shell 中
from tasks import add, send_email

# 1. 異步調(diào)用 - 最常用的方式
# 這行代碼會立即返回一個 AsyncResult 對象，而不會阻塞程序
result = add.delay(4, 6) # .delay() 是快捷的異步調(diào)用方法
print(f"任務(wù)ID：{result.id}") # 立即打印任務(wù)ID，此時任務(wù)正在后臺運(yùn)行

# 2. 獲取任務(wù)結(jié)果（可選）
# 因?yàn)槲覀兊娜蝿?wù)有 5 秒休眠，所以需要等待一下再獲取結(jié)果
print("正在等待結(jié)果...")
value = result.get(timeout=10) # .get() 會阻塞，直到任務(wù)完成并返回結(jié)果
print(f"任務(wù)結(jié)果是：{value}")

# 3. 檢查任務(wù)狀態(tài)（可選）
print(f"任務(wù)狀態(tài)：{result.status}") # 可能是 PENDING, SUCCESS, FAILURE 等

# 4. 調(diào)用發(fā)送郵件的任務(wù)
email_result = send_email.delay('user@example.com', '歡迎光臨', '感謝您注冊！')
print(f"郵件任務(wù)ID：{email_result.id}")

步驟 5：監(jiān)控任務(wù)（可選）

Celery 提供了一個強(qiáng)大的命令行工具來監(jiān)控 Worker 的狀態(tài)和任務(wù)隊(duì)列。你還可以使用更強(qiáng)大的工具如 Flower。

# 安裝 Flower
pip install flower

# 啟動 Flower (在項(xiàng)目目錄下)
celery -A tasks flower

# 然后在瀏覽器中訪問 http://localhost:5555

第五部分：在 Web 框架（以 Flask 為例）中使用

將 Celery 集成到 Web 框架中是其最常見的用法。

# app.py
from flask import Flask, jsonify
from celery import Celery

# 配置 Flask
app = Flask(__name__)

# 配置 Celery（通常從配置文件讀?。?
def make_celery(app):
    celery = Celery(
        app.import_name,
        broker='redis://localhost:6379/0',
        backend='redis://localhost:6379/0'
    )
    # 可選：更新Celery配置，使其與Flask配置保持一致
    celery.conf.update(app.config)
    return celery

celery = make_celery(app)

@celery.task
def background_task():
    # 一個模擬的耗時任務(wù)
    import time
    time.sleep(10)
    return "后臺任務(wù)完成！"

@app.route('/start_task', methods=['POST'])
def start_task():
    # 在視圖函數(shù)中觸發(fā)后臺任務(wù)
    task = background_task.delay()
    return jsonify({'task_id': task.id}), 202 # 202 Accepted 表示請求已被接受處理

@app.route('/check_status/<task_id>')
def check_status(task_id):
    task = background_task.AsyncResult(task_id)
    if task.state == 'PENDING':
        response = {'state': task.state, 'status': '任務(wù)正在排隊(duì)或執(zhí)行中...'}
    elif task.state == 'SUCCESS':
        response = {'state': task.state, 'result': task.result}
    else: # FAILURE 等其他狀態(tài)
        response = {'state': task.state, 'status': str(task.info)}
    return jsonify(response)

if __name__ == '__main__':
    app.run(debug=True)

總結(jié)

Celery 是 Python 生態(tài)中處理后臺任務(wù)的標(biāo)桿工具，熟練掌握它能極大地提升你構(gòu)建復(fù)雜、高性能應(yīng)用的能力。

到此這篇關(guān)于Python中Celery分布式任務(wù)隊(duì)列的介紹與使用指南的文章就介紹到這了,更多相關(guān)Python Celery使用內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評論