44. Python Celery多實例 定時任務
celery可以支持多臺不同的計算機執行不同的任務或者相同的任務。
如果要說celery的分布式應用的話,就要提到celery的消息路由機制,提到AMQP協議。
具體可以查看AMQP文檔詳細了解。
簡單理解:
可以有多個"消息隊列"(message Queue),不同的消息可以指定發送給不同的Message Queue,
而這是通過Exchange來實現的,發送消息到"消息隊列"中時,可以指定routiing_key,Exchange通過routing_key來吧消息路由(routes)到不同的"消息隊列"中去。
如圖:
exchange 對應 一個消息隊列(queue),即:通過"消息路由"的機制使exchange對應queue,每個queue對應每個worker
寫個例子:
vim demon3.py
from celery import Celery app = Celery() app.config_from_object("celeryconfig") @app.task def taskA(x, y): return x * y @app.task def taskB(x, y, z): return x + y + z @app.task def add(x, y): return x + y
vim celeryconfig.py
from kombu import Queue BORKER_URL = "redis://192.168.48.131:6379/1" #1庫 CELERY_RESULT_BACKEND = "redis://192.168.48.131:6379/2" #2庫 CELERY_QUEUES = { Queue("default", Exchange("default"), routing_key = "default"), Queue("for_task_A", Exchange("for_task_A"), routing_key = "for_task_A"), Queue("for_task_B", Exchange("for_task_B"), routing_key = "for_task_B") } #路由 CELERY_ROUTES = { "demon3.taskA":{"queue": "for_task_A", "routing_key": "for_task_A"}, "demon3.taskB":{"queue": "for_task_B", "routing_key": "for_task_B"} }
下面把兩個腳本導入服務器:
指定taskA啟動一個worker:
# celery -A demon3 worker -l info -n workerA.%h -Q for_task_A
同理:
# celery -A demon3 worker -l info -n workerB.%h -Q for_task_B
下面遠程客戶端調用:新文件
vim remote.py
from demon3 import * r1 = taskA.delay(10, 20) print (r1.result) print (r1.status) r2 = taskB.delay(10, 20, 30) time.sleep(1) prnit (r2.result) print (r2.status) #print (dir(r2)) r3 = add.delay(100, 200) print (r3.result) print (r3.status) #PENDING
看到狀態是PENDING,表示沒有執行,這個是因為沒有celeryconfig.py文件中指定改route到哪一個Queue中,所以會被發動到默認的名字celery的Queue中,但是我們還沒有啟動worker執行celery中的任務。
下面,我們來啟動一個worker來執行celery隊列中的任務
# celery -A tasks worker -l info -n worker.%h -Q celery ##默認的
可以看到這行的結果為success
print(re3.status) #SUCCESS
定時任務:
Celery 與 定時任務
在celery中執行定時任務非常簡單,只需要設置celery對象中的CELERYBEAT_SCHEDULE屬性即可。
下面我們接著在配置文件:celeryconfig.py,添加關於 CELERYBEAT_SCHEDULE 變量到腳本中去:
CELERY_TIMEZONE = 'UTC' CELERYBEAT_SCHEDULE = { 'taskA_schedule' : { 'task':'tasks.taskA', 'schedule':20, 'args':(5,6) }, 'taskB_scheduler' : { 'task':"tasks.taskB", "schedule":200, "args":(10,20,30) }, 'add_schedule': { "task":"tasks.add", "schedule":10, "args":(1,2) } }
註意格式,否則會有問題
啟動:
celery -A demon3 worker -l info -n workerA.%h -Q for_task_A
celery -A demon3 worker -l info -n workerB.%h -Q for_task_B
celery -A tasks worker -l info -n worker.%h -Q celery
celery -A demon3 beat
44. Python Celery多實例 定時任務