文件:

筆記：

• Celery 對名字空間有較為嚴格要求，應該採用類似 Django 的以專案名開始的路徑，例如 proj.tasks.taskA
• Celery的最簡單的訊息處理方式是採用預設的celery queue，direct exchange
• Celery預設會建立一個程序池，其中worker程序數等於CPU核心數
• 用TERM訊號結束後臺執行的Celery例項，即 kill -15 pid

Example 1

結合 Supervisor 寫了個後臺自動執行worker的Demo
類似 ping 伺服器，收到請求後，睡眠一段時間，然後返回響應資訊

Step 1 建立工程目錄

# 目錄結構如下，注意 client.py 和 test1 同級
test1\__init__.py
      celery.py
      celeryconfig.py
      tasks.py
client.py   # producer

Step 2 程式碼編寫

celeryconfig.py

BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost'

celery.py
建立app例項，進行配置

from __future__ import absolute_import

from
 
 celery import Celery

app = Celery('test1',
             include=['test1.tasks'])
app.config_from_object('test1.celeryconfig')

if __name__ == '__main__':
    app.start()

tasks.py

from __future__ import absolute_import

from test1.celery import app
import time
import random


@app.task
def ping
 
(seq, start):
    delay = random.randint(0, 3)
    time.sleep(delay)
    ret = time.time()
    return '{0} - pong {1}, run {2} seconds'.format(seq, delay, ret-start)

client.py

import time

from test1 import tasks


def test():
    results = set(tasks.ping.delay(i, time.time()) for i in range(10))

    while results:
        for i in results:
            if i.ready():
                sav = i
                results.remove(i)
                print sav.get()
                break

test()

Step 3 測試執行

cd 到 test1 的父目錄，執行

celery -A test1 worker -l info
python client.py

成功執行則會看到類似以下輸出
1 - pong 0, run 0.00337100028992 seconds
2 - pong 1, run 1.00772404671 seconds
0 - pong 3, run 3.04080796242 seconds
3 - pong 3, run 3.00674390793 seconds
7 - pong 0, run 3.00365185738 seconds
5 - pong 2, run 3.00629997253 seconds
9 - pong 2, run 5.00392007828 seconds
4 - pong 3, run 6.00150322914 seconds
6 - pong 3, run 5.99989199638 seconds
8 - pong 2, run 7.99660491943 seconds

前面幾個任務的執行時間和睡眠時間一致，後面的程序執行時間更長，和 round-robin 的排程方式相符

Step 4 加入supervisor管理

如果使用虛擬環境，只需使用相應二進位制檔案的絕對路徑即可。
/etc/supervisor/test1.conf

[program:test1]
directory=/Users/apple/PycharmProjects/Fun
command=/Users/apple/.pyenv/versions/my-virtual-env-2.7.10/bin/celery -A test1 worker -l info

autostart=true
autorestart=true

startsecs=10

stdout_logfile=/Users/apple/test1.log
redirect_stderr=true

killasgroup=true

啟動 supervisor，用 supervisorctl 確認 test1正常執行。用 ps aux | grep python 應該可以看到多個worker程序。

執行 client.py 驗證。

————————————————————–

supervisorctl 中
stop + remove 可以移除一個被管理的程序
add program 加入一個曾管理過的程序
更多的指令用 ? command 即可檢視