使用celery之了解celery（轉）

原文   http://www.dongwm.com/archives/shi-yong-celeryzhi-liao-jie-celery/

 

前言

我想很多做開發和運維的都會涉及一件事:crontab, 也就是在服務器上設定定時任務,按期執行一些任務.但是假如你有上千台的服務器， 你有上千種任務,那么對於這個定時任務的管理恐怕是一件很頭疼的事情.哪怕你只是幾十個任務分配的不同的機器上怎么樣合理的管理和實現以下功能呢:

1. 查看定時任務的執行情況.比如執行是否成功，當前狀態，執行花費的時間.
2. 一個友好的界面或者命令行下實現添加,刪除任務
3. 怎么樣簡單實現不同的機器設定不同種任務,某些機器執行不同的隊列
4. 假如你需要生成一個任務怎么樣不阻塞剩下來的過程(異步了唄)
5. 怎么樣並發的執行任務

幾種選擇

1. 有錢有人有時間自己實現一套,優點是完全符合公司業務的需要,有專門的團隊維護和服務
2. 使用 Gearman ,聽說過沒用過，因為是C/java/perl,對我們這種python開發者或者運維來說,假如沒有這方面經驗之后沒有能力了解底層實現和二次開發的能力
3. 使用 rq , rq是搞gitflow的那個作者寫的，簡介里面說的很清楚:Simple job queues for Python. 怕它不夠復雜,但是假如業務沒有那么復雜或者應用不是那么嚴格，完全可以嘗試下
4. 好吧我選擇了 celery , 現在用了快半年，可能是歷史遺留問題,版本較低.有很多坑.但是很不錯

消息隊列

RabbitMQ,ZeroMQ這樣的消息隊列總是出現在我們視線中, 其實意義是很簡單: 消息就是一個要傳送的數據,celery是一個分布式的任務隊列.這個”任務”其實就是一種消息, 任務被生成到隊列中，被RabbitMQ等容器接收和存儲，在適當的時候又被要執行的機器把這個消息取走.

celery任務可以使用RabbitMQ/Redis/SQLAlchemy/Django的orm/Mongodb等等容器(這里叫做Broker).我使用的是RabbitMQ,因為作者在github主頁的介紹里面很明確的寫了這個

所謂隊列,你可以設想一個問題,我有一大推的東西要執行,但是我並不是需要每個服務器都執行這個任務,因為業務不同嘛. 所以就要做個隊列, 比如任務A,B,C A可以在X,Y服務器執行, 但是不需要或者不能在Z服務器上執行.那么在X,Y你啟動worker(下面會說，其實就是消費者和生產者的消費者)加上這個隊列,Z服務器就不需要指定這個隊列,也就不會執行這個隊列的任務

celery的原理,我這里的角度是django+celery+django-celery

首先說一下流程:

1. 使用django-celery或者直接操作數據庫(settings.py里面指定)添加任務,設置的相關屬性(包含定時任務的間隔)存入數據庫.
2. celerybeat通過djcelery.schedulers.DatabaseScheduler獲取django內你設置的任務周期性的檢查(默認5s),發現需要執行某任務講其丟入你設置的broker(我這里是rabbitmq),他會更具settings.py的設置放到對應的隊列
3. 在你啟動了celery worker(以前是celeryd)的服務器上,根據worker也會定期(默認5s)去broker里面查找需要它執行的隊列里面是否有任務
4. 當發現隊列有要執行的任務,worker將它取出來執行,執行完的結果通過celerycam(默認30s,所以這個進程也要啟動)寫入django設置的數據庫,更新了這個任務的狀態.比如花費的時間

supervisor進程管理

不知道有沒有人用過 supervise ，我以前經常在最初的項目開發中經常使用它監視我的程序,當程序死掉自動啟動, supervisor 確是一個 進程管理的工具，我在這里使用它管理celery的程序和uwsgi

粘貼下我的一個本地環境的配置,並直接進行一下說明:

;程序名字
[program:celery-queue-fetch]
;程序要執行的命令, -Q 指定了生成和接受任務的隊列,多個用都好分開 -c為workr的數量,原意為並發數量
command=python /home/dongwm/work/manage.py celery worker -E --settings=settings_local --loglevel=INFO -Q fetch_ -c 30
;程序執行時候所在目錄
directory=/home/dongwm/work/
;執行程序使用的用戶
user=dongwm
;啟動的程序的實例數,默認是1
numprocs=1
stdout_logfile=/home/dongwm/work/celerylog/celery.log
stderr_logfile=/home/dongwm/work/celerylog/celery.log
;在啟動supervisor時候自動啟動
autostart=true
;當程序可能因為某些原因沒有啟動成功會自動重啟
autorestart=true
;啟動的等待時候,我想是為了重啟能殺掉原來進程預留的時間
startsecs=10
;進程發送停止信號等待os返回SIGCHILD的時間
stopwaitsecs=10
;低優先級的會首先啟動最后關閉
priority=998
;以下2句是為了保證殺掉進程和其子進程而不會只殺死其控制的程序主進程而留下子進程變為孤立進程的問題
stopsignal=KILL
stopasgroup=true

[program:celery-queue-feed]
command=python /home/dongwm/work/manage.py celeryd -E --settings=settings_local --loglevel=INFO -Q feed
directory=/home/dongwm/work/
user=dongwm
numprocs=1
stdout_logfile=/home/dongwm/work/celerylog/celery.log
stderr_logfile=/home/dongwm/work/celerylog/celery.log
autostart=true
autorestart=true
startsecs=10
stopwaitsecs=10
priority=998
stopsignal=KILL
stopasgroup=true

[program:celerycam]
;任務快照的間隔時間為10s
command=python /home/dongwm/work/manage.py celerycam -F 10 --settings=settings_local
directory=/home/dongwm/work/
user=dongwm
numprocs=1
stdout_logfile=/home/dongwm/work/celerylog/celerycam.log
stderr_logfile=/home/dongwm/work/celerylog/celerycam.log
autostart=true
autorestart=true
startsecs=5
stopwaitsecs=5
priority=998
stopsignal=KILL
stopasgroup=true

[program:celerybeat]
command=python /home/dongwm/work/manage.py celerybeat --settings=settings_real_old --loglevel=DEBUG
directory=/home/dongwm/work/
user=dongwm
numprocs=1
stdout_logfile=/home/dongwm/work/celerylog/celery_beat.log
stderr_logfile=/home/dongwm/work/celerylog/celery_beat.log
autostart=true
autorestart=true
startsecs=10
priority=999
stopsignal=KILL
stopasgroup=true

;這是supervisor官方的一個監控進程狀態異常退出的腳本,我對它進行了較大的修改,這樣在程序奇怪退出的時候會給我發郵件
[eventlistener:crashmail]
command=python /home/dongwm/superlance/superlance/crashmail.py -a -m [e-mail]
events=PROCESS_STATE_EXITED

[program:uwsgi]
user = dongwm
numprocs=1
command=/usr/local/bin/uwsgi -s /tmp/uwsgi-sandbox.sock --processes 4  --enable-threads \
     --pythonpath /home/dongwm/uwsgi --buffer-size 32768 --listen 100 --daemonize /home/dongwm/ulog/uwsgi_out.log
directory=/home/dongwm/work
autostart=true
autorestart=true
redirect_stderr=true
stopsignal=KILL
stopasgroup=true

 

nginx+uwsgi的實踐

nginx進程數會直接影響性能, 如何使用到的模塊不會出現阻塞式的調用，應該有多少cpu就配多少worker_processes,否則才需要配置更多的進程數. 比如你的用戶大量讀取你的本地靜態文件，並且服務器上面內存較少，硬盤的I/O調用可能會阻塞worker少量時間,那么就要多配

為了更好利用多核的優勢,我綁定了worker和對應的內核:

worker_processes     4;
worker_cpu_affinity 0001 0010 0100 1000;