1. 程式人生 > >44. Python Celery多實例 定時任務

44. Python Celery多實例 定時任務

celery

celery是一個分布式的任務調度模塊,那麽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多實例 定時任務