摘要：因?yàn)樗蔷€程安全的，所以多個(gè)線程很輕松地使用同一個(gè)實(shí)例。后進(jìn)先出隊(duì)列使用后進(jìn)先出順序，與棧結(jié)構(gòu)相似這就是全部代碼了，這正是設(shè)計(jì)很棒的一個(gè)原因，它將底層的數(shù)據(jù)操作抽象成四個(gè)操作函數(shù)，本身來(lái)處理線程安全的問(wèn)題，使得其子類只需關(guān)注底層的操作。

queue 模塊提供適用于多線程編程的先進(jìn)先出（FIFO）數(shù)據(jù)結(jié)構(gòu)。因?yàn)樗蔷€程安全的，所以多個(gè)線程很輕松地使用同一個(gè)實(shí)例。

先從初始化的函數(shù)來(lái)看：

class Queue:
    def __init__(self, maxsize=0):
        # 設(shè)置隊(duì)列的最大容量
        self.maxsize = maxsize
        self._init(maxsize)

        # 線程鎖，互斥變量
        self.mutex = threading.Lock()
        # 由鎖衍生出三個(gè)條件變量
        self.not_empty = threading.Condition(self.mutex)
        self.not_full = threading.Condition(self.mutex)
        self.all_tasks_done = threading.Condition(self.mutex)

        self.unfinished_tasks = 0

    def _init(self, maxsize):
        # 初始化底層數(shù)據(jù)結(jié)構(gòu)
        self.queue = deque()

從這初始化函數(shù)能得到哪些信息呢？首先，隊(duì)列是可以設(shè)置其容量大小的，并且具體的底層存放元素的它使用了 collections.deque() 雙端列表的數(shù)據(jù)結(jié)構(gòu)，這使得能很方便的做先進(jìn)先出操作。這里還特地抽象為 _init 函數(shù)是為了方便其子類進(jìn)行覆蓋，允許子類使用其他結(jié)構(gòu)來(lái)存放元素（比如優(yōu)先隊(duì)列使用了 list）。

然后就是線程鎖 self.mutex ，對(duì)于底層數(shù)據(jù)結(jié)構(gòu) self.queue 的操作都要先獲得這把鎖；再往下是三個(gè)條件變量，這三個(gè) Condition 都以 self.mutex 作為參數(shù)，也就是說(shuō)它們共用一把鎖；從這可以知道諸如 with self.mutex 與 with self.not_empty 等都是互斥的。

基于這些鎖而做的一些簡(jiǎn)單的操作：

class Queue:
    ...
    def qsize(self):
        # 返回隊(duì)列中的元素?cái)?shù)
        with self.mutex:
            return self._qsize()

    def empty(self):
        # 隊(duì)列是否為空
        with self.mutex:
            return not self._qsize()

    def full(self):
        # 隊(duì)列是否已滿
        with self.mutex:
            return 0 < self.maxsize <= self._qsize()

    def _qsize(self):
        return len(self.queue)

這個(gè)代碼片段挺好理解的，無(wú)需分析。

作為隊(duì)列，主要得完成入隊(duì)與出隊(duì)的操作，首先是入隊(duì)：

class Queue:
    ...
    def put(self, item, block=True, timeout=None):
        with self.not_full: # 獲取條件變量not_full
            if self.maxsize > 0:
                if not block:
                    if self._qsize() >= self.maxsize:
                        raise Full # 如果 block 是 False，并且隊(duì)列已滿，那么拋出 Full 異常
                elif timeout is None:
                    while self._qsize() >= self.maxsize:
                        self.not_full.wait() # 阻塞直到由剩余空間
                elif timeout < 0: # 不合格的參數(shù)值，拋出ValueError
                    raise ValueError(""timeout" must be a non-negative number")
                else:
                    endtime = time() + timeout  # 計(jì)算等待的結(jié)束時(shí)間
                    while self._qsize() >= self.maxsize:
                        remaining = endtime - time()
                        if remaining <= 0.0:
                            raise Full # 等待期間一直沒(méi)空間，拋出 Full 異常
                        self.not_full.wait(remaining)
            self._put(item) # 往底層數(shù)據(jù)結(jié)構(gòu)中加入一個(gè)元素
            self.unfinished_tasks += 1
            self.not_empty.notify()

    def _put(self, item):
        self.queue.append(item)

盡管只有二十幾行的代碼，但這里的邏輯還是比較復(fù)雜的。它要處理超時(shí)與隊(duì)列剩余空間不足的情況，具體幾種情況如下：

如果 block 是 False，忽略timeout參數(shù)

若此時(shí)隊(duì)列已滿，則拋出 Full 異常；

若此時(shí)隊(duì)列未滿，則立即把元素保存到底層數(shù)據(jù)結(jié)構(gòu)中；

如果 block 是 True

若 timeout 是 None 時(shí)，那么put操作可能會(huì)阻塞，直到隊(duì)列中有空閑的空間（默認(rèn)）；

若 timeout 是非負(fù)數(shù)，則會(huì)阻塞相應(yīng)時(shí)間直到隊(duì)列中有剩余空間，在這個(gè)期間，如果隊(duì)列中一直沒(méi)有空間，拋出 Full 異常；

處理好參數(shù)邏輯后，，將元素保存到底層數(shù)據(jù)結(jié)構(gòu)中，并遞增unfinished_tasks，同時(shí)通知 not_empty ，喚醒在其中等待數(shù)據(jù)的線程。

出隊(duì)操作：

class Queue:
    ...
    def get(self, block=True, timeout=None):
        with self.not_empty:
            if not block:
                if not self._qsize():
                    raise Empty
            elif timeout is None:
                while not self._qsize():
                    self.not_empty.wait()
            elif timeout < 0:
                raise ValueError(""timeout" must be a non-negative number")
            else:
                endtime = time() + timeout
                while not self._qsize():
                    remaining = endtime - time()
                    if remaining <= 0.0:
                        raise Empty
                    self.not_empty.wait(remaining)
            item = self._get()
            self.not_full.notify()
            return item

    def _get(self):     
        return self.queue.popleft()

get() 操作是 put() 相反的操作，代碼塊也及其相似，get() 是從隊(duì)列中移除最先插入的元素并將其返回。

如果 block 是 False，忽略timeout參數(shù)

若此時(shí)隊(duì)列沒(méi)有元素，則拋出 Empty 異常；

若此時(shí)隊(duì)列由元素，則立即把元素保存到底層數(shù)據(jù)結(jié)構(gòu)中；

如果 block 是 True

若 timeout 是 None 時(shí)，那么get操作可能會(huì)阻塞，直到隊(duì)列中有元素（默認(rèn)）；

若 timeout 是非負(fù)數(shù)，則會(huì)阻塞相應(yīng)時(shí)間直到隊(duì)列中有元素，在這個(gè)期間，如果隊(duì)列中一直沒(méi)有元素，則拋出 Empty 異常；

最后，通過(guò) self.queue.popleft() 將最早放入隊(duì)列的元素移除，并通知 not_full ，喚醒在其中等待數(shù)據(jù)的線程。

這里有個(gè)值得注意的地方，在 put() 操作中遞增了 self.unfinished_tasks ，而 get() 中卻沒(méi)有遞減，這是為什么？

這其實(shí)是為了留給用戶一個(gè)消費(fèi)元素的時(shí)間，get() 僅僅是獲取元素，并不代表消費(fèi)者線程處理的該元素，用戶需要調(diào)用 task_done() 來(lái)通知隊(duì)列該任務(wù)處理完成了：

class Queue:
    ...
    def task_done(self):
        with self.all_tasks_done:
            unfinished = self.unfinished_tasks - 1
            if unfinished <= 0:
                if unfinished < 0: # 也就是成功調(diào)用put()的次數(shù)小于調(diào)用task_done()的次數(shù)時(shí)，會(huì)拋出異常
                    raise ValueError("task_done() called too many times")
                self.all_tasks_done.notify_all() # 當(dāng)unfinished為0時(shí)，會(huì)通知all_tasks_done
            self.unfinished_tasks = unfinished

    def join(self):
        with self.all_tasks_done:
            while self.unfinished_tasks: # 如果有未完成的任務(wù)，將調(diào)用wait()方法等待
                self.all_tasks_done.wait()

由于 task_done() 使用方調(diào)用的，當(dāng) task_done() 次數(shù)大于 put() 次數(shù)時(shí)會(huì)拋出異常。

task_done() 操作的作用是喚醒正在阻塞的 join() 操作。join() 方法會(huì)一直阻塞，直到隊(duì)列中所有的元素都被取出，并被處理了（和線程的join方法類似）。也就是說(shuō) join() 方法必須配合 task_done() 來(lái)使用才行。

LifoQueue使用后進(jìn)先出順序，與棧結(jié)構(gòu)相似：

class LifoQueue(Queue):
    """Variant of Queue that retrieves most recently added entries first."""

    def _init(self, maxsize):
        self.queue = []

    def _qsize(self):
        return len(self.queue)

    def _put(self, item):
        self.queue.append(item)

    def _get(self):
        return self.queue.pop()

這就是 LifoQueue 全部代碼了，這正是 Queue 設(shè)計(jì)很棒的一個(gè)原因，它將底層的數(shù)據(jù)操作抽象成四個(gè)操作函數(shù)，本身來(lái)處理線程安全的問(wèn)題，使得其子類只需關(guān)注底層的操作。

LifoQueue 底層數(shù)據(jù)結(jié)構(gòu)改用 list 來(lái)存放，通過(guò) self.queue.pop() 就能將 list 中最后一個(gè)元素移除，無(wú)需重置索引。

from heapq import heappush, heappop

class PriorityQueue(Queue):
    """Variant of Queue that retrieves open entries in priority order (lowest first).

    Entries are typically tuples of the form:  (priority number, data).
    """

    def _init(self, maxsize):
        self.queue = []

    def _qsize(self):
        return len(self.queue)

    def _put(self, item):
        heappush(self.queue, item)

    def _get(self):
        return heappop(self.queue)

優(yōu)先隊(duì)列使用了 heapq 模塊的結(jié)構(gòu)，也就是最小堆的結(jié)構(gòu)。優(yōu)先隊(duì)列更為常用，隊(duì)列中項(xiàng)目的處理順序需要基于這些項(xiàng)目的特征，一個(gè)簡(jiǎn)單的例子：

import queue

class A:
    def __init__(self, priority, value):
        self.priority = priority
        self.value = value

    def __lt__(self, other):
        return self.priority < other.priority


q = queue.PriorityQueue()

q.put(A(1, "a"))
q.put(A(0, "b"))
q.put(A(1, "c"))

print(q.get().value)  # "b"

使用優(yōu)先隊(duì)列的時(shí)候，需要定義 __lt__ 魔術(shù)方法，來(lái)定義它們之間如何比較大小。若元素的 priority 相同，依然使用先進(jìn)先出的順序。

https://pymotw.com/3/queue/in...