任務(wù)異步化
打開瀏覽器阶界，輸入地址，按下回車坛梁，打開了頁面而姐。于是一個HTTP請求（request）就由客戶端發(fā)送到服務(wù)器，服務(wù)器處理請求划咐，返回響應(yīng)（response）內(nèi)容拴念。
我們每天都在瀏覽網(wǎng)頁，發(fā)送大大小小的請求給服務(wù)器褐缠。有時候政鼠，服務(wù)器接到了請求，會發(fā)現(xiàn)他也需要給另外的服務(wù)器發(fā)送請求送丰，或者服務(wù)器也需要做另外一些事情缔俄，于是最初們發(fā)送的請求就被阻塞了弛秋，也就是要等待服務(wù)器完成其他的事情器躏。
更多的時候，服務(wù)器做的額外事情蟹略，并不需要客戶端等待登失，這時候就可以把這些額外的事情異步去做。從事異步任務(wù)的工具有很多挖炬。主要原理還是處理通知消息揽浙，針對通知消息通常采取是隊列結(jié)構(gòu)。生產(chǎn)和消費消息進(jìn)行通信和業(yè)務(wù)實現(xiàn)。
生產(chǎn)消費與隊列
上述異步任務(wù)的實現(xiàn)馅巷，可以抽象為生產(chǎn)者消費模型膛虫。如同一個餐館，廚師在做飯钓猬，吃貨在吃飯稍刀。如果廚師做了很多，暫時賣不完敞曹，廚師就會休息账月；如果客戶很多，廚師馬不停蹄的忙碌澳迫，客戶則需要慢慢等待局齿。實現(xiàn)生產(chǎn)者和消費者的方式用很多，下面使用Python標(biāo)準(zhǔn)庫Queue寫個小例子：
import random
import time
from Queue import Queue
from threading import Thread
queue = Queue(10)
class Producer(Thread):
def run(self):
while True:
elem = random.randrange(9)
queue.put(elem)
print "廚師 {} 做了 {} 飯 --- 還剩 {} 飯沒賣完".format(self.name, elem, queue.qsize())
time.sleep(random.random())
class Consumer(Thread):
def run(self):
while True:
elem = queue.get()
print "吃貨{} 吃了 {} 飯 --- 還有 {} 飯可以吃".format(self.name, elem, queue.qsize())
time.sleep(random.random())
def main():
for i in range(3):
p = Producer()
p.start()
for i in range(2):
c = Consumer()
c.start()
if __name__ == '__main__':
main()
大概輸出如下：
廚師 Thread-1 做了 1 飯 --- 還剩 1 飯沒賣完
廚師 Thread-2 做了 8 飯 --- 還剩 2 飯沒賣完
廚師 Thread-3 做了 3 飯 --- 還剩 3 飯沒賣完
吃貨Thread-4 吃了 1 飯 --- 還有 2 飯可以吃
吃貨Thread-5 吃了 8 飯 --- 還有 1 飯可以吃
吃貨Thread-4 吃了 3 飯 --- 還有 0 飯可以吃
廚師 Thread-1 做了 0 飯 --- 還剩 1 飯沒賣完
廚師 Thread-2 做了 0 飯 --- 還剩 2 飯沒賣完
廚師 Thread-1 做了 1 飯 --- 還剩 3 飯沒賣完
廚師 Thread-1 做了 1 飯 --- 還剩 4 飯沒賣完
吃貨Thread-4 吃了 0 飯 --- 還有 3 飯可以吃
廚師 Thread-3 做了 3 飯 --- 還剩 4 飯沒賣完
吃貨Thread-5 吃了 0 飯 --- 還有 3 飯可以吃
吃貨Thread-5 吃了 1 飯 --- 還有 2 飯可以吃
廚師 Thread-2 做了 8 飯 --- 還剩 3 飯沒賣完
廚師 Thread-2 做了 8 飯 --- 還剩 4 飯沒賣完
Redis 隊列
Python內(nèi)置了一個好用的隊列結(jié)構(gòu)橄登。我們也可以是用redis實現(xiàn)類似的操作抓歼。并做一個簡單的異步任務(wù)。
Redis提供了兩種方式來作消息隊列示绊。一個是使用生產(chǎn)者消費模式模式锭部，另外一個方法就是發(fā)布訂閱者模式。前者會讓一個或者多個客戶端監(jiān)聽消息隊列面褐，一旦消息到達(dá)拌禾，消費者馬上消費，誰先搶到算誰的展哭，如果隊列里沒有消息湃窍，則消費者繼續(xù)監(jiān)聽。后者也是一個或多個客戶端訂閱消息頻道匪傍，只要發(fā)布者發(fā)布消息您市，所有訂閱者都能收到消息，訂閱者都是平等的役衡。
生產(chǎn)消費模式（不建議使用）
主要使用了redis提供的blpop獲取隊列數(shù)據(jù)茵休，如果隊列沒有數(shù)據(jù)則阻塞等待，也就是監(jiān)聽手蝎。
import redis
class Task(object):
def __init__(self):
self.rcon = redis.StrictRedis(host='localhost', db=5)
self.queue = 'task:prodcons:queue'
def listen_task(self):
while True:
task = self.rcon.blpop(self.queue, 0)[1]
print "Task get", task
if __name__ == '__main__':
print 'listen task queue'
Task().listen_task()
使用redis的brpop方式做隊列榕莺，經(jīng)過一段時間，會發(fā)現(xiàn)程序莫名其妙的卡主棵介。也就是進(jìn)程一切ok钉鸯，redis的lpush也正常，唯獨brpop不再消費邮辽。該問題十分不好復(fù)現(xiàn)唠雕，但是總是過了一段時間就會重現(xiàn)贸营。本人采用了高并發(fā)，高延遲岩睁，弱網(wǎng)絡(luò)環(huán)境等方式試圖復(fù)現(xiàn)都沒有成功钞脂，目前仍然在尋找解決方案。目測依賴redis做brokers的隊列的celery也遇到同樣的問題 捕儒，并且其他語言也有類似問題芳肌，但是作者的解決方不適用。猜測問題的原因是redis在處理brpop的時候連接長時間不適用會自動假死肋层。后來采用比較low的方案亿笤，每當(dāng)凌晨3點左右重啟一下隊列服務(wù)。目前設(shè)置了更短的idle連接時間（config set timeout 10）栋猖，再觀察一下是否能復(fù)現(xiàn)垒拢。
不建議在生成環(huán)境使用該方案喻括。如果使用類似方案也遇到了問題赦拘，并且有了解決方案忽冻，希望您能聯(lián)系我哈哈哈。
升級了 redis 3.2 版本之后雌团，運行了一個多月燃领，目前沒有再出現(xiàn)卡住的問題。
發(fā)布訂閱模式
使用redis的pubsub功能锦援，訂閱者訂閱頻道猛蔽，發(fā)布者發(fā)布消息到頻道了，頻道就是一個消息隊列灵寺。
import redis
class Task(object):
def __init__(self):
self.rcon = redis.StrictRedis(host='localhost', db=5)
self.ps = self.rcon.pubsub()
self.ps.subscribe('task:pubsub:channel')
def listen_task(self):
for i in self.ps.listen():
if i['type'] == 'message':
print "Task get", i['data']
if __name__ == '__main__':
print 'listen task channel'
Task().listen_task()
Flask 入口
我們分別實現(xiàn)了兩種異步任務(wù)的后端服務(wù)曼库，直接啟動他們，就能監(jiān)聽redis隊列或頻道的消息了略板。簡單的測試如下：
import redis
import random
import logging
from flask import Flask, redirect
app = Flask(__name__)
rcon = redis.StrictRedis(host='localhost', db=5)
prodcons_queue = 'task:prodcons:queue'
pubsub_channel = 'task:pubsub:channel'
@app.route('/')
def index():
html = """
<br>
<center><h3>Redis Message Queue</h3>
<br>
<a href="/prodcons">生產(chǎn)消費者模式</a>
<br>
<br>
<a href="/pubsub">發(fā)布訂閱者模式</a>
</center>
"""
return html
@app.route('/prodcons')
def prodcons():
elem = random.randrange(10)
rcon.lpush(prodcons_queue, elem)
logging.info("lpush {} -- {}".format(prodcons_queue, elem))
return redirect('/')
@app.route('/pubsub')
def pubsub():
ps = rcon.pubsub()
ps.subscribe(pubsub_channel)
elem = random.randrange(10)
rcon.publish(pubsub_channel, elem)
return redirect('/')
if __name__ == '__main__':
app.run(debug=True)
啟動腳本毁枯，使用
siege -c10 -r 5 http://127.0.0.1:5000/prodcons
siege -c10 -r 5 http://127.0.0.1:5000/pubsub
可以分別在監(jiān)聽的腳本輸入中看到異步消息。在異步的任務(wù)中叮称，可以執(zhí)行一些耗時間的操作种玛，當(dāng)然目前這些做法并不知道異步的執(zhí)行結(jié)果，如果需要知道異步的執(zhí)行結(jié)果瓤檐，可以考慮設(shè)計協(xié)程任務(wù)或者使用一些工具如RQ或者celery等赂韵。
