一、關於celery
芹菜celery是一個python實現的異步任務隊列,可以用於爬蟲、web后台查詢、計算等等。通過任務隊列,當一個任務來臨時不再傻傻等待。
他的架構如下:
- Broker
我們的生產者創建任務后會進入celery的任務調度隊列中間件Broker,Broker通過調度規則將消息(任務)調度消息隊列,Broker依賴第三方隊列消息代理如rabbitmq
、redis
等。
- Worker
廣大勞動者,盯着消息隊列,當隊列中有消息時把它拿過來給處理了。
- Backend
用於結果存儲經worker處理的結果,比如常用的數據庫等。
使用celery
在本文中咱們使用rabbitmq
(celery推薦)作為消息代理中間件。
我們創建的celery目錄如下
learn_celery/
...celery_env/
...celery.py
...my_task1.py
...my_task2.py
...task1_run.py
...task2_run.py
1. 創建虛擬環境並安裝celery、flower(web監控),這里不做贅述。
2.安裝咱們的消息隊列中間件rabbitmq
這里以docker的方式運行並配置,指定主機名為rabbit
(rabbitmq是以主機名來訪問的,所以這是必須的),容器名稱為celery_rabbitmq
docker run -d -p 5672:5672 -h rabbit --name celery_rabbitmq rabbitmq
添加用於celery訪問的用戶,以及配置configure
、write
和read
權限,在下面我們配置rabbit_user擁有所有配置、寫入和讀取權限。
docker exec -it celery_rabbitmq rabbitmqctl add_user rabbit_user rabbit_pass
docker exec -it celery_rabbitmq rabbitmqctl add_vhost rabbit_vhost
docker exec -it celery_rabbitmq rabbitmqctl set_user_tags rabbit_user celery
docker exec -it celery_rabbitmq rabbitmqctl set_permissions -p rabbit_vhost rabbit_user ".*" ".*" ".*"
3.創建celery應用
#celery.py
from celery import Celery
broker_rabbitmq="amqp://rabbit_user:rabbit_pass@i-k9pwet2d/rabbit_vhost"
app=Celery("learn_celery",broker=broker_rabbitmq,backend="rpc://",include=["learn_celery.my_task2","learn_celery.my_task2"])
我們通過創建app來實例化Celery,項目包的名稱為learn_celery
,通過broker_rabbitmq
來連接rabbitmq,rabbitmq的amqp協議格式為
amqp://userid:password@hostname:port/virtual_host
由於我們是在docker中啟動的rabbitmq,所以我們的hostname應該為宿主機的hostname。
指定后端通過rpc回傳數據,include加載帶worker處理的任務learn_celery.my_task1
、learn_celery.my_task2
4.創建兩個任務(消息)
#my_task1.py
from .celery import app
import time
@app.task
def args_add1(x,y):
print("start task no.1 now!")
time.sleep(10)
print("task no.1 end!")
return x+y
#my_task12.py
from .celery import app
import time
@app.task
def args_add2(x,y):
print("start task no.2 now!")
time.sleep(20)
print("task no.2 end!")
return x+y
在這里我們導入了celery中的app,並用它來裝飾我們的方法args_add
,在args_add中模擬任務處理時間分別為10s、20s然后返回結果。
5.發送任務給celery
#tasks1_run.py
from .my_task1 import args_add1
import time
reslut=args_add1.delay(11,22)
print("task over?{}".format(reslut.ready()))
print("task reslut:{}".format(reslut.result))
time.sleep(15)
print("task over?{}".format(reslut.ready()))
print("task reslut:{}".format(reslut.result))
#tasks2_run.py
from .my_task2 import args_add2
import time
reslut=args_add2.delay(33,44)
print("task over?{}".format(reslut.ready()))
print("task reslut:{}".format(reslut.result))
time.sleep(25)
print("task over?{}".format(reslut.ready()))
print("task reslut:{}".format(reslut.result))
關於任務的delay
,官方文檔(參考)是這樣描述的,我把它理解為發送任務給celery或者celery調用待進來的任務。
reslut.ready()
返回任務執行是否執行完成True
or False
reslut.result
返回任務執行結果
我們在任務進入celery和結束分別檢查一次。
二、看看結果
1.啟動worker
進入learn_celery的父目錄。啟動learn_celery的這個應用worker,並指定並發數為10個
celery -A learn_celery worker --loglevel=info --concurrency=10
若celery連接rabbitmq正常,我們可以看到如下的info
2.執行任務
為了便於觀察,我們另外開啟一個窗口2,到learn_celery父目錄運行task1_run模塊
python -m learn_celery.tasks1_run
開啟窗口3,到learn_celery父目錄運行task2_run模塊
python -m learn_celery.tasks2_run
可以看到經過各自任務的等待時間后,兩個任務都順利執行結束,並得到結果,接下來我們到worker上看一下info
由於celery的並發性,收到任務馬上被調入執行,任務1耗時10s結果為33,任務2耗時20s結果為77
三、使用Flower監控celery
1.啟動flower
celery -A learn_celery flower
2. 查看web監控 http://ip:5555
在Tasks
中可以查看到當前任務隊列的狀態、參數、接收和啟動、執行時間。
在Dashborad
中查看當前worker節點的相關信息
文章有不足的地方歡迎指出。
歡迎收藏、點贊、提問。關注頂級飲水機管理員,除了管燒熱水,有時還做點別的。
NEXT
-
celery的深入了解
-
celery在django中的使用