技術標籤：python資料庫python

程式碼實現

看完了設計模式的原理，我們下面來試著用程式碼來實現一下，

由於這個設計模式非常簡單，這個程式碼並不長只有幾行：

fromqueueimportQueue
fromthreadingimportThread

defproducer(que):
data=0
whileTrue:
data+=1
que.put(data)

defconsumer(que):
whileTrue:
data=que.get()
print(data)


que=Queue()
t1=Thread(target=consumer,args=(que,))
t2=Thread(target=producer,args=(que,))
 
t1.start()
t2.start()

我們執行一下就會發現它是可行的，並且由於佇列先進先出的限制，可以保證了consumer執行緒讀取到的內容的順序和producer生產的順序是一致的。

如果我們執行一下這個程式碼會發現它是不會結束的，因為consumer和producer當中都用到了while True構建的死迴圈，假設我們希望可以控制程式的結束，應該怎麼辦？

其實也很簡單，我們也可以利用佇列。我們建立一個特殊的訊號量，約定好當consumer接受到這個特殊值的時候就停止程式。這樣當我們要結束程式的時候，我們只需要把這個訊號量加入佇列即可。

singal=object()

defproducer(que):
 
data=0
whiledata<20:
data+=1
que.put(data)
que.put(singal)

defconsumer(que):
whileTrue:
data=que.get()
ifdataissingal:
#繼續插入singal
que.put(singal)
break
print(data)

這裡有一個細節是我們在consumer當中，當讀取到singal的時候，在跳出迴圈之前我們又把singal放回了佇列。原因也很簡單，因為有時候consumer執行緒不止一個，這個singal上游只放置了一個，只會被一個執行緒讀取進來，其他執行緒並不會知道已經獲得了singal的訊息，所以還是會繼續執行。

而當consumer關閉之前放入singal就可以保證每一個consumer在關閉的之前都會再傳遞一個結束的訊號給其他未關閉的consumer讀取。這樣一個一個的傳遞，就可以保證所有consumer都關閉。

這裡還有一個小細節，雖然利用佇列可以解決生產者和消費者通訊的問題，但是上游的生產者並不知道下游的消費者是否已經執行完成了。假如我們想要知道，應該怎麼辦？

Python的設計者們也考慮到了這個問題，所以他們在Queue這個類當中加入了task_done和join方法。利用task_done，消費者可以通知queue這一個任務已經執行完成了。而通過呼叫join，可以等待所有的consumer完成。

fromqueueimportQueue
fromthreadingimportThread

defproducer(que):
data=0
whiledata<20:
data+=1
que.put(data)

defconsumer(que):
whileTrue:
data=que.get()
print(data)
que.task_done()


que=Queue()
t1=Thread(target=consumer,args=(que,))
t2=Thread(target=producer,args=(que,))
t1.start()
t2.start()

que.join()

除了使用task_done之外，我們還可以在que傳遞的訊息當中加入一個Event，這樣我們還可以繼續感知到每一個Event執行的情況。

優先佇列與其他設定

我們之前在介紹一些分散式排程系統的時候曾經說到過，在排程系統當中，排程者會用一個優先佇列來管理所有的任務。當有機器空閒的時候，會有限排程那些優先順序高的任務。

其實這個排程系統也是基於我們剛才介紹的生產消費者模型開發的，只不過將排程佇列從普通佇列換成了優先佇列而已。所以如果我們也希望我們的consumer能夠根據任務的優先順序來改變執行順序的話，也可以使用優先佇列來進行管理任務。

關於優先佇列的實現我們已經很熟悉了，但是有一個問題是我們需要實現掛起等待的阻塞功能。這個我們自己實現是比較麻煩的，但好在我們可以通過呼叫相關的庫來實現。比如threading中的Condition，Condition是一個條件變數可以通知其他執行緒，也可以實現掛起等待。

fromthreadingimportThread,Condition

classPriorityQueue:
def__init__(self):
self._queue=[]
self._cv=Condition()

defput(self,item,priority):
withself._cv:
heapq.heappush(self._queue,(-priority,self._count,item))
#通知下游，喚醒wait狀態的執行緒
self._cv.notify()

defget(self):
withself._cv:
#如果對列為空則掛起
whilelen(self._queue)==0:
self._cv.wait()
#否則返回優先順序最大的
returnheapq.heappop(self._queue)[-1]

最後介紹一下Queue的其他設定，比如我們可以通過size引數設定佇列的大小，由於這是一個阻塞式佇列，所以如果我們設定了佇列的大小，那麼當佇列被裝滿的時候，往其中插入資料的操作也會被阻塞。此時producer執行緒會被掛起，一直到佇列不再滿為止。

當然我們也可以通過block引數將佇列的操作設定成非阻塞。比如que.get(block=False)，那麼當佇列為空的時候，將會丟擲一個佇列為空的異常。同樣，que.put(data, block=False)時也一樣會得到一個佇列已滿的異常。

總結

比如kafka等訊息系統，以及yarn等排程系統等等，幾乎只要是涉及到多執行緒上下游通訊的，往往都會用到。也正因此它的使用場景太廣了，所以它經常在各種面試當中出現，也可以認為是工程師必須知道的幾種基礎設計模式之一，加入我們之後又有更好更多的經典案例：Q群：625223082

另外，佇列也是一個在設計模式以及使用場景當中經常出現的資料結構。從側面也說明了，為什麼演算法和資料結構非常重要，許多大公司喜歡問一些演算法題，也是因為有實際的使用場景，並且的的確確能鍛鍊工程師的思維能力。經常有同學問我演算法和資料結構的使用案例，這就是一個很好的例子。