背景

最近在看Python基礎(chǔ)，剛好看到生產(chǎn)者與消費(fèi)者這快內(nèi)容，視頻使用了queue模塊來處理，這里記錄一下學(xué)習(xí)的內(nèi)容

概念

生產(chǎn)者與消費(fèi)者是一個比較容易理解的概念，比如游泳池中一頭進(jìn)水一頭出水，就是很典型的例子。

視頻中的內(nèi)容

視頻中的代碼主要是下面這塊：

# ecoding=utf-8
# Author: 翁彥彬 | Sven_Weng
# Email : sven_weng@wengyb.com
# Web   : http://wybblog.applinzi.com
from threading import current_thread, Thread
import time
import random
import queue

q = queue.Queue(5)


class Productor(Thread):
    def run(self):
        name = current_thread().getName()
        nums = range(100)
        while 1:
            nowput = random.choice(nums)
            if q.full():  # 消息隊列滿則停止生產(chǎn)
                print "隊列已經(jīng)達(dá)到上限{0}".format(q.qsize())
                time.sleep(10)
            q.put(nowput)
            print "生產(chǎn)者{0}生產(chǎn)了{(lán)1}".format(name, nowput)
            sl = random.choice([1, 2, 3])
            time.sleep(sl)  
            print "生產(chǎn)者休息了{(lán)0}秒".format(sl)


class Consumer(Thread):
    def run(self):
        name = current_thread().getName()
        while 1:
            if q.empty():  # 消息隊列空的時候則暫停消費(fèi)
                print "隊列空了，暫停消費(fèi)"
                time.sleep(5)
            num = q.get()
            q.task_done()
            print "消費(fèi)者{0}消費(fèi)了{(lán)1}".format(name, num)
            sl = random.choice([1, 2, 3])
            time.sleep(sl)
            print "消費(fèi)者休息了{(lán)0}秒".format(sl)


if __name__ == '__main__':
    p1 = Productor()
    p1.start()
    p2 = Productor()
    p2.start()
    c1 = Consumer()
    c1.start()
    c2 = Consumer()
    c2.start()
    c3 = Consumer()
    c3.start()

用的是threading模塊來一邊生產(chǎn)內(nèi)容一邊消費(fèi)內(nèi)容，整個代碼是比較簡單，但是不是特別容易理解，尤其是新手理解的時候不夠直觀。

個人理解

消息隊列這個東西可以理解為一個復(fù)雜的list，比如要實(shí)現(xiàn)先進(jìn)先出，那么每次返回list[0]就行了，同理，如果要實(shí)現(xiàn)后進(jìn)先出，那么每次返回list[len(list)]就行了，這么理解起來比較容易。
當(dāng)然，消息隊列相比起list來是復(fù)雜了一些，但是原理基本上就是這樣，比如queue使用的是threading模塊來實(shí)現(xiàn)的，再復(fù)雜的消息隊列，比如Python中用的比較多的celery，可選的消息隊列就有RabbitMQ或者Redis

一個直觀的例子

鑒于視頻中的例子過于復(fù)雜，我自己寫了一個簡單直觀的例子，用Flask寫了一個簡單的Web化的消息隊列服務(wù)。

from flask import Flask, jsonify, request
from queue import Queue

app = Flask(__name__)


@app.before_first_request
def init_queue():
    app.q = Queue(5)


@app.route('/')
def hello_world():
    data = {
        "code": "0000",
        "queue_count": app.q.qsize()
    }
    return jsonify(data)


@app.route('/put')
def put_to_queue():
    num = request.args['num']
    print num
    if app.q.full():
        data = {
            "code": "0001",
            "msg": "The Queue is full"
        }
    else:
        app.q.put(num)
        data = {
            "code": "0000",
            "msg": "Success put {1} to the Queue, current count is {0}".format(app.q.qsize(), num)
        }
    return jsonify(data)


@app.route('/get')
def get_from_queue():
    if app.q.empty():
        data = {
            "code": "0002",
            "msg": "The Queue is empty"
        }
    else:
        data = {
            "code": "0000",
            "msg": "Success get from the Queue, current count is {0}".format(app.q.qsize()),
            "num": app.q.get()
        }
        app.q.task_done()
    return jsonify(data)


if __name__ == '__main__':
    app.run(debug=True)

首先在啟動的時候調(diào)用before_first_request來初始化隊列，放到app.q這個全局變量中，用不同的請求來執(zhí)行不同的隊列操作，在頁面上就可以看到不同的結(jié)果了。演示例子中隊列數(shù)量5個就滿了，再執(zhí)行put就會返回錯誤信息，同理，如果空了，也會返回錯誤信息。