Python爬蟲(四)--多線程 · Python爬蟲

## [Python-Socket網絡編程](http://www.jianshu.com/p/e062b3dd110c) # 1\. thread模塊 * * * * python是支持多線程的, 主要是通過thread和threading這兩個模塊來實現的。 * python的thread模塊是比較底層的模塊(或者說輕量級)，python的threading模塊是對thread做了一些包裝的，可以更加方便的被使用。簡要的看一下thread模塊中含函數和常量 ~~~ import thread thread.LockType #鎖對象的一種, 用于線程的同步 thread.error #線程的異常 thread.start_new_thread(function, args[, kwargs]) #創建一個新的線程 function : 線程執行函數 args : 線程執行函數的參數, 類似為tuple, kwargs : 是一個字典返回值: 返回線程的標識符 thread.exit() #線程退出函數 thread.allocate_lock() #生成一個未鎖狀態的鎖對象返回值: 返回一個鎖對象 ~~~ `鎖對象`的方法 ~~~ lock.acquire([waitflag]) #獲取鎖無參數時, 無條件獲取鎖, 無法獲取時, 會被阻塞, 知道可以鎖被釋放有參數時, waitflag = 0 時,表示只有在不需要等待的情況下才獲取鎖, 非零情況與上面相同返回值 :　獲得鎖成功返回True, 獲得鎖失敗返回False lock.release() #釋放鎖 lock.locked() #獲取當前鎖的狀態返回值 : 如果鎖已經被某個線程獲取,返回True, 否則為False ~~~ # 1.1\. thread多線程 ~~~ #!/usr/bin/env python # -*- coding:utf-8 -*- import thread import time def print_time(thread_name, delay) : count = 0 while count < 5 : time.sleep(delay) count += 1 print "%s : %s" % (thread_name, time.ctime(time.time())) try : thread.start_new_thread(print_time, ("Thread-1", 2, )) thread.start_new_thread(print_time, ("Thread-2", 4, )) except : print "Error: unable to start the thread" while True : pass ~~~ # 2\. threading模塊 * * * > python的threading模塊是對thread做了一些包裝的，可以更加方便的被使用。經常和[Queue](https://docs.python.org/2/library/queue.html)結合使用,Queue模塊中提供了同步的、線程安全的隊列類，包括`FIFO（先入先出)隊列Queue`，`LIFO（后入先出）隊列LifoQueue`，和`優先級隊列PriorityQueue`。這些隊列都實現了`鎖原語`，能夠在多線程中直接使用。可以使用隊列來實現線程間的同步 ## 2.1\. 常用函數和對象 ~~~ #函數 threading.active_count() #返回當前線程對象Thread的個數 threading.enumerate() #返回當前運行的線程對象Thread(包括后臺的)的list threading.Condition() #返回條件變量對象的工廠函數, 主要用戶線程的并發 threading.current_thread() #返回當前的線程對象Thread, 文檔后面解釋沒看懂 threading.Lock() #返回一個新的鎖對象, 是在thread模塊的基礎上實現的與acquire()和release()結合使用 #類 threading.Thread #一個表示線程控制的類, 這個類常被繼承 thraeding.Timer #定時器,線程在一定時間后執行 threading.ThreadError #引發中各種線程相關異常 ~~~ ### 2.1.1\. Thread對象 > 一般來說，使用線程有兩種模式, 一種是創建線程要執行的函數, 把這個函數傳遞進Thread對象里，讓它來執行. 另一種是直接從Thread繼承，創建一個新的class，把線程執行的代碼放到這個新的class里。常用兩種方式運行線程(線程中包含name屬性) : * 在構造函數中傳入用于線程運行的函數(這種方式更加靈活) * 在子類中重寫threading.Thread基類中run()方法(`只重寫__init__()和run()方法`) 創建線程對象后, 通過調用start()函數運行線程, 然后會自動調用`run()`方法. > 　通過設置｀daemon｀屬性, 可以將線程設置為守護線程 ~~~ threading.Thread(group = None, target = None, name = None, args = () kwars = {}) group : 應該為None target : 可以傳入一個函數用于run()方法調用, name : 線程名默認使用"Thread-N" args : 元組, 表示傳入target函數的參數 kwargs : 字典, 傳入target函數中關鍵字參數屬性: name #線程表示, 沒有任何語義 doemon #布爾值, 如果是守護線程為True, 不是為False, 主線程不是守護線程, 默認threading.Thread.damon = False 類方法: run() #用以表示線程活動的方法。 start() #啟動線程活動。 join([time]) #等待至線程中止。這阻塞調用線程直至線程的join() 方法被調用中止-正常退出或者拋出未處理的異常-或者是可選的超時發生。 isAlive(): 返回線程是否活動的。 getName(): 返回線程名。 setName(): 設置線程名。 ~~~ 范例: ~~~ #!/usr/bin/env python # -*- coding:utf-8 -*- import threading import time def test_thread(count) : while count > 0 : print "count = %d" % count count = count - 1 time.sleep(1) def main() : my_thread = threading.Thread(target = test_thread, args = (10, )) my_thread.start() my_thread.join() if __name__ == '__main__': main() ~~~ ## 2.2\. 常用多線程寫法 * 固定線程運行的函數 ~~~ #!/usr/bin/env python # -*- coding:utf-8 -*- import threading, thread import time class MyThread(threading.Thread): """docstring for MyThread""" def __init__(self, thread_id, name, counter) : super(MyThread, self).__init__() #調用父類的構造函數 self.thread_id = thread_id self.name = name self.counter = counter def run(self) : print "Starting " + self.name print_time(self.name, self.counter, 5) print "Exiting " + self.name def print_time(thread_name, delay, counter) : while counter : time.sleep(delay) print "%s %s" % (thread_name, time.ctime(time.time())) counter -= 1 def main(): #創建新的線程 thread1 = MyThread(1, "Thread-1", 1) thread2 = MyThread(2, "Thread-2", 2) #開啟線程 thread1.start() thread2.start() thread1.join() thread2.join() print "Exiting Main Thread" if __name__ == '__main__': main() ~~~ * 外部傳入線程運行的函數 ~~~ #/usr/bin/env python # -*- coding: utf-8 -*- import threading import time class MyThread(threading.Thread): """ 屬性: target: 傳入外部函數, 用戶線程調用 args: 函數參數 """ def __init__(self, target, args): super(MyThread, self).__init__() #調用父類的構造函數 self.target = target self.args = args def run(self) : self.target(self.args) def print_time(counter) : while counter : print "counter = %d" % counter counter -= 1 time.sleep(1) def main() : my_thread = MyThread(print_time, 10) my_thread.start() my_thread.join() if __name__ == '__main__': main() ~~~ ## 2.3\. 生產者消費者問題 > 試著用python寫了一個生產者消費者問題(偽生產者消費者), 只是使用簡單的鎖, 感覺有點不太對, 下面另一個程序會寫出正確的生產者消費者問題 ~~~ #!/usr/bin/env python # -*- coding:utf-8 -*- import thread, threading import urllib2 import time, random import Queue share_queue = Queue.Queue() #共享隊列 my_lock = thread.allocate_lock() class Producer(threading.Thread) : def run(self) : products = range(5) global share_queue while True : num = random.choice(products) my_lock.acquire() share_queue.put(num) print "Produce : ", num my_lock.release() time.sleep(random.random()) class Consumer(threading.Thread) : def run(self) : global share_queue while True: my_lock.acquire() if share_queue.empty() : #這里沒有使用信號量機制進行阻塞等待, print "Queue is Empty..." my_lock.release() time.sleep(random.random()) continue num = share_queue.get() print "Consumer : ", num my_lock.release() time.sleep(random.random()) def main() : producer = Producer() consumer = Consumer() producer.start() consumer.start() if __name__ == '__main__': main() ~~~ 殺死多線程程序方法: 使用`control + z`掛起程序(程序依然在后臺, 可以使用`ps aux`查看), 獲得程序的進程號, 然后使用`kill -9 進程號`殺死進程 > 參考一篇帖子解決了上述問題,重寫了生產者消費者問題程序, 參考鏈接慣例放在最后. 使用了wait()和notify()解決 > 當然最簡答的方法是直接使用`Queue`,Queue封裝了Condition的行為, 如wait(), notify(), acquire(), 沒看文檔就這樣, 使用了Queue竟然不知道封裝了這些函數, 繼續滾去看文檔了 ~~~ #!/usr/bin/env python # -*- coding:utf-8 -*- import threading import random, time, Queue MAX_SIZE = 5 SHARE_Q = [] #模擬共享隊列 CONDITION = threading.Condition() class Producer(threading.Thread) : def run(self) : products = range(5) global SHARE_Q while True : CONDITION.acquire() if len(SHARE_Q) == 5 : print "Queue is full.." CONDITION.wait() print "Consumer have comsumed something" product = random.choice(products) SHARE_Q.append(product) print "Producer : ", product CONDITION.notify() CONDITION.release() time.sleep(random.random()) class Consumer(threading.Thread) : def run(self) : global SHARE_Q while True: CONDITION.acquire() if not SHARE_Q : print "Queue is Empty..." CONDITION.wait() print "Producer have producted something" product = SHARE_Q.pop(0) print "Consumer :", product CONDITION.notify() CONDITION.release() time.sleep(random.random()) def main() : producer = Producer() consumer = Consumer() producer.start() consumer.start() if __name__ == '__main__': main() ~~~ ## 2.4.簡單鎖 > 如果只是簡單的加鎖解鎖可以直接使用threading.Lock()生成鎖對象, 然后使用acquire()和release()方法例如: ~~~ #!/usr/bin/env python # -*- coding:utf-8 -*- import threading import time class MyThread(threading.Thread) : def __init__(self, thread_id, name, counter) : threading.Thread.__init__(self) self.thread_id = thread_id self.name = name self.counter = counter def run(self) : #重寫run方法, 添加線程執行邏輯, start函數運行會自動執行 print "Starting " + self.name threadLock.acquire() #獲取所 print_time(self.name, self.counter, 3) threadLock.release() #釋放鎖 def print_time(thread_name, delay, counter) : while counter : time.sleep(delay) print "%s %s" % (thread_name, time.ctime(time.time())) counter -= 1 threadLock = threading.Lock() threads = [] #存放線程對象 thread1 = MyThread(1, "Thread-1", 1) thread2 = MyThread(2, "Thread-2", 2) #開啟線程 thread1.start() thread2.start() for t in threads : t.join() #等待線程直到終止 print "Exiting Main Thread" ~~~ ## 2.5\. Condition > 如果是向生產者消費者類似的情形, 使用Condition類或者直接使用`Queue`模塊 **Condition** 條件變量中有`acquire()和release方法用來調用鎖的方法`, 有`wait(), notify(), notifyAll()方法`, 后面是三個方法必須在獲取鎖的情況下調用, 否則產生`RuntimeError`錯誤. * 當一個線程獲得鎖后, 發現沒有期望的資源或者狀態, 就會調用wait()阻塞, 并釋放已經獲得鎖, 知道期望的資源或者狀態發生改變 * 當一個線程獲得鎖, 改變了資源或者狀態, 就會調用notify()和notifyAll()去通知其他線程, ~~~ #官方文檔中提供的生產者消費者模型 # Consume one item cv.acquire() while not an_item_is_available(): cv.wait() get_an_available_item() cv.release() # Produce one item cv.acquire() make_an_item_available() cv.notify() cv.release() ~~~ ~~~ #threading.Condition類 thread.Condition([lock]) 可選參數lock: 必須是Lock或者RLock對象, 并被作為underlying鎖(悲觀鎖?), 否則, 會創建一個新的RLock對象作為underlying鎖類方法: acquire() #獲得鎖 release() #釋放鎖 wait([timeout]) #持續等待直到被notify()或者notifyAll()通知或者超時(必須先獲得鎖), #wait()所做操作, 先釋放獲得的鎖, 然后阻塞, 知道被notify或者notifyAll喚醒或者超時, 一旦被喚醒或者超時, 會重新獲取鎖(應該說搶鎖), 然后返回 notify() #喚醒一個wait()阻塞的線程. notify_all()或者notifyAll() #喚醒所有阻塞的線程 ~~~ > `參考程序可以查看上面的生產者消費者程序` # 3\. 參考鏈接 * * * * [Producer–consumer problem](http://en.wikipedia.org/wiki/Producer%E2%80%93consumer_problem) * [Python中的生產者消費者問題](http://blog.jobbole.com/52412/) * [Python多線程](http://www.w3cschool.cc/python/python-multithreading.html) * [threading官方文檔](https://docs.python.org/2/library/threading.html) * [thread官方文檔](https://docs.python.org/2/library/thread.html#module-thread)