Python爬虫进阶五之多线程的用法
前言
我們之前寫的爬蟲都是單個(gè)線程的？這怎么夠？一旦一個(gè)地方卡到不動(dòng)了，那不就永遠(yuǎn)等待下去了？為此我們可以使用多線程或者多進(jìn)程來處理。
首先聲明一點(diǎn)！
多線程和多進(jìn)程是不一樣的！一個(gè)是 thread 庫，一個(gè)是 multiprocessing 庫。而多線程 thread 在 Python 里面被稱作雞肋的存在！而沒錯(cuò)！本節(jié)介紹的是就是這個(gè)庫 thread。
不建議你用這個(gè)，不過還是介紹下了，如果想看可以看看下面，不想浪費(fèi)時(shí)間直接看
multiprocessing 多進(jìn)程
雞肋點(diǎn)
名言：
“Python下多線程是雞肋，推薦使用多進(jìn)程！”
那當(dāng)然有同學(xué)會(huì)問了，為啥？
背景
1、GIL是什么？
GIL的全稱是Global Interpreter Lock(全局解釋器鎖)，來源是python設(shè)計(jì)之初的考慮，為了數(shù)據(jù)安全所做的決定。
2、每個(gè)CPU在同一時(shí)間只能執(zhí)行一個(gè)線程（在單核CPU下的多線程其實(shí)都只是并發(fā)，不是并行，并發(fā)和并行從宏觀上來講都是同時(shí)處理多路請求的概念。但并發(fā)和并行又有區(qū)別，并行是指兩個(gè)或者多個(gè)事件在同一時(shí)刻發(fā)生；而并發(fā)是指兩個(gè)或多個(gè)事件在同一時(shí)間間隔內(nèi)發(fā)生。）
在Python多線程下，每個(gè)線程的執(zhí)行方式：
- 獲取GIL
- 執(zhí)行代碼直到sleep或者是python虛擬機(jī)將其掛起。
- 釋放GIL
可見，某個(gè)線程想要執(zhí)行，必須先拿到GIL，我們可以把GIL看作是“通行證”，并且在一個(gè)python進(jìn)程中，GIL只有一個(gè)。拿不到通行證的線程，就不允許進(jìn)入CPU執(zhí)行。
在Python2.x里，GIL的釋放邏輯是當(dāng)前線程遇見IO操作或者ticks計(jì)數(shù)達(dá)到100（ticks可以看作是Python自身的一個(gè)計(jì)數(shù)器，專門做用于GIL，每次釋放后歸零，這個(gè)計(jì)數(shù)可以通過 sys.setcheckinterval 來調(diào)整），進(jìn)行釋放。
而每次釋放GIL鎖，線程進(jìn)行鎖競爭、切換線程，會(huì)消耗資源。并且由于GIL鎖存在，python里一個(gè)進(jìn)程永遠(yuǎn)只能同時(shí)執(zhí)行一個(gè)線程(拿到GIL的線程才能執(zhí)行)，這就是為什么在多核CPU上，python的多線程效率并不高。
那么是不是python的多線程就完全沒用了呢？
在這里我們進(jìn)行分類討論：
1、CPU密集型代碼(各種循環(huán)處理、計(jì)數(shù)等等)，在這種情況下，由于計(jì)算工作多，ticks計(jì)數(shù)很快就會(huì)達(dá)到閾值，然后觸發(fā)GIL的釋放與再競爭（多個(gè)線程來回切換當(dāng)然是需要消耗資源的），所以python下的多線程對CPU密集型代碼并不友好。
2、IO密集型代碼(文件處理、網(wǎng)絡(luò)爬蟲等)，多線程能夠有效提升效率(單線程下有IO操作會(huì)進(jìn)行IO等待，造成不必要的時(shí)間浪費(fèi)，而開啟多線程能在線程A等待時(shí)，自動(dòng)切換到線程B，可以不浪費(fèi)CPU的資源，從而能提升程序執(zhí)行效率)。所以python的多線程對IO密集型代碼比較友好。
而在python3.x中，GIL不使用ticks計(jì)數(shù)，改為使用計(jì)時(shí)器（執(zhí)行時(shí)間達(dá)到閾值后，當(dāng)前線程釋放GIL），這樣對CPU密集型程序更加友好，但依然沒有解決GIL導(dǎo)致的同一時(shí)間只能執(zhí)行一個(gè)線程的問題，所以效率依然不盡如人意。
多核性能
多核多線程比單核多線程更差，原因是單核下多線程，每次釋放GIL，喚醒的那個(gè)線程都能獲取到GIL鎖，所以能夠無縫執(zhí)行，但多核下，CPU0釋放GIL后，其他CPU上的線程都會(huì)進(jìn)行競爭，但GIL可能會(huì)馬上又被CPU0拿到，導(dǎo)致其他幾個(gè)CPU上被喚醒后的線程會(huì)醒著等待到切換時(shí)間后又進(jìn)入待調(diào)度狀態(tài)，這樣會(huì)造成線程顛簸(thrashing)，導(dǎo)致效率更低
多進(jìn)程為什么不會(huì)這樣？
每個(gè)進(jìn)程有各自獨(dú)立的GIL，互不干擾，這樣就可以真正意義上的并行執(zhí)行，所以在python中，多進(jìn)程的執(zhí)行效率優(yōu)于多線程(僅僅針對多核CPU而言)。
所以在這里說結(jié)論：多核下，想做并行提升效率，比較通用的方法是使用多進(jìn)程，能夠有效提高執(zhí)行效率。
所以，如果不想浪費(fèi)時(shí)間，可以直接看多進(jìn)程。
直接利用函數(shù)創(chuàng)建多線程
Python中使用線程有兩種方式：函數(shù)或者用類來包裝線程對象。
函數(shù)式：調(diào)用thread模塊中的start_new_thread()函數(shù)來產(chǎn)生新線程。語法如下：
?
| 1 | <span class="s1">thread</span><span class="s2">.</span><span class="s1">start_new_thread </span><span class="s2">(</span> <span class="s3">function</span><span class="s2">,</span><span class="s1"> args</span><span class="s2">[,</span><span class="s1"> kwargs</span><span class="s2">]</span> <span class="s2">)</span> |
?
參數(shù)說明:
- function – 線程函數(shù)。
- args – 傳遞給線程函數(shù)的參數(shù),他必須是個(gè)tuple類型。
- kwargs – 可選參數(shù)。
先用一個(gè)實(shí)例感受一下：
?
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 | # -*- coding: UTF-8 -*- import thread import time # 為線程定義一個(gè)函數(shù) def print_time(threadName, delay): ????count = 0 ????while count < 5: ????????time.sleep(delay) ????????count += 1 ????????print "%s: %s" % (threadName, time.ctime(time.time())) # 創(chuàng)建兩個(gè)線程 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 thread" while 1: ?? pass print "Main Finished" |
?
運(yùn)行結(jié)果如下：
?
| 1 2 3 4 5 6 7 8 9 10 | Thread-1: Thu Nov??3 16:43:01 2016 Thread-2: Thu Nov??3 16:43:03 2016 Thread-1: Thu Nov??3 16:43:03 2016 Thread-1: Thu Nov??3 16:43:05 2016 Thread-2: Thu Nov??3 16:43:07 2016 Thread-1: Thu Nov??3 16:43:07 2016 Thread-1: Thu Nov??3 16:43:09 2016 Thread-2: Thu Nov??3 16:43:11 2016 Thread-2: Thu Nov??3 16:43:15 2016 Thread-2: Thu Nov??3 16:43:19 2016 |
?
可以發(fā)現(xiàn)，兩個(gè)線程都在執(zhí)行，睡眠2秒和4秒后打印輸出一段話。
注意到，在主線程寫了
?
| 1 2 | while 1: ?? pass |
?
這是讓主線程一直在等待
如果去掉上面兩行，那就直接輸出
?
| 1 | Main Finished |
?
程序執(zhí)行結(jié)束。
使用Threading模塊創(chuàng)建線程
使用Threading模塊創(chuàng)建線程，直接從threading.Thread繼承，然后重寫init方法和run方法：
?
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 | #!/usr/bin/python # -*- coding: UTF-8 -*- import threading import time import thread exitFlag = 0 class myThread (threading.Thread):?? #繼承父類threading.Thread ????def __init__(self, threadID, name, counter): ????????threading.Thread.__init__(self) ????????self.threadID = threadID ????????self.name = name ????????self.counter = counter ????def run(self):?????????????????? #把要執(zhí)行的代碼寫到run函數(shù)里面 線程在創(chuàng)建后會(huì)直接運(yùn)行run函數(shù) ????????print "Starting " + self.name ????????print_time(self.name, self.counter, 5) ????????print "Exiting " + self.name def print_time(threadName, delay, counter): ????while counter: ????????if exitFlag: ????????????thread.exit() ????????time.sleep(delay) ????????print "%s: %s" % (threadName, time.ctime(time.time())) ????????counter -= 1 # 創(chuàng)建新線程 thread1 = myThread(1, "Thread-1", 1) thread2 = myThread(2, "Thread-2", 2) # 開啟線程 thread1.start() thread2.start() print "Exiting Main Thread" |
?
運(yùn)行結(jié)果：
?
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 | Starting Thread-1Starting Thread-2 Exiting Main Thread Thread-1: Thu Nov??3 18:42:19 2016 Thread-2: Thu Nov??3 18:42:20 2016 Thread-1: Thu Nov??3 18:42:20 2016 Thread-1: Thu Nov??3 18:42:21 2016 Thread-2: Thu Nov??3 18:42:22 2016 Thread-1: Thu Nov??3 18:42:22 2016 Thread-1: Thu Nov??3 18:42:23 2016 Exiting Thread-1 Thread-2: Thu Nov??3 18:42:24 2016 Thread-2: Thu Nov??3 18:42:26 2016 Thread-2: Thu Nov??3 18:42:28 2016 Exiting Thread-2 |
?
有沒有發(fā)現(xiàn)什么奇怪的地方？打印的輸出格式好奇怪。比如第一行之后應(yīng)該是一個(gè)回車的，結(jié)果第二個(gè)進(jìn)程就打印出來了。
那是因?yàn)槭裁?#xff1f;因?yàn)檫@幾個(gè)線程沒有設(shè)置同步。
線程同步
如果多個(gè)線程共同對某個(gè)數(shù)據(jù)修改，則可能出現(xiàn)不可預(yù)料的結(jié)果，為了保證數(shù)據(jù)的正確性，需要對多個(gè)線程進(jìn)行同步。
使用Thread對象的Lock和Rlock可以實(shí)現(xiàn)簡單的線程同步，這兩個(gè)對象都有acquire方法和release方法，對于那些需要每次只允許一個(gè)線程操作的數(shù)據(jù)，可以將其操作放到acquire和release方法之間。如下：
多線程的優(yōu)勢在于可以同時(shí)運(yùn)行多個(gè)任務(wù)（至少感覺起來是這樣）。但是當(dāng)線程需要共享數(shù)據(jù)時(shí)，可能存在數(shù)據(jù)不同步的問題。
考慮這樣一種情況：一個(gè)列表里所有元素都是0，線程”set”從后向前把所有元素改成1，而線程”print”負(fù)責(zé)從前往后讀取列表并打印。
那么，可能線程”set”開始改的時(shí)候，線程”print”便來打印列表了，輸出就成了一半0一半1，這就是數(shù)據(jù)的不同步。為了避免這種情況，引入了鎖的概念。
鎖有兩種狀態(tài)——鎖定和未鎖定。每當(dāng)一個(gè)線程比如”set”要訪問共享數(shù)據(jù)時(shí)，必須先獲得鎖定；如果已經(jīng)有別的線程比如”print”獲得鎖定了，那么就讓線程”set”暫停，也就是同步阻塞；等到線程”print”訪問完畢，釋放鎖以后，再讓線程”set”繼續(xù)。
經(jīng)過這樣的處理，打印列表時(shí)要么全部輸出0，要么全部輸出1，不會(huì)再出現(xiàn)一半0一半1的尷尬場面。
看下面的例子：
?
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 | # -*- coding: UTF-8 -*- import threading import time class myThread (threading.Thread): ????def __init__(self, threadID, name, counter): ????????threading.Thread.__init__(self) ????????self.threadID = threadID ????????self.name = name ????????self.counter = counter ????def run(self): ????????print "Starting " + self.name ?????? # 獲得鎖，成功獲得鎖定后返回True ?????? # 可選的timeout參數(shù)不填時(shí)將一直阻塞直到獲得鎖定 ?????? # 否則超時(shí)后將返回False ????????threadLock.acquire() ????????print_time(self.name, self.counter, 3) ????????# 釋放鎖 ????????threadLock.release() def print_time(threadName, delay, counter): ????while counter: ????????time.sleep(delay) ????????print "%s: %s" % (threadName, time.ctime(time.time())) ????????counter -= 1 threadLock = threading.Lock() threads = [] # 創(chuàng)建新線程 thread1 = myThread(1, "Thread-1", 1) thread2 = myThread(2, "Thread-2", 2) # 開啟新線程 thread1.start() thread2.start() # 添加線程到線程列表 threads.append(thread1) threads.append(thread2) # 等待所有線程完成 for t in threads: ????t.join() print "Exiting Main Thread" |
?
在上面的代碼中運(yùn)用了線程鎖還有join等待。
運(yùn)行結(jié)果如下：
?
| 1 2 3 4 5 6 7 8 9 | Starting Thread-1 Starting Thread-2 Thread-1: Thu Nov??3 18:56:49 2016 Thread-1: Thu Nov??3 18:56:50 2016 Thread-1: Thu Nov??3 18:56:51 2016 Thread-2: Thu Nov??3 18:56:53 2016 Thread-2: Thu Nov??3 18:56:55 2016 Thread-2: Thu Nov??3 18:56:57 2016 Exiting Main Thread |
?
這樣一來，你可以發(fā)現(xiàn)就不會(huì)出現(xiàn)剛才的輸出混亂的結(jié)果了。
線程優(yōu)先級隊(duì)列
Python的Queue模塊中提供了同步的、線程安全的隊(duì)列類，包括FIFO（先入先出)隊(duì)列Queue，LIFO（后入先出）隊(duì)列LifoQueue，和優(yōu)先級隊(duì)列PriorityQueue。這些隊(duì)列都實(shí)現(xiàn)了鎖原語，能夠在多線程中直接使用。可以使用隊(duì)列來實(shí)現(xiàn)線程間的同步。
Queue模塊中的常用方法:
- Queue.qsize() 返回隊(duì)列的大小
- Queue.empty() 如果隊(duì)列為空，返回True,反之False
- Queue.full() 如果隊(duì)列滿了，返回True,反之False
- Queue.full 與 maxsize 大小對應(yīng)
- Queue.get([block[, timeout]])獲取隊(duì)列，timeout等待時(shí)間
- Queue.get_nowait() 相當(dāng)Queue.get(False)
- Queue.put(item) 寫入隊(duì)列，timeout等待時(shí)間
- Queue.put_nowait(item) 相當(dāng)Queue.put(item, False)
- Queue.task_done() 在完成一項(xiàng)工作之后，Queue.task_done()函數(shù)向任務(wù)已經(jīng)完成的隊(duì)列發(fā)送一個(gè)信號
- Queue.join() 實(shí)際上意味著等到隊(duì)列為空，再執(zhí)行別的操作
用一個(gè)實(shí)例感受一下：
?
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 | # -*- coding: UTF-8 -*- import Queue import threading import time exitFlag = 0 class myThread (threading.Thread): ????def __init__(self, threadID, name, q): ????????threading.Thread.__init__(self) ????????self.threadID = threadID ????????self.name = name ????????self.q = q ????def run(self): ????????print "Starting " + self.name ????????process_data(self.name, self.q) ????????print "Exiting " + self.name def process_data(threadName, q): ????while not exitFlag: ????????queueLock.acquire() ????????if not workQueue.empty(): ????????????data = q.get() ????????????queueLock.release() ????????????print "%s processing %s" % (threadName, data) ????????else: ????????????queueLock.release() ????????time.sleep(1) threadList = ["Thread-1", "Thread-2", "Thread-3"] nameList = ["One", "Two", "Three", "Four", "Five"] queueLock = threading.Lock() workQueue = Queue.Queue(10) threads = [] threadID = 1 # 創(chuàng)建新線程 for tName in threadList: ????thread = myThread(threadID, tName, workQueue) ????thread.start() ????threads.append(thread) ????threadID += 1 # 填充隊(duì)列 queueLock.acquire() for word in nameList: ????workQueue.put(word) queueLock.release() # 等待隊(duì)列清空 while not workQueue.empty(): ????pass # 通知線程是時(shí)候退出 exitFlag = 1 # 等待所有線程完成 for t in threads: ????t.join() print "Exiting Main Thread" |
?
運(yùn)行結(jié)果：
?
| 1 2 3 4 5 6 7 8 9 10 11 12 | Starting Thread-1 Starting Thread-2 Starting Thread-3 Thread-3 processing One Thread-1 processing Two Thread-2 processing Three Thread-3 processing Four Thread-2 processing Five Exiting Thread-2 Exiting Thread-3 Exiting Thread-1 Exiting Main Thread |
?
上面的例子用了FIFO隊(duì)列。當(dāng)然你也可以換成其他類型的隊(duì)列。
參考文章
2.?http://www.runoob.com/python/python-multithreading.html
轉(zhuǎn)載：靜覓???Python爬蟲進(jìn)階五之多線程的用法
轉(zhuǎn)載于:https://www.cnblogs.com/BigFishFly/p/6380048.html
與50位技術(shù)專家面對面20年技術(shù)見證，附贈(zèng)技術(shù)全景圖總結(jié)
以上是生活随笔為你收集整理的Python爬虫进阶五之多线程的用法的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： Linux权限管理 - 特殊权限之文件特
- 下一篇： Spring Boot 入门例子 Hel