Queue（隊列對象）

Queue是python中的標準庫，可以直接import Queue引用;隊列是線程間最常用的交換數據的形式

python下多線程的思考

對於資源，加鎖是個重要的環節。因為python原生的list,dict等，都是not thread safe的。而Queue，是線程安全的，因此在滿足使用條件下，建議使用隊列

1. 初始化： class Queue.Queue(maxsize) FIFO 先進先出

2. 包中的常用方法:

   • Queue.qsize() 返回隊列的大小

   • Queue.empty() 如果隊列為空，返回True,反之False

   • Queue.full() 如果隊列滿了，返回True,反之False

   • Queue.full 與 maxsize 大小對應

   • Queue.get([block[, timeout]])獲取隊列，timeout等待時間

3. 創建一個“隊列”對象

   • import Queue
   • myqueue = Queue.Queue(maxsize = 10)

4. 將一個值放入隊列中

   • myqueue.put(10)

5. 將一個值從隊列中取出

• myqueue.get()

  1 # -*- coding:utf-8 -*-
  2 
  3 # 使用了線程庫
  4 import threading
  5 # 隊列
  6 from Queue import Queue
 
  7 # 解析庫
  8 from lxml import etree
  9 # 請求處理
 10 import requests
 11 # json處理
 12 import json
 13 import time
 14 
 15 class ThreadCrawl(threading.Thread):
 16     def __init__(self, threadName, pageQueue, dataQueue):
 17         #threading.Thread.__init__(self)
 18         # 調用父類初始化方法
 19         super(ThreadCrawl, self).__init__
 
()
 20         # 線程名
 21         self.threadName = threadName
 22         # 頁碼隊列
 23         self.pageQueue = pageQueue
 24         # 數據隊列
 25         self.dataQueue = dataQueue
 26         # 請求報頭
 27         self.headers = {"User-Agent" : "Mozilla/5.0 (compatible; MSIE 9.0; Windows NT 6.1; Trident/5.0;"}
 28 
 29     def run(self):
 30         print "啟動 " + self.threadName
 31         while not CRAWL_EXIT:
 32             try:
 33                 # 取出一個數字，先進先出
 34                 # 可選參數block，默認值為True
 35                 #1. 如果對列為空，block為True的話，不會結束，會進入阻塞狀態，直到隊列有新的數據
 36                 #2. 如果隊列為空，block為False的話，就彈出一個Queue.empty()異常，
 37                 page = self.pageQueue.get(False)
 38                 url = "http://www.qiushibaike.com/8hr/page/" + str(page) +"/"
 39                 #print url
 40                 content = requests.get(url, headers = self.headers).text
 41                 time.sleep(1)
 42                 self.dataQueue.put(content)
 43                 #print len(content)
 44             except:
 45                 pass
 46         print "結束 " + self.threadName
 47 
 48 class ThreadParse(threading.Thread):
 49     def __init__(self, threadName, dataQueue, filename, lock):
 50         super(ThreadParse, self).__init__()
 51         # 線程名
 52         self.threadName = threadName
 53         # 數據隊列
 54         self.dataQueue = dataQueue
 55         # 保存解析後數據的文件名
 56         self.filename = filename
 57         # 鎖
 58         self.lock = lock
 59 
 60     def run(self):
 61         print "啟動" + self.threadName
 62         while not PARSE_EXIT:
 63             try:
 64                 html = self.dataQueue.get(False)
 65                 self.parse(html)
 66             except:
 67                 pass
 68         print "退出" + self.threadName
 69 
 70     def parse(self, html):
 71         # 解析為HTML DOM
 72         html = etree.HTML(html)
 73 
 74         node_list = html.xpath(‘//div[contains(@id, "qiushi_tag")]‘)
 75 
 76         for node in node_list:
 77             # xpath返回的列表，這個列表就這一個參數，用索引方式取出來，用戶名
 78             username = node.xpath(‘.//img/@alt‘)[0]
 79             # 圖片連接
 80             image = node.xpath(‘.//div[@class="thumb"]//@src‘)#[0]
 81             # 取出標簽下的內容,段子內容
 82             content = node.xpath(‘.//div[@class="content"]/span‘)[0].text
 83             # 取出標簽裏包含的內容，點贊
 84             zan = node.xpath(‘.//i‘)[0].text
 85             # 評論
 86             comments = node.xpath(‘.//i‘)[1].text
 87 
 88             items = {
 89                 "username" : username,
 90                 "image" : image,
 91                 "content" : content,
 92                 "zan" : zan,
 93                 "comments" : comments
 94             }
 95 
 96             # with 後面有兩個必須執行的操作：__enter__ 和 _exit__
 97             # 不管裏面的操作結果如何，都會執行打開、關閉
 98             # 打開鎖、處理內容、釋放鎖
 99             with self.lock:
100                 # 寫入存儲的解析後的數據
101                 self.filename.write(json.dumps(items, ensure_ascii = False).encode("utf-8") + "\n")
102 
103 CRAWL_EXIT = False
104 PARSE_EXIT = False
105 
106 
107 def main():
108     # 頁碼的隊列，表示20個頁面
109     pageQueue = Queue(20)
110     # 放入1~10的數字，先進先出
111     for i in range(1, 21):
112         pageQueue.put(i)
113 
114     # 采集結果(每頁的HTML源碼)的數據隊列，參數為空表示不限制
115     dataQueue = Queue()
116 
117     filename = open("duanzi.json", "a")
118     # 創建鎖
119     lock = threading.Lock()
120 
121     # 三個采集線程的名字
122     crawlList = ["采集線程1號", "采集線程2號", "采集線程3號"]
123     # 存儲三個采集線程的列表集合
124     threadcrawl = []
125     for threadName in crawlList:
126         thread = ThreadCrawl(threadName, pageQueue, dataQueue)
127         thread.start()
128         threadcrawl.append(thread)
129 
130 
131     # 三個解析線程的名字
132     parseList = ["解析線程1號","解析線程2號","解析線程3號"]
133     # 存儲三個解析線程
134     threadparse = []
135     for threadName in parseList:
136         thread = ThreadParse(threadName, dataQueue, filename, lock)
137         thread.start()
138         threadparse.append(thread)
139 
140     # 等待pageQueue隊列為空，也就是等待之前的操作執行完畢
141     while not pageQueue.empty():
142         pass
143 
144     # 如果pageQueue為空，采集線程退出循環
145     global CRAWL_EXIT
146     CRAWL_EXIT = True
147 
148     print "pageQueue為空"
149 
150     for thread in threadcrawl:
151         thread.join()
152         print "1"
153 
154     while not dataQueue.empty():
155         pass
156 
157     global PARSE_EXIT
158     PARSE_EXIT = True
159 
160     for thread in threadparse:
161         thread.join()
162         print "2"
163 
164     with lock:
165         # 關閉文件
166         filename.close()
167     print "謝謝使用！"
168 
169 if __name__ == "__main__":
170     main()

多線程糗事百科案例