資訊專欄INFORMATION COLUMN
摘要：是分布式任務(wù)隊(duì)列，能實(shí)時(shí)處理任務(wù)，同時(shí)支持官方文檔工作原理如下發(fā)送給從中消費(fèi)消息，并將結(jié)果存儲(chǔ)在中本文中使用的是，使用的是現(xiàn)在有兩個(gè)，分別是加法運(yùn)算和乘法運(yùn)算。假定乘法運(yùn)算的事件優(yōu)先級(jí)高事件也很多，對(duì)于加法運(yùn)算，要求每分鐘最多處理個(gè)事件。
Celery是分布式任務(wù)隊(duì)列，能實(shí)時(shí)處理任務(wù)， 同時(shí)支持task scheduling. 官方文檔
Celery工作原理如下：
celery client發(fā)送message給broker
worker 從broker中消費(fèi)消息，并將結(jié)果存儲(chǔ)在result_end中
本文中使用的broker是Rabbit MQ，result_end使用的是Redis.
Scenario現(xiàn)在有兩個(gè)task，分別是加法運(yùn)算和乘法運(yùn)算。假定乘法運(yùn)算的事件優(yōu)先級(jí)高&事件也很多，對(duì)于加法運(yùn)算，要求每分鐘最多處理10個(gè)事件。
框架Celery Worker：
在2 臺(tái)server上部署worker，其中：
server1上的worker處理queue priority_low和priority_high上的事件
server2上的worker只處理priority_high上的事件
Celery Client：在應(yīng)用中調(diào)用
Rabbit MQ：在server3上啟動(dòng)
Redis：在localhost啟動(dòng)
Code tasks.py & callback對(duì)兩個(gè)任務(wù)加上callback的處理，如果成功，打印“----[task_id] is done”
from celery import Celery
from kombu import Queue
import time
app = Celery("tasks", backend="redis://127.0.0.1:6379/6")
app.config_from_object("celeryconfig")
class CallbackTask(Task):
def on_success(self, retval, task_id, args, kwargs):
print "----%s is done" % task_id
def on_failure(self, exc, task_id, args, kwargs, einfo):
pass
@app.task(base=CallbackTask)
def add(x, y):
return x + y
@app.task(base=CallbackTask)
def multiply(x,y):
return x * y
celeryconfig.py
from kombu import Queue
from kombu import Exchange
result_serializer = "json"
broker_url = "amqp://guest:guest@192.168.xx.xxx:5672/%2f"
task_queues = (
Queue("priority_low", exchange=Exchange("priority", type="direct"), routing_key="priority_low"),
Queue("priority_high", exchange=Exchange("priority", type="direct"), routing_key="priority_high"),
)
task_routes = ([
("tasks.add", {"queue": "priority_low"}),
("tasks.multiply", {"queue": "priority_high"}),
],)
task_annotations = {
"tasks.add": {"rate_limit": "10/m"}
}
Celery Server and Client
Worker on Server1
消費(fèi)priority_high事件
celery -A tasks worker -Q priority_high --concurrency=4 -l info -E -n worker1@%hWorker on Server2
消費(fèi)priority_high和priority_low事件
celery -A tasks worker -Q priority_high,priority_low --concurrency=4 -l info -E -n worker2@%hClient
生產(chǎn)者，pushlish 事件到broker
from tasks import add
from tasks import multiply
for i in xrange(50):
add.delay(2, 2)
multiply.delay(10,10)
監(jiān)控
install
pip install flower啟動(dòng)flower
假設(shè)在server2上啟動(dòng)flower，flower默認(rèn)的端口是5555.
celery flower --broker=amqp://guest:guest@192.168.xx.xxx:5672//監(jiān)控界面
在瀏覽器上輸入 http://server2_ip:5555, 可以看到如下界面：
從queued tasks途中，可以看出 priority_high中的task先消費(fèi)完，和預(yù)期是一樣的。
文章版權(quán)歸作者所有，未經(jīng)允許請(qǐng)勿轉(zhuǎn)載,若此文章存在違規(guī)行為，您可以聯(lián)系管理員刪除。
轉(zhuǎn)載請(qǐng)注明本文地址：http://m.hztianpu.com/yun/38606.html
摘要：結(jié)論執(zhí)行完任務(wù)不釋放內(nèi)存與原一直沒(méi)有被銷毀有關(guān)，因此可以適當(dāng)配置小點(diǎn)，而任務(wù)并發(fā)數(shù)與配置項(xiàng)有關(guān)，每增加一個(gè)必然增加內(nèi)存消耗，同時(shí)也影響到一個(gè)何時(shí)被銷毀，因?yàn)槭蔷鶆蛘{(diào)度任務(wù)至每個(gè)，因此也不宜配置過(guò)大，適當(dāng)配置。 1.實(shí)際使用 ? 監(jiān)控task的執(zhí)行結(jié)果：任務(wù)id，結(jié)果，traceback，children，任務(wù)狀態(tài) ? 配置 backend=redis://127...
摘要：本文將介紹如何使用和抓取主流的技術(shù)博客文章，然后用搭建一個(gè)小型的技術(shù)文章聚合平臺(tái)。是谷歌開源的基于和的自動(dòng)化測(cè)試工具，可以很方便的讓程序模擬用戶的操作，對(duì)瀏覽器進(jìn)行程序化控制。相對(duì)于，是新的開源項(xiàng)目，而且是谷歌開發(fā)，可以使用很多新的特性。 背景 說(shuō)到爬蟲，大多數(shù)程序員想到的是scrapy這樣受人歡迎的框架。scrapy的確不錯(cuò)，而且有很強(qiáng)大的生態(tài)圈，有g(shù)erapy等優(yōu)秀的可視化界面。但...
摘要：本文將介紹如何使用和抓取主流的技術(shù)博客文章，然后用搭建一個(gè)小型的技術(shù)文章聚合平臺(tái)。是谷歌開源的基于和的自動(dòng)化測(cè)試工具，可以很方便的讓程序模擬用戶的操作，對(duì)瀏覽器進(jìn)行程序化控制。相對(duì)于，是新的開源項(xiàng)目，而且是谷歌開發(fā)，可以使用很多新的特性。 背景 說(shuō)到爬蟲，大多數(shù)程序員想到的是scrapy這樣受人歡迎的框架。scrapy的確不錯(cuò)，而且有很強(qiáng)大的生態(tài)圈，有g(shù)erapy等優(yōu)秀的可視化界面。但...
閱讀 1628·2021-11-24 09:39
閱讀 1921·2021-11-22 15:25
閱讀 3908·2021-11-19 09:40
閱讀 3440·2021-09-22 15:31
閱讀 1451·2021-07-29 13:49
閱讀 1388·2019-08-26 11:59
閱讀 1465·2019-08-26 11:39
閱讀 1060·2019-08-26 11:00
