2. 程式人生 > >python celery多worker、多隊列、定時任務

python celery多worker、多隊列、定時任務

阿新 • • 發佈：2018-12-15
end fig 多隊列 erb minutes copy src span task

多worker、多隊列

celery是一個分布式的任務調度模塊，那麽怎麽實現它的分布式功能呢，celery可以支持多臺不同的計算機執行不同的任務或者相同的任務。

如果要說celery的分布式應用的話，就要提到celery的消息路由機制，提到AMQP協議。

簡單理解：

可以有多個"消息隊列"（message Queue），不同的消息可以指定發送給不同的Message Queue，

而這是通過Exchange來實現的，發送消息到"消息隊列"中時，可以指定routiing_key，Exchange通過routing_key來吧消息路由（routes）到不同的"消息隊列"中去。

技術分享圖片

exchange 對應 一個消息隊列(queue)，即：通過"消息路由"的機制使exchange對應queue，每個queue對應每個worker。

下面我們來看一個列子：

技術分享圖片 
vi tasks.py

#!/usr/bin/env python
#-*- coding:utf-8 -*-
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
技術分享圖片

編寫配置文件，配置文件一般單獨寫在一個文件中。

技術分享圖片 
vi celeryconfig.py

#!/usr/bin/env python
#-*- coding:utf-8 -*-

from kombu import Exchange,Queue

BROKER_URL = "redis://47.106.106.220:5000/1" 
CELERY_RESULT_BACKEND = "redis://47.106.106.220:5000/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 = {
‘tasks.taskA‘:{"queue":"for_task_A","routing_key":"for_task_A"},
‘tasks.taskB‘:{"queue":"for_task_B","routing_key":"for_task_B"}
}
技術分享圖片

遠程客戶端上編寫測試腳本

技術分享圖片 
vi test.py

from tasks import *
re1 = taskA.delay(100, 200)
print(re1.result)
re2 = taskB.delay(1, 2, 3)
print(re2.result)
re3 = add.delay(1, 2)
print(re3.status)
技術分享圖片

啟動兩個worker來分別指定taskA、taskB，開兩個窗口分別執行下面語句。

celery -A tasks worker -l info -n workerA.%h -Q for_task_A

celery -A tasks worker -l info -n workerB.%h -Q for_task_B

遠程客戶端上執行腳本可以看到如下輸出：

python test.py 
300
6
PENDING

在taskA所在窗口可以看到如下輸出：

技術分享圖片 
.......
.......
.......
task_A

[tasks]
  . tasks.add
  . tasks.taskA
  . tasks.taskB

[2018-05-27 19:23:49,235: INFO/MainProcess] Connected to redis://47.106.106.220:5000/1
[2018-05-27 19:23:49,253: INFO/MainProcess] mingle: searching for neighbors
[2018-05-27 19:23:50,293: INFO/MainProcess] mingle: all alone
[2018-05-27 19:23:50,339: INFO/MainProcess] [email protected] ready.
[2018-05-27 19:23:56,051: INFO/MainProcess] sync with [email protected]
[2018-05-27 19:24:28,855: INFO/MainProcess] Received task: tasks.taskA[8860e78a-b82b-4715-980c-ae125dcab2f9]  
[2018-05-27 19:24:28,872: INFO/ForkPoolWorker-1] Task tasks.taskA[8860e78a-b82b-4715-980c-ae125dcab2f9] succeeded in 0.0162177120219s: 300
技術分享圖片

在taskB所在窗口可以看到如下輸出：

技術分享圖片 
.......
.......
.......
task_B
[tasks]
  . tasks.add
  . tasks.taskA
  . tasks.taskB

[2018-05-27 19:23:56,012: INFO/MainProcess] Connected to redis://47.106.106.220:5000/1
[2018-05-27 19:23:56,022: INFO/MainProcess] mingle: searching for neighbors
[2018-05-27 19:23:57,064: INFO/MainProcess] mingle: sync with 1 nodes
[2018-05-27 19:23:57,064: INFO/MainProcess] mingle: sync complete
[2018-05-27 19:23:57,112: INFO/MainProcess] [email protected] ready.
[2018-05-27 19:24:33,885: INFO/MainProcess] Received task: tasks.taskB[5646d0b7-3dd5-4b7f-8994-252c5ef03973]  
[2018-05-27 19:24:33,910: INFO/ForkPoolWorker-1] Task tasks.taskB[5646d0b7-3dd5-4b7f-8994-252c5ef03973] succeeded in 0.0235358460341s: 6
技術分享圖片

我們看到狀態是PENDING，表示沒有執行，這個是因為沒有celeryconfig.py文件中指定改route到哪一個Queue中，所以會被發動到默認的名字celery的Queue中，但是我們還沒有啟動worker執行celery中的任務。下面，我們來啟動一個worker來執行celery隊列中的任務。

celery -A tasks worker -l info -n worker.%h -Q celery

再次在遠程客戶端執行test.py，可以看到結果執行成功，並且剛新啟動的worker窗口有如下輸出：

技術分享圖片 
.......
.......
.......
[tasks]
  . tasks.add
  . tasks.taskA
  . tasks.taskB

[2018-05-27 19:25:44,596: INFO/MainProcess] Connected to redis://47.106.106.220:5000/1
[2018-05-27 19:25:44,611: INFO/MainProcess] mingle: searching for neighbors
[2018-05-27 19:25:45,660: INFO/MainProcess] mingle: sync with 2 nodes
[2018-05-27 19:25:45,660: INFO/MainProcess] mingle: sync complete
[2018-05-27 19:25:45,711: INFO/MainProcess] [email protected] ready.
[2018-05-27 19:25:45,868: INFO/MainProcess] Received task: tasks.add[f9c5ca2b-623e-4c0a-9c45-a99fb0b79ed5]  
[2018-05-27 19:25:45,880: INFO/ForkPoolWorker-1] Task tasks.add[f9c5ca2b-623e-4c0a-9c45-a99fb0b79ed5] succeeded in 0.0107084610499s: 3
技術分享圖片

Celery與定時任務

在celery中執行定時任務非常簡單，只需要設置celery對象中的CELERYBEAT_SCHEDULE屬性即可。
下面我們接著在celeryconfig.py中添加CELERYBEAT_SCHEDULE變量：

技術分享圖片 
cat celeryconfig.py

#!/usr/bin/env python
#-*- coding:utf-8 -*-

from kombu import Exchange,Queue

BROKER_URL = "redis://47.106.106.220:5000/1" 
CELERY_RESULT_BACKEND = "redis://47.106.106.220:5000/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 = {
‘tasks.taskA‘:{"queue":"for_task_A","routing_key":"for_task_A"},
‘tasks.taskB‘:{"queue":"for_task_B","routing_key":"for_task_B"}
}

# 新增加的定時任務部分
CELERY_TIMEZONE = ‘UTC‘
CELERYBEAT_SCHEDULE = {
    ‘taskA_schedule‘ : {
        ‘task‘:‘tasks.taskA‘,
        ‘schedule‘:2,
        ‘args‘:(5,6)
    },
    ‘taskB_scheduler‘ : {
        ‘task‘:"tasks.taskB",
        "schedule":10,
        "args":(10,20,30)
    },
    ‘add_schedule‘: {
        "task":"tasks.add",
        "schedule":5,
        "args":(1,2)
    }
}
技術分享圖片

還是按之前啟動三個worker

celery -A tasks worker -l info -n workerA.%h -Q for_task_A

celery -A tasks worker -l info -n workerB.%h -Q for_task_B

celery -A tasks worker -l info -n worker.%h -Q celery

啟動定時任務

技術分享圖片 
[root@izwz920j4zsv1q15yhii1qz scripts]# celery -A tasks beat
celery beat v4.1.1 (latentcall) is starting.
__    -    ... __   -        _
LocalTime -> 2018-05-27 19:39:29
Configuration ->
    . broker -> redis://47.106.106.220:5000/1
    . loader -> celery.loaders.app.AppLoader
    . scheduler -> celery.beat.PersistentScheduler
    . db -> celerybeat-schedule
    . logfile -> [stderr]@%WARNING
    . maxinterval -> 5.00 minutes (300s)
技術分享圖片

在之前啟動worker的三個窗口分別可以看到定時任務正在運行：

技術分享圖片 
celery -A tasks worker -l info -n workerA.%h -Q for_task_A

[2018-05-27 19:41:27,432: INFO/ForkPoolWorker-1] Task tasks.taskA[60f41780-c9a2-477b-be46-6620ef07631f] succeeded in 0.00289130600868s: 11
[2018-05-27 19:41:29,428: INFO/MainProcess] Received task: tasks.taskA[27220f52-dde2-471a-a87c-3f533d67217c]
......
......

celery -A tasks worker -l info -n workerB.%h -Q for_task_B

[2018-05-27 19:41:18,420: INFO/ForkPoolWorker-1] Task tasks.taskB[b6f9aee3-e6b4-4f10-9428-457d9bb844cf] succeeded in 0.00282042898471s: 60
[2018-05-27 19:41:28,416: INFO/MainProcess] Received task: tasks.taskB[44dfea0b-b725-4874-bea2-9b66e8da573b]
......
......


celery -A tasks worker -l info -n worker.%h -Q celery 

[2018-05-27 19:41:23,428: INFO/ForkPoolWorker-1] Task tasks.add[315a9cca-3c95-4517-9289-2ece15cd46a4] succeeded in 0.00355823297286s: 3
[2018-05-27 19:41:28,423: INFO/MainProcess] Received task: tasks.add[c4a1b2c7-ecb7-4af4-85c1-a341b3ec6726]
......
......
技術分享圖片

python celery多worker、多隊列、定時任務

相關推薦

進擊的Python【第九章】：paramiko模塊、線程與進程、各種線程鎖、queue隊列、生產者消費者模型

password locking form maxsize 廁所 sorted [0 hostname nbsp 一、paramiko模塊 他是什麽東西？ 　　paramiko模塊是用python語言寫的一個模塊，遵循SSH2協議，支持以加密和認證的方式，進行遠程服務器的連

.NET Core中使用IHostedService結合隊列執行定時任務

組隊 屬性獲取 響應時間 取數據 reg ddt sage 過多 ret 最近遇到了這樣的場景：每隔一段時間，需要在後臺使用隊列對一批數據進行業務處理。 Quartz.NET是一種選擇，在 .NET Core中，可以使用IHostedService執行後臺定時任務。在本

python celery多worker、多隊列、定時任務

end fig 多隊列 erb minutes copy src span task 多worker、多隊列 celery是一個分布式的任務調度模塊，那麽怎麽實現它的分布式功能呢，celery可以支持多臺不同的計算機執行不同的任務或者相同的任務。 如果要說celery的分布

25、Java並發性和多線程-阻塞隊列

多線程 throws clas cep ted this return exception exceptio 以下內容轉自http://ifeve.com/blocking-queues/： 阻塞隊列與普通隊列的區別在於，當隊列是空的時，從隊列中獲取元素的操作將會被阻塞，

Python 多線程同步隊列模型

並且 highlight 多線程 use lib star 保存 env module Python 多線程同步隊列模型 我面臨的問題是有個非常慢的處理邏輯（比如分詞、句法），有大量的語料，想用多線程來處理。 這一個過程可以抽象成一個叫“同步隊列”的模型。 具體來講，有

python 多線程與隊列

pro 錯誤 rotation 變化 lam exce num 開始 dex 各位好，之前寫了多線程，但是在實際的生產中，往往情況比較復雜，要處理一批任務（比如要處理列表中所有元素），這時候不可能創建很多的線程，線程過多反而不好，還會造成資源開銷太大，這時候想到了隊列。

進程對象的其他方法、守護進程、使用多進程實現 socket tcp協議 server端的並發（搶票程序）、隊列、進程之間的通信（IPC）

nat pip nec 中間 表示 pipe 數據類型 參數 bsp # 進程對象的其他方法 from multiprocessing import Process import time class MyProcess(Process): def __

使用goroutine+channel和java多線程+queue隊列的方式開發各有什麽優缺點？

多線程 nec ref 線程安全 dex ood .net 結構 使用 我感覺很多項目使用java或者c的多線程庫+線程安全的queue數據結構基本上可以實現goroutine+channel開發能達到的需求，所以請問一下為什麽說golang更適合並發服務端的開發呢？使用g

基數排序之多keyword排序運用隊列

printf mod n) key sort article name str oid 源碼例如以下： #include <stdlib.h> #include <stdio.h> typedef struct QUEUEnode* li

C# 多線程與隊列操作小練刀

art nds b- ext next else and 一把手 timespan 之前寫Web沒有應用到多線程與隊列的知識，寫了個小程序，練了一把手，模擬商品搶購，代碼如下： class Program { public static Ran

python並發編程基礎之守護進程、隊列、鎖

lock pytho locking ipc 讀取 多進程 sin 默認值 lee 並發編程2 1.守護進程 什麽是守護進程？ 表示進程A守護進程B，當被守護進程B結束後，進程A也就結束。 from multiprocessing import Process imp

Hive裡實現行資料的拆分、一行轉多行、行轉列、explode使用

    在做資料轉換、清洗的過程中，經常會對行進行拆分、一行資料拆分為多行資料，下面是簡單的例子:    如圖表裡有如下資料：最終要得到結果如下：先用split進行分割得到一個數組，在用explode拆分陣列。LATERAL VIEW explode(split(b,',')

Day28：Event對象、隊列、multiprocessing模塊

effective emp 接受 pytho 提醒 info event對象 fig 提示 一、Event對象 　　線程的一個關鍵特性是每個線程都是獨立運行且狀態不可預測。如果程序中的其他線程需要通過判斷某個線程的狀態來確定自己下一步的操作，這時線程同步問題就 會變得非常棘

轉---隊列、堆、棧、堆棧的區別

都是 fix 包括 article manage class alloc 速度 -a 隊列、堆、棧、堆棧的區別 堆棧：先進後出（就像放在箱子的衣服，先放進去的後拿出來） 隊列：先進先出（就像一條路，有一個入口和一個出口，先進去的就可以先出去） 　　進程

jQuery源代碼解析（3）—— ready載入、queue隊列

else ng- settime eve ref promise ont 出隊 function ready、queue放在一塊寫，沒有特殊的意思，僅僅是相對來說它倆可能源代碼是最簡單的了。ready是在dom載入完畢後。以最高速度觸發，非常實用。que

第十一篇：消息隊列、緩存

dir hash 廣播 body 消息分發 ash rec 模式 edi 一、消息隊列簡紹 二、RabbitMQ基本實列： 三、RabbitMQ消息分發輪詢 四、消息持久化 五、Fanout廣播模式 六、Direct廣播模式 七、Redis基本操作 八、Redis Hash

第二章、最基本的數據結構——隊列、棧、鏈表

算法基礎還記得大學算法課，老師提過，程序=算法+數據結構，不一定非常正確，但表明了一個事實：算法和數據結構的重要性。 在做題的過程，用經典的算法+合適的數據結構，一道題基本也完成了80%，考慮下取值範圍的限制以及輸入輸出條件，就差不多了。第二章、最基本的數據結構——隊列、棧、鏈表

linux驅動---等待隊列、工作隊列、Tasklets【轉】

kylin 重新 工作 tail 打開 進程控制 軟中斷 聯系 try 轉自：https://blog.csdn.net/ezimu/article/details/54851148 概述: 等待隊列、工作隊列、Tasklet都是linux驅動很重要的API,下面主要從

python數據結構之雙隊列（二）

self for __init__ pri solid pen odin __name__ urn 書接上文，雙端隊列區別於單隊列為：雙端隊列可以對隊列頭和尾部同時進行操作，單隊列不行#coding:utf-8 class DoubleQueue(object):

消息隊列、socket(UDP)實現簡易聊天系統

sed sele perror 語句 sprintf display clu 隊列 success 　　前言： 　　最近在學進程間通信，所以做了一個小項目練習一下。主要用消息隊列和socket(UDP)實現這個系統，並數據庫存儲數據，對C語言操作數據庫不熟悉的可以參照我的