python - 經常使用模塊 - queue

Python中，隊列是線程間最經常使用的交換數據的形式。queue模塊是提供隊列操做的模塊，雖然簡單易用，可是不當心的話，仍是會出現一些意外。

一、queue簡單說明

 1 #!/usr/bin/env python
 2 # -*- coding: utf-8 -*-
 3 
 4 '''
 5  queue隊列：經常使用在多線程裏面，可以直接在線程安全的在多個線程之間進行數據交換，不須要當參數傳遞
 6  
 7  class queue.Queue(maxsize=0)  #先進先出
 8  class queue.LifoQueue(maxsize=0)  #後進先出 last in first out
 9  class queue.PriorityQueue(maxsize=0)  #存儲數據時能夠設置優先級的隊列
10  
11  隊列中能夠存聽任意類型數據
12  
13  隊列經常使用方法：
14     put(x)  將x存儲到隊列，存不下就就會阻塞，timeout設置阻塞超時時間，不設置會一直阻塞
15     put_nowait(x)  不阻塞，放不進去就報錯
16     get()  從隊列中取出一個值（先進先出），其他的跟put同樣
17     get_nowait()  從隊列中取出一個值，隊列沒有值時直接報錯
18     full()  判斷隊列是否滿了
19     empty()  判斷隊列是否爲空
20     qsize()  返回隊列實際的長度(存了多少個值)
21     
22 '''
23 
24 import queue
25 
26 class Human(object):
27     def __init__(self,n):
28         self.n = n
29 
30 print('\033[31;1m 先進先出隊列Queue \033[0m')
31 q = queue.Queue(maxsize=3)   #隊列中最多存儲幾個值,能夠不加maxsize
32 # q.get() #get是阻塞函數，當隊列裏面沒有數據時，就會一直等待
33 # q.get(timeout=3)   # 阻塞3秒，沒有數據就報錯
34 # q.get_nowait()  #不阻塞，沒有數據就直接報錯 queue.Empty Error
35 q.put([1,2,'a',[5,6]])  #這裏傳入的列表在隊列中只是一個數據
36 q.put(1)
37 
38 h = Human(2)
39 q.put(h)
40 print(q.full())   # full 判斷隊列是否存滿了  empty 判斷隊列是否爲空
41 # q.put(4,timeout=2)
42 
43 print('\033[31;1m 先進後出隊列LifoQueue \033[0m')
44 
45 lq = queue.LifoQueue()
46 for i in range(5):
47     lq.put(i)
48 print('LifoQueue values:',lq) #LifoQueue values: <queue.LifoQueue object at 0x0000022EA3959DA0>
49 print('取第一個值：',lq.get())
50 
51 print('\033[31;1m 設置優先級隊列PriorityQueue \033[0m')
52 pq = queue.PriorityQueue()
53 pq.put((1,'asb'))
54 pq.put((10,223))     #優先級和數據以元組兩個值的形式傳遞給put，由於put第二個參數默認是timeout
55 pq.put((8,[1,5]))
56 pq.put((3,(1,'c')))
57 
58 print(pq.get())
59 print(pq.get())
60 print(pq.get())
61 print(pq.get())
62 '''
63 (1, 'asb')
64 (3, (1, 'c'))
65 (8, [1, 5])
66 (10, 223)
67 '''

　　執行結果

 先進先出隊列Queue 
True
 先進後出隊列LifoQueue 
LifoQueue values: <queue.LifoQueue object at 0x000001C9D6CE3CC0>
取第一個值： 4
 設置優先級隊列PriorityQueue 
(1, 'asb')
(3, (1, 'c'))
(8, [1, 5])
(10, 223)

 

二、queue簡單使用

 1 #!/usr/bin/env python
 2 # -*- coding: utf-8 -*-
 3 
 4 import queue
 5 import threading
 6 import time
 7 
 8 def get_value(i,q):
 9     time.sleep(2)
10     value = q.get()
11     print('線程[%s]獲取的值：%s' %(i,value))
12 
13 if __name__ == '__main__':
14     q = queue.Queue()
15     for i in range(20):
16         q.put(i**2)
17 
18     thread_list = []
19     for i in range(20):
20         t = threading.Thread(target=get_value,args=[i,q,])
21         t.start()
22         thread_list.append(t)
23 
24     for i in thread_list:
25         i.join()
26 
27 print('\033[32;1m ----main thread end----\033[0m')

　　執行結果

線程[3]獲取的值：0
線程[2]獲取的值：1
線程[1]獲取的值：4
線程[0]獲取的值：9
線程[18]獲取的值：16
線程[19]獲取的值：25
線程[17]獲取的值：36
線程[10]獲取的值：49
線程[15]獲取的值：64
線程[14]獲取的值：81
線程[12]獲取的值：100
線程[16]獲取的值：121
線程[11]獲取的值：144
線程[9]獲取的值：169
線程[8]獲取的值：196
線程[7]獲取的值：225
線程[5]獲取的值：256
線程[13]獲取的值：289
線程[6]獲取的值：324
線程[4]獲取的值：361
 ----main thread end----

 

三、基於queue的生產者消費者模型

 1 #!/usr/bin/env python
 2 # -*- coding: utf-8 -*-
 3 
 4 import threading
 5 import queue
 6 import time
 7 
 8 def customer(n):
 9     while True:
10         # print('hello')
11         print("\033[32;1m customer [%s] \033[0m get task: %s" % (n,q.get()))
12             #消費者線程第一次執行到get時會阻塞，因此每次都是生產者在前面
13         time.sleep(1)
14         q.task_done()  #每一個從隊列裏面取值的線程執行結束後，都會通知隊列說任務結束
15 
16 def producer(n):
17     count = 1
18     while True:
19         print("producer [%s] producted a new task: %s " %(n,count))
20         q.put(count)
21         count += 1
22         q.join()  #當隊列裏面還有值就處於阻塞狀態，沒值時就會解除阻塞(阻塞狀態是不佔用任何資源的)
23         print('all task has benn cosumed by consumers...')
24 
25 
26 if __name__ == '__main__':
27     q = queue.Queue()
28     thread_list = [] #用來存放全部的線程，目的是後面的join，讓父線程等這些線程結束後再結束
29 
30     for i in range(3):
31         c = threading.Thread(target=customer,args=[i,])
32         c.start()
33         thread_list.append(c)
34 
35     for i in range(2):
36         p = threading.Thread(target=producer,args=[i,])
37         p.start()
38         thread_list.append(p)
39 
40     for thread in thread_list:
41         thread.join()
42 
43 
44     # print("\033[32;1m ----main thread end---- \033[0m")

　　執行結果

producer [0] producted a new task: 1 
 customer [0]  get task: 1
producer [1] producted a new task: 1 
 customer [1]  get task: 1
all task has benn cosumed by consumers...
all task has benn cosumed by consumers...
producer [0] producted a new task: 2 
producer [1] producted a new task: 2 
 customer [2]  get task: 2
 customer [1]  get task: 2
all task has benn cosumed by consumers...
all task has benn cosumed by consumers...
producer [0] producted a new task: 3 
producer [1] producted a new task: 3 
 customer [0]  get task: 3
 customer [1]  get task: 3
all task has benn cosumed by consumers...
producer [0] producted a new task: 4 
 customer [2]  get task: 4
all task has benn cosumed by consumers...
all task has benn cosumed by consumers...
producer [0] producted a new task: 5 
producer [1] producted a new task: 4 
 customer [0]  get task: 4
 customer [1]  get task: 5