在 上一篇文章 中，我們對(duì) Celery 有了初步的認(rèn)識(shí)魔招，接下來(lái)就通過(guò)兩個(gè)例子來(lái)進(jìn)行實(shí)戰(zhàn)。最終的效果圖如下：
項(xiàng)目 github 地址：https://github.com/happy-python/flask_celery
使用環(huán)境及版本：
- python 2.7.10
- celery==4.0.0
- Flask==0.11.1
- Flask-Mail==0.9.1
- flower==0.9.1
- redis==2.10.5
簡(jiǎn)單例子：異步發(fā)送郵件
此示例是應(yīng)用程序非常普通的需求：能夠發(fā)送郵件但是不阻塞主應(yīng)用缸托。
本例使用了 Flask-Mail 擴(kuò)展饭聚。
- Flask-Mail==0.9.1
此示例應(yīng)用是一個(gè)只有一個(gè)輸入文本框的簡(jiǎn)單表單。要求用戶在此文本框中輸入一個(gè)電子郵件地址止潘，點(diǎn)擊提交掺炭，服務(wù)器會(huì)發(fā)送一個(gè)測(cè)試電子郵件到這個(gè)郵件地址。表單中包含兩個(gè)提交按鈕凭戴，一個(gè)立即發(fā)送郵件涧狮，一個(gè)是一分鐘后發(fā)送郵件。
<h2>異步發(fā)送郵件</h2>
{% for message in get_flashed_messages() %}
<p style="color: red;">{{ message }}</p>
{% endfor %}
<form method="POST" action="{{ url_for('index') }}">
<p>Send test email to: <input type="text" name="email" value="{{ email }}"></p>
<input type="submit" name="submit" value="Send">
<input type="submit" name="submit" value="Send in 1 minute">
</form>
這里沒(méi)有什么特別的東西么夫。只是一個(gè)普通的 HTML 表單者冤，再加上 Flask 閃現(xiàn)消息。
Flask-Mail 擴(kuò)展配置
# Flask-Mail 配置
app.config['MAIL_SERVER'] = 'smtp.163.com' # 電子郵件服務(wù)器的主機(jī)名或IP地址
app.config['MAIL_PORT'] = 25 # 電子郵件服務(wù)器的端口
app.config['MAIL_USE_TLS'] = True # 啟用傳輸層安全協(xié)議
app.config['MAIL_USE_SSL'] = False # 啟用安全套接層協(xié)議
app.config['MAIL_USERNAME'] = 'your-mail-username' # 郵件賬戶用戶名
app.config['MAIL_PASSWORD'] = 'your-mail-password' # 郵件賬戶的密碼
后臺(tái)路由處理
@app.route('/', methods=['GET', 'POST'])
def index():
if request.method == 'GET':
return render_template('index.html', email=session.get('email', ''))
email = request.form['email']
session['email'] = email
msg = Message('Hello from Flask', sender=app.config['MAIL_USERNAME'], recipients=[email])
msg.body = 'This is a test email sent from a background Celery task.'
if request.form['submit'] == 'Send':
# 立即發(fā)送
# delay 是 apply_async 的快捷快捷方式
# 相比于 delay档痪，當(dāng)使用 apply_async 時(shí)涉枫，我們能夠?qū)笈_(tái)任務(wù)的執(zhí)行方式有更多的控制。例如任務(wù)在何時(shí)執(zhí)行
# delay 和 apply_async 的返回值是一個(gè) AsyncResult 的對(duì)象腐螟。通過(guò)該對(duì)象愿汰，能夠獲得任務(wù)的狀態(tài)信息
async_send_email.delay(msg)
flash('Sending email to {0}'.format(email))
else:
# 1分鐘后發(fā)送
async_send_email.apply_async(args=[msg], countdown=60)
flash('An email will be sent to {0} in one minute'.format(email))
return redirect(url_for('index'))
在這里將用戶在文本框中輸入的值保存在 session 中，以便頁(yè)面重新加載時(shí)能夠使用該信息乐纸。
后臺(tái)異步任務(wù)
@celery.task
def async_send_email(msg):
app = create_app()
with app.app_context():
Mail(app).send(msg)
任何需要作為后臺(tái)任務(wù)的函數(shù)都需要使用 celery.task 裝飾器裝飾衬廷。
需要注意的是 Flask-Mail 需要在應(yīng)用的上下文中運(yùn)行，因此在調(diào)用 send() 之前需要?jiǎng)?chuàng)建一個(gè)應(yīng)用上下文汽绢。
重點(diǎn)注意此異步調(diào)用返回值并不保留吗跋，因此應(yīng)用本身無(wú)法知道是否調(diào)用成功或者失敗。運(yùn)行這個(gè)示例的時(shí)候宁昭，需要檢查 Celery worker 的輸出來(lái)排查發(fā)送郵件過(guò)程是否有問(wèn)題跌宛。
復(fù)雜例子：顯示進(jìn)度更新和結(jié)果
此示例展示一個(gè)虛構(gòu)的長(zhǎng)時(shí)間運(yùn)行的任務(wù)。用戶點(diǎn)擊按鈕啟動(dòng)一個(gè)或者更多的長(zhǎng)時(shí)間運(yùn)行的任務(wù)久窟，在瀏覽器上的頁(yè)面使用 ajax 輪詢服務(wù)器更新所有任務(wù)的狀態(tài)秩冈。每一個(gè)任務(wù)，頁(yè)面都會(huì)顯示一個(gè)進(jìn)度條斥扛，一個(gè)當(dāng)前進(jìn)度信息和一個(gè)當(dāng)前執(zhí)行結(jié)果入问。
后臺(tái)異步任務(wù)
@celery.task(bind=True)
def long_task(self):
total = random.randint(10, 50)
for i in range(total):
# 自定義狀態(tài) state
self.update_state(state=u'處理中', meta={'current': i, 'total': total})
time.sleep(1)
return {'current': 100, 'total': 100, 'result': u'完成'}
對(duì)于這個(gè)任務(wù)（此任務(wù)在一個(gè) Celery worker 進(jìn)程中運(yùn)行）丹锹，在 Celery 裝飾器中添加了 bind=True 參數(shù)，這使得 Celery 向函數(shù)中傳入了 self 參數(shù)芬失，因此在函數(shù)中能夠使用它(self)來(lái)記錄狀態(tài)更新楣黍。
self.update_state() 指明 Celery 如何接收任務(wù)更新。有一些內(nèi)置的狀態(tài)棱烂，比如 STARTED租漂，SUCCESS 等等，但是 Celery 也支持自定義狀態(tài)颊糜。在這里使用了一個(gè)叫做 '處理中' 的自定義狀態(tài)哩治。連同狀態(tài)，還有一個(gè)元數(shù)據(jù)衬鱼，該元數(shù)據(jù)是 Python 字典形式业筏，包含目前和總的迭代數(shù)∧窈眨客戶端可以使用這些元素來(lái)顯示一個(gè)漂亮的進(jìn)度條蒜胖。每迭代一次休眠一秒，以模擬正在做一些工作抛蚤。當(dāng)循環(huán)退出台谢，一個(gè) Python 字典作為函數(shù)結(jié)果返回。
啟動(dòng)后臺(tái)任務(wù)
@app.route('/longtask')
def longtask():
# 開(kāi)啟異步任務(wù)
task = long_task.apply_async()
return jsonify({}), 202, {'Location': url_for('taskstatus', task_id=task.id)}
客戶端需要發(fā)起一個(gè) GET 請(qǐng)求到 /longtask 來(lái)觸發(fā)后臺(tái)任務(wù)執(zhí)行岁经。對(duì)于響應(yīng)狀態(tài)碼 202朋沮，這個(gè)狀態(tài)碼通常是在 REST API 中使用用來(lái)表明一個(gè)請(qǐng)求正在處理中。同時(shí)添加了 Location 頭蒿偎，值為一個(gè)客戶端用來(lái)獲取狀態(tài)信息的 URL朽们。這個(gè) URL 指向另一個(gè)叫做 taskprogress 的 Flask 路由，并且該 URL 包含 task.id诉位。
獲取任務(wù)狀態(tài)信息的路由
@app.route('/status/<task_id>')
def taskstatus(task_id):
# 獲取異步任務(wù)結(jié)果
task = long_task.AsyncResult(task_id)
# 等待處理
if task.state == 'PENDING':
response = {'state': task.state, 'current': 0, 'total': 1}
elif task.state != 'FAILURE':
response = {'state': task.state, 'current': task.info.get('current', 0), 'total': task.info.get('total', 1)}
# 處理完成
if 'result' in task.info:
response['result'] = task.info['result']
else:
# 后臺(tái)任務(wù)出錯(cuò)
response = {'state': task.state, 'current': 1, 'total': 1}
return jsonify(response)
這個(gè)路由生成一個(gè) JSON 響應(yīng)，該響應(yīng)包含任務(wù)的狀態(tài)以及在 update_state() 調(diào)用中設(shè)置的 meta 參數(shù)的所有值菜枷。
第一個(gè) if 代碼塊是當(dāng)任務(wù)還沒(méi)有開(kāi)始的時(shí)候(PENDING 狀態(tài))苍糠。在這種情況下暫時(shí)沒(méi)有狀態(tài)信息，因此人為地制造了些數(shù)據(jù)啤誊。接下來(lái)的 elif 代碼塊返回后臺(tái)任務(wù)的狀態(tài)信息岳瞭。任務(wù)提供的信息可以通過(guò)訪問(wèn) task.info 獲得。如果數(shù)據(jù)中包含 result 蚊锹，這就意味著這是最終的結(jié)果并且任務(wù)已經(jīng)結(jié)束瞳筏，因此我把這些信息也加到響應(yīng)中。最后的 else 代碼塊是任務(wù)執(zhí)行失敗的情況牡昆，在這種情況下 task.info 中會(huì)包含異常的信息姚炕。
客戶端
<h2>顯示進(jìn)度更新和結(jié)果</h2>
<button id="start-bg-job">Start Long Task</button>
<br><br>
<div id="progress" style="width: auto;text-align: center;"></div>
對(duì)于圖形進(jìn)度條使用 nanobar.js。同樣還需要引入 jQuery，它能夠簡(jiǎn)化 ajax 的調(diào)用柱宦。
<script src="http://cdn.bootcss.com/nanobar/0.4.2/nanobar.min.js"></script>
<script src="http://cdn.bootcss.com/jquery/3.1.1/jquery.min.js"></script>
啟動(dòng)后臺(tái)任務(wù)：
$('button').on('click', start_long_task);
function start_long_task() {
// 添加任務(wù)狀態(tài)元素
var div = $('<div class="progress"><div></div><div>0%</div><div></div></div><hr>');
$('#progress').append(div);
// 創(chuàng)建進(jìn)度條（progress bar）
var nanobar = new Nanobar({
bg: '#44f',
target: div[0].childNodes[0]
});
// 向后臺(tái)發(fā)送請(qǐng)求開(kāi)啟任務(wù)
var longTask = $.get('/longtask');
longTask.done(function (data, status, request) {
status_url = request.getResponseHeader('Location');
update_progress(status_url, nanobar, div[0]);
});
}
div 代碼解釋：
<div class="progress">
<div></div> <-- 進(jìn)度條
<div>0%</div> <-- 當(dāng)前進(jìn)度
<div> </div> <-- 當(dāng)前結(jié)果
</div>
<hr>
更新進(jìn)度信息：
function update_progress(status_url, nanobar, status_div) {
// 獲取狀態(tài)信息
$.getJSON(status_url, function (data) {
// 更新進(jìn)度
percent = parseInt(data['current'] * 100 / data['total']);
nanobar.go(percent);
$(status_div.childNodes[1]).text('當(dāng)前進(jìn)度: ' + percent + '%');
// 輪詢
if (data['state'] == 'PENDING' || data['state'] == '處理中') {
setTimeout(function () {
update_progress(status_url, nanobar, status_div);
}, 2000);
}
// 更新結(jié)果
if ('result' in data) {
// 處理完成
$(status_div.childNodes[2]).text('當(dāng)前結(jié)果: ' + data['result']);
}
else {
// 處理中
$(status_div.childNodes[2]).text('當(dāng)前結(jié)果: ' + data['state']);
}
});
}
當(dāng)后臺(tái)任務(wù)運(yùn)行時(shí)些椒，為了能夠持續(xù)獲得任務(wù)狀態(tài)并更新頁(yè)面，這里使用了定時(shí)器掸刊，每隔兩秒更新一次直到后臺(tái)任務(wù)完成免糕。
運(yùn)行
啟動(dòng) celery worker：
$ celery -A app worker -l info
-------------- celery@ubuntu v4.0.0 (latentcall)
---- **** -----
--- * *** * -- Linux-3.16.0-57-generic-i686-with-Ubuntu-14.04-trusty 2016-11-26 22:05:09
-- * - **** ---
- ** ---------- [config]
- ** ---------- .> app: app:0xb6472acc
- ** ---------- .> transport: redis://localhost:6379/15
- ** ---------- .> results: redis://localhost:6379/15
- *** --- * --- .> concurrency: 4 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** -----
-------------- [queues]
.> celery exchange=celery(direct) key=celery
[tasks]
. app.tasks.async_send_email
. app.tasks.long_task
[2016-11-26 22:05:10,177: INFO/MainProcess] Connected to redis://localhost:6379/15
[2016-11-26 22:05:10,192: INFO/MainProcess] mingle: searching for neighbors
[2016-11-26 22:05:11,231: INFO/MainProcess] mingle: all alone
[2016-11-26 22:05:11,244: INFO/MainProcess] celery@ubuntu ready.
[2016-11-26 22:05:31,132: INFO/MainProcess] Events of group {task} enabled by remote.
[2016-11-26 22:06:33,626: INFO/MainProcess] Received task: app.tasks.long_task[9eeed278-0e14-48ae-9762-4de447ca79ff]
[2016-11-26 22:06:35,137: INFO/MainProcess] Received task: app.tasks.long_task[8a0b46d1-abba-4bfa-844a-06ecf6f8b749]
[2016-11-26 22:07:05,259: INFO/PoolWorker-1] Task app.tasks.long_task[8a0b46d1-abba-4bfa-844a-06ecf6f8b749] succeeded in 30.118909819s: {'current': 100, 'total': 100, 'result': '完成'}
[2016-11-26 22:07:10,752: INFO/PoolWorker-3] Task app.tasks.long_task[9eeed278-0e14-48ae-9762-4de447ca79ff] succeeded in 37.123306606s: {'current': 100, 'total': 100, 'result': '完成'}
啟動(dòng) flower 進(jìn)程：
$ flower -A app
[I 161126 22:05:26 command:136] Visit me at http://localhost:5555
[I 161126 22:05:26 command:141] Broker: redis://localhost:6379/15
[I 161126 22:05:26 command:144] Registered tasks:
[u'app.tasks.async_send_email',
u'app.tasks.long_task',
u'celery.accumulate',
u'celery.backend_cleanup',
u'celery.chain',
u'celery.chord',
u'celery.chord_unlock',
u'celery.chunks',
u'celery.group',
u'celery.map',
u'celery.starmap']
[I 161126 22:05:26 mixins:224] Connected to redis://localhost:6379/15
[W 161126 22:05:28 control:44] 'active' inspect method failed
[W 161126 22:05:28 control:44] 'reserved' inspect method failed
[W 161126 22:05:28 control:44] 'conf' inspect method failed
