RQ Celery 分散式任務佇列實踐

– 8 分鐘閱讀

在現代 Web 應用中，處理耗時操作時，非同步任務佇列是不可或缺的元件。本文將介紹兩種 Python 分散式任務佇列框架：RQ 和 Celery，並探討它們的應用場景和實務技巧。RQ 根據 Redis，提供簡潔的任務管理方式，適合輕量級應用。Celery 功能更為豐富，支援多種 broker 和後端，可處理更複雜的工作流程。兩種框架都支援任務的非同步執行、結果儲存和過期設定，讓開發者能有效管理後端任務，提升系統效能和可擴充套件性。Celery 更進一步提供任務重試機制，確保任務在失敗時能自動重試，提高系統穩定性。此外，Celery 還允許忽略任務結果，減少不必要的儲存開銷，適用於只需執行任務而不需關心結果的場景。

RQ 任務佇列系統

RQ 是一種根據 Redis 的分散式任務佇列系統。它提供了一種簡單的方式來將任務新增到佇列中，並由工作者處理這些任務。RQ 支援多個佇列和工作者，可以根據需要進行擴充套件。

推播任務到佇列中

要將任務推播到佇列中，可以使用 q.enqueue 方法。這個方法需要一個可呼叫的函式和任務的引數。例如：

from rq import Queue
from redis import Redis

q = Queue(connection=Redis())
job = q.enqueue(sum, [42, 43])

這個例子將 sum 函式和 [42, 43] 引數推播到佇列中。

工作者處理任務

要啟動工作者，可以使用 rq worker 命令。這個命令會啟動一個工作者，並開始處理佇列中的任務。

$ rq worker
RQ worker 'rq:worker:abydos.87255' started, version 0.8.2
Cleaning registries for queue: default
*** Listening on default...
default: builtins.sum([42, 43]) (631eeb5f-66f2-44f8-817b-2d
default: Job OK (631eeb5f-66f2-44f8-817b-2d0ef285f8e3)
Result is kept for 500 seconds (3)
*** Listening on default...

工作者會處理佇列中的任務，並將結果儲存在 Redis 中。

結果儲存和過期

RQ 支援設定結果的儲存時間（TTL）。這意味著結果會在設定的時間後自動過期和刪除。例如：

q = Queue(name="http", connection=Redis(), ttl=60, result_ttl=300)

這個例子設定了佇列的 TTL 為 60 秒，結果的 TTL 為 300 秒。

注意事項

  • RQ 使用 pickle 來序列化任務，這意味著生產者和工作者必須使用相同的 Python 版本和程式碼版本。
  • 任務函式必須是純函式，不依賴於任何全域性變數或上下文。
  • RQ 支援多個佇列和工作者，可以根據需要進行擴充套件。

圖表翻譯：

  graph LR
    A[生產者] -->| 推播任務 | B[佇列]
    B -->| 處理任務 | C[工作者]
    C -->| 儲存結果 | D[Redis]
    D -->| 過期和刪除 | E[結果]

這個圖表展示了 RQ 的基本工作流程：生產者推播任務到佇列中，工作者處理任務，並將結果儲存在 Redis 中。結果會在設定的時間後過期和刪除。

使用RQ和Celery進行任務佇列管理

任務佇列管理是許多應用程式中的一個重要組成部分，能夠幫助我們實作非同步和分散式工作流程。下面，我們將介紹兩種常用的任務佇列管理工具：RQ和Celery。

RQ

RQ（Redis Queue）是一種根據Redis的任務佇列管理工具。它提供了一種簡單的方式來實作非同步工作流程。使用RQ，可以輕鬆地建立和管理任務佇列，並且可以使用Redis作為broker。

以下是使用RQ的基本步驟：

  1. 安裝RQ：可以使用pip安裝RQ。
  2. 建立任務佇列：可以使用rq worker命令建立任務佇列。
  3. 新增任務：可以使用rq put命令新增任務到佇列中。
  4. 執行任務：可以使用rq worker命令執行任務佇列中的任務。

RQ提供了一種簡單的方式來實作非同步工作流程，但是它也有一些限制。例如，它只能使用Redis作為broker，且不支援多個broker。

Celery

Celery是一種更為強大的任務佇列管理工具。它支援多個broker，包括Redis、RabbitMQ和Amazon SQS等。同時，Celery也支援多個後端儲存結果，包括Redis、MongoDB、SQL資料函式庫等。

以下是使用Celery的基本步驟：

  1. 安裝Celery：可以使用pip安裝Celery。
  2. 建立任務佇列：可以使用celery worker命令建立任務佇列。
  3. 新增任務：可以使用celery put命令新增任務到佇列中。
  4. 執行任務：可以使用celery worker命令執行任務佇列中的任務。

Celery提供了一種更為強大的方式來實作非同步工作流程。它支援多個broker和後端儲存結果，且可以實作更為複雜的工作流程。

Celery任務

在Celery中，任務是函式，可以被非同步呼叫。當呼叫任務時，Celery會將其放入broker佇列中，然後遠端工作者會執行該任務，並將結果儲存到後端。

以下是Celery任務的基本結構：

import celery

app = celery.Celery('celery-task',
                    broker='redis://localhost',
                    backend='redis://localhost')

@app.task
def add(x, y):
    return x + y

在這個例子中，add函式是一個Celery任務，可以被非同步呼叫。當呼叫add函式時，Celery會將其放入broker佇列中，然後遠端工作者會執行該任務，並將結果儲存到後端。

Celery結果

當執行Celery任務時，會傳回一個AsyncResult物件。該物件提供了一種方式來取得任務的結果。

result = add.delay(2, 2)
print(result.get())  # 輸出：4

在這個例子中，add.delay(2, 2)會將add函式放入broker佇列中，然後傳回一個AsyncResult物件。然後，可以使用get()方法來取得任務的結果。

非同步任務處理：Celery 的應用

在現代軟體開發中，非同步任務處理是一個非常重要的概念。它允許我們在不阻塞主程式的情況下執行耗時任務，從而提高系統的效率和可擴充套件性。Celery 是一種流行的非同步任務處理框架，廣泛用於 Python 開發中。

Celery 的基本概念

Celery 是一個分散式任務佇列，它允許我們將任務分配給多個作業員（worker）執行。作業員可以執行在不同的機器上，從而實作任務的平行執行。Celery 支援多種訊息佇列，包括 RabbitMQ、Redis 等。

Celery 的工作原理

當我們建立一個 Celery 任務時，Celery 會將任務新增到訊息佇列中。然後，作業員會從訊息佇列中取出任務並執行它。當任務完成後，作業員會將結果傳回給 Celery，然後 Celery 會將結果儲存在後端儲存中。

示例程式碼

以下是一個簡單的 Celery 任務示例：

from celery import Celery

app = Celery('celery-task', broker='redis://localhost:6379//')

@app.task
def add(x, y):
    return x + y

result = add.delay(4, 4)
print("Task state: %s" % result.state)
print("Result: %s" % result.get())
print("Task state: %s" % result.state)

在這個示例中，我們建立了一個 Celery 應用程式，並定義了一個 add 任務。然後，我們使用 delay 方法將任務新增到訊息佇列中。最後，我們列印預出任務的狀態和結果。

啟動作業員

要啟動作業員，我們需要使用 Celery 的命令列工具：

$ celery worker --app celery-task

這會啟動一個作業員，作業員會從訊息佇列中取出任務並執行它。

結果

當作業員啟動後，程式會列印預出任務的狀態和結果：

Task state: PENDING
Result: 8
Task state: SUCCESS

這表明任務已經完成，並且結果已經傳回給 Celery。

5.2.1 處理任務失敗

任務執行可能會失敗，因此正確地處理失敗情況至關重要。任務通常依賴外部服務，例如遠端資料函式庫或REST API。連線失敗可能是暫時的，因此以適當的方式處理失敗並稍後重試是更好的做法。

5.4 範例：具有重試支援的 Celery 任務

import celery

app = celery.Celery('celery-task-retry',
                    broker='redis://localhost',
                    backend='redis://localhost')

@app.task(bind=True, retry_backoff=True, retry_kwargs={'max_retries': 5})
def add(self, x, y):
    try:
        return x + y
    except OverflowError as exc:
        self.retry(exc=exc)

if __name__ == '__main__':
    result = add.delay(4, 4)
    print("Task state: %s" % result.state)
    print("Result: %s" % result.get())
    print("Task state: %s" % result.state)

在範例 5.4 中，實作了一個簡單的重試邏輯，如果發生 OverflowError 時會重試。retry_backoff 引數確保 Celery 使用指數退避演算法在延遲之間重試，而 max_retries 引數則限制重試次數不超過五次。限制重試次數很重要，因為您永遠不希望因為永久性錯誤或 bug 而導致工作卡在佇列中。

重試任務會以相同的引數呼叫相同的函式，這意味著這些任務的最佳設計是完全冪等的。如果任務具有副作用並在執行中途失敗，稍後處理任務重複執行可能會更加複雜。例如，以下程式碼：

@app.task(autoretry_for=(DatabaseError,))
def record_visit(user_id):
    database.increment_visitor_counter()
    remote_api.record_last_visit_time(user_id)

如果在呼叫 remote_api.record_last_visit_time 時發生錯誤，訪客計數器已經被遞增。當任務被重試時，計數器將再次被遞增，從而導致訪客被計算兩次。這樣的任務應該被重寫，以便如果多次執行，最終系統產生的結果相同。

忽略任務結果

由玄貓所述，Celery 將結果儲存到指定的後端儲存中。但是，有時候任務沒有有趣的傳回值。在這種情況下，可以傳遞 ignore_result=True 到 app.task 裝飾器，以確保結果被忽略。

RQ 和 Celery 作為 Python 生態中兩種主流的任務佇列框架，各有其優勢和適用場景。深入剖析兩者的架構和功能後，可以發現 Celery 在功能豐富度、擴充套件性和靈活性方面更勝一籌，尤其在處理複雜工作流程、多種 broker 支援以及結果儲存方面表現出色。而 RQ 則以其簡潔易用和輕量級特性，更適合於小型專案或對 Redis 生態高度依賴的應用。

透過多維比較分析，RQ 的輕量級特性降低了學習曲線和佈署成本，但功能相對簡潔，缺乏 Celery 提供的精細控制和監控能力。Celery 雖然功能強大，但也引入了更高的複雜度，需要更全面的組態和管理。技術限制深析顯示，RQ 受限於 Redis 作為唯一 broker，而 Celery 則需考量其本身的資源消耗和組態複雜度。

從技術演進預測來看，隨著微服務架構和分散式系統的普及，任務佇列的重要性將持續提升。Celery 的生態系統發展更為活躍，預計未來會持續整合更多新興技術和功能，例如更精細的任務排程、更豐富的監控指標以及更便捷的佈署方案。對於重視長期穩定性和可擴充套件性的企業，採用 Celery 將是更具前瞻性的選擇。

玄貓認為，開發團隊應根據專案規模、複雜度和團隊技術堆疊的實際情況，權衡 RQ 和 Celery 的優缺點，選擇最適合的任務佇列解決方案。在資源有限的條件下，優先將精力集中於理解和解決任務冪等性、錯誤處理和結果儲存策略等核心挑戰，才能最大程度地發揮任務佇列的效能和價值。

玄貓

玄貓

技術愛好者，專注於分享程式開發、雲端技術與 AI 應用的心得體會。