Python 輕量級分散式任務佇列 Huey 非同步任務實作指南

Huey 是純 Python 實作的輕量任務佇列，依賴 Redis 作為後端，支援多執行緒執行、排程任務（crontab）、週期性任務、失敗重試、結果儲存，無需 Celery 的繁重依賴，適合中小型專案快速導入非同步處理。

安裝與基本架構

pip install huey redis

三大核心元件：

• Huey 實例：佇列與結果儲存的進入點。
• 任務裝飾器：@huey.task()、@huey.periodic_task(crontab(...))。
• Consumer 執行緒：huey_consumer.py 持續輪詢佇列執行任務。

快速開始

config.py：建立 Huey 實例

from huey import RedisHuey
huey = RedisHuey('my-app', host='localhost', port=6379)

tasks.py：定義任務

from config import huey

@huey.task()
def add_numbers(a, b):
    return a + b

@huey.periodic_task(crontab(minute='0', hour='3'))
def nightly_backup():
    sync_all_data()

main.py：送入任務（生產者）

from config import huey
from tasks import add_numbers

if __name__ == '__main__':
    result = add_numbers(10, 20)  # 非同步送入，回傳 AsyncData
    print(f"任務已送入，結果: {result.get()}")  # 阻塞等待結果
    # 或排程延遲執行
    add_numbers.schedule(args=(5, 5), delay=60)  # 60秒後執行

啟動 Consumer（消費者）

huey_consumer.py config.huey -w 4 -l /var/log/huey.log

• -w 4：4 個工作執行緒。
• -l：日誌檔案。
• -p：啟用週期性任務排程執行緒（預設開啟）。

進階功能

結果儲存

from huey.backends.redis_backend import RedisDataStore
result_store = RedisDataStore('results', host='localhost', port=6379)
huey = RedisHuey('my-app', result_store=result_store)

失敗重試

@huey.task(retries=3, retry_delay=10)
def flaky_task():
    if random.random() < 0.7:
        raise Exception('暫時性失敗')
    return '成功'

任務撤銷

result = add_numbers.schedule(args=(100,), delay=300)
result.revoke()  # 取消尚未執行的任務

優先級（需自行實作佇列邏輯）

Huey 本身不內建優先級佇列，可透過多個 Huey 實例對應不同 Redis List 實現。

與 Celery/RQ/Mrq 比較

Huey 以「輕量、零額外依賴、API 直觀」為核心優勢，是 Python 專案導入非同步任務的優質選擇。