Python Celery分布式任務隊列的安裝方法

本篇內容介紹了“Python Celery分布式任務隊列的安裝方法”的有關知識，在實際案例的操作過程中，不少人都會遇到這樣的困境，接下來就讓小編帶領大家學習一下如何處理這些情況吧！希望大家仔細閱讀，能夠學有所成！

Celery是一個基于Python編寫的分布式任務隊列(Distributed Task Queue), 通過對Celery進行簡單操作就可以實現任務(耗時任務, 定時任務)的異步處理

一. Celery的安裝

Celery4.0版本開始,不支持windows平臺

1.1 通過pip方式安裝celery

pip install -U "Celery[redis]"

注意事項:

　　在windows上安裝后,可能會出現如下報錯:

ValueError: '__name__' in __slots__ conflicts with class variable

 　　此時先卸載celery, 然后嘗試通過如下命令重新進行安裝

pip install -U https://github.com/celery/py-amqp/zipball/master
pip install -U https://github.com/celery/billiard/zipball/master
pip install -U https://github.com/celery/kombu/zipball/master
pip install -U https://github.com/celery/celery/zipball/master
pip install -U "Celery[redis]"

1.2 給celery創建一個軟連接

ln -s ~/.venv/project_dj/bin/celery /usr/bin/celery

1.3 執行celery命令

[root@localhost ~]$ celery --helpOptions:  -A, --app APPLICATION  -b, --broker TEXT  --result-backend TEXT  --loader TEXT  --config TEXT  --workdir PATH  -C, --no-color  -q, --quiet  --version  --help                 Show this message and exit.Commands:
  amqp     AMQP Administration Shell.
  beat     Start the beat periodic task scheduler.
  call     Call a task by name.
  control  Workers remote control.
  events   Event-stream utilities.
  graph    The ``celery graph`` command.
  inspect  Inspect the worker at runtime.
  list     Get info from broker.
  logtool  The ``celery logtool`` command.
  migrate  Migrate tasks from one broker to another.
  multi    Start multiple worker instances.
  purge    Erase all messages from all known task queues.
  report   Shows information useful to include in bug-reports.
  result   Print the return value for a given task id.
  shell    Start shell session with convenient access to celery symbols.
  status   Show list of workers that are online.
  upgrade  Perform upgrade between versions.  worker   Start worker instance.

二. Celery的基本使用

2.1 創建celery應用, 并定義任務

# -*- coding: utf-8 -*-# @Time    : 2021/5/24 11:20# @Author  : chinablue# @File    : task.pyfrom celery import Celery# 創建一個app(Celery實例),作為所有celery操作的切入點broker_url = f"redis://:123456@127.0.0.1:6379/5"backend_url = f"redis://:123456@127.0.0.1:6379/6"app = Celery("tasks", broker=broker_url, backend=backend_url)# 定義一個任務@app.taskdef add(x, y):return x + y

事項說明：

　　1) 創建Celery實例時，需要指定一個消息代理(broker)來接收和發送任務消息. 本文使用的是Redis(docker redis搭建)

　　2) broker和backend參數的格式: redis://:password@hostname:port/db_number

2.2 啟動celery worker服務端

celery -A tasks worker --loglevel=INFO

事項說明：

　　1) 在生產環境中, 會使用supervisor工具將celery服務作為守護進程在后臺運行

2.3 調用任務

打開終端, 進入python命令行模式:

>>> result = add.delay(4, 4)
>>> result = add.apply_async((4, 4), countdown=5)

事項說明：

　　1) add.apply_async((4, 4)) 可以簡寫為 add.delay(4, 4)

　　2) add.apply_async((4, 4), countdown=5) 表示任務發出5秒后再執行

2.4 追蹤任務信息

若想獲取每個任務的執行信息,在創建Celery實例時, 需要指定一個后端(backend). 本文使用的是Redis(docker redis搭建)

result = add.delay(4, 4)        
result.ready()       # 任務狀態: 進行中, 已完成result.failed()      # 任務完成, 任務失敗result.successful()  # 任務完成, 任務成功result.state         # 任務狀態: PENDING, STARTED, SUCCESSresult.get()         # 獲取任務的返回值        result.get(timeout=10)
result.get(propagate=False)  # 如果任務引發了異常, propagate=False表示異常不會被拋出來(默認情況會拋出來)result.id            # 任務id

注意事項：

　　1) 在celery中，如果想配置backend參數，有如下三種方式

# -*- coding: utf-8 -*-# @Time    : 2021/5/24 11:20# @Author  : chinablue# @File    : task.pyfrom celery import Celery# 創建一個app(Celery實例),作為所有celery操作的切入點broker_url = f"redis://:123456@127.0.0.1:6379/5"backend_url = f"redis://:123456@127.0.0.1:6379/6"app = Celery("tasks", broker=broker_url, backend=backend_url)# 定義一個任務@app.taskdef add(x, y):return x + y

方式1: 實例化Celery時傳入

# -*- coding: utf-8 -*-# @Time    : 2021/5/24 11:20# @Author  : chinablue# @File    : task.pyfrom celery import Celery

broker_url = f"redis://:123456@127.0.0.1:6379/5"backend_url = f"redis://:123456@127.0.0.1:6379/6"app = Celery("tasks")

app.conf.update({"broker_url": broker_url,"result_backend": backend_url,
})# 定義一個任務@app.taskdef add(x, y):return x + y

方式2: 通過conf的update方法

# -*- coding: utf-8 -*-# @Time    : 2021/5/24 11:20# @Author  : chinablue# @File    : task.pyfrom celery import Celery

broker_url = f"redis://:123456@127.0.0.1:6379/5"backend_url = f"redis://:123456@127.0.0.1:6379/6"app = Celery("tasks")

app.conf.broker_url = broker_url
app.conf.result_backend = backend_url# 定義一個任務@app.taskdef add(x, y):return x + y

“Python Celery分布式任務隊列的安裝方法”的內容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業相關的知識可以關注億速云網站，小編將為大家輸出更多高質量的實用文章！