網絡爬蟲之第四章爬蟲進階之多線程爬蟲

時間 2019-12-14

原文原文鏈接

多線程爬蟲

有些時候，好比下載圖片，由於下載圖片是一個耗時的操做。若是採用以前那種同步的方式下載。那效率肯會特別慢。這時候咱們就能夠考慮使用多線程的方式來下載圖片。html

多線程介紹：

多線程是爲了同步完成多項任務，經過提升資源使用效率來提升系統的效率。線程是在同一時間須要完成多項任務的時候實現的。
最簡單的比喻多線程就像火車的每一節車箱，而進程則是火車。車箱離開火車是沒法跑動的，同理火車也能夠有多節車箱。多線程的出現就是爲了提升效率。同時它的出現也帶來了一些問題。更多介紹請參考：https://baike.baidu.com/item/多線程/1190404?fr=aladdinpython

threading模塊介紹：

threading模塊是python中專門提供用來作多線程編程的模塊。threading模塊中最經常使用的類是Thread。如下看一個簡單的多線程程序：編程

import threading
import time

def coding():
    for x in range(3):
        print('%s正在寫代碼' % x)
        time.sleep(1)

def drawing():
    for x in range(3):
        print('%s正在畫圖' % x)
        time.sleep(1)


def single_thread():
    coding()
    drawing()

def multi_thread():
    t1 = threading.Thread(target=coding)
    t2 = threading.Thread(target=drawing)

    t1.start()
    t2.start()

if __name__ == '__main__':
    multi_thread()

查看線程數：

使用threading.enumerate()函數即可以看到當前線程的數量。安全

查看當前線程的名字：

使用threading.current_thread()能夠看到當前線程的信息。網絡

繼承自`threading.Thread`類：

爲了讓線程代碼更好的封裝。可使用threading模塊下的Thread類，繼承自這個類，而後實現run方法，線程就會自動運行run方法中的代碼。示例代碼以下：多線程

import threading
import time

class CodingThread(threading.Thread):
    def run(self):
        for x in range(3):
            print('%s正在寫代碼' % threading.current_thread())
            time.sleep(1)

class DrawingThread(threading.Thread):
    def run(self):
        for x in range(3):
            print('%s正在畫圖' % threading.current_thread())
            time.sleep(1)

def multi_thread():
    t1 = CodingThread()
    t2 = DrawingThread()

    t1.start()
    t2.start()

if __name__ == '__main__':
    multi_thread()

多線程共享全局變量的問題：

多線程都是在同一個進程中運行的。所以在進程中的全局變量全部線程都是可共享的。這就形成了一個問題，由於線程執行的順序是無序的。有可能會形成數據錯誤。好比如下代碼：app

import threading

tickets = 0

def get_ticket():
    global tickets
    for x in range(1000000):
        tickets += 1
    print('tickets:%d'%tickets)

def main():
    for x in range(2):
        t = threading.Thread(target=get_ticket)
        t.start()

if __name__ == '__main__':
    main()

以上結果正常來說應該是6，可是由於多線程運行的不肯定性。所以最後的結果多是隨機的。dom

鎖機制：

爲了解決以上使用共享全局變量的問題。threading提供了一個Lock類，這個類能夠在某個線程訪問某個變量的時候加鎖，其餘線程此時就不能進來，直到當前線程處理完後，把鎖釋放了，其餘線程才能進來處理。示例代碼以下：ide

import threading

VALUE = 0

gLock = threading.Lock()

def add_value():
    global VALUE
    gLock.acquire()
    for x in range(1000000):
        VALUE += 1
    gLock.release()
    print('value：%d'%VALUE)

def main():
    for x in range(2):
        t = threading.Thread(target=add_value)
        t.start()

if __name__ == '__main__':
    main()

Lock版本生產者和消費者模式：

生產者和消費者模式是多線程開發中常常見到的一種模式。生產者的線程專門用來生產一些數據，而後存放到一箇中間的變量中。消費者再從這個中間的變量中取出數據進行消費。可是由於要使用中間變量，中間變量常常是一些全局變量，所以須要使用鎖來保證數據完整性。如下是使用threading.Lock鎖實現的「生產者與消費者模式」的一個例子：函數

import threading
import random
import time

gMoney = 1000
gLock = threading.Lock()
# 記錄生產者生產的次數，達到10次就再也不生產
gTimes = 0

class Producer(threading.Thread):
    def run(self):
        global gMoney
        global gLock
        global gTimes
        while True:
            money = random.randint(100, 1000)
            gLock.acquire()
            # 若是已經達到10次了，就再也不生產了
            if gTimes >= 10:
                gLock.release()
                break
            gMoney += money
            print('%s當前存入%s元錢，剩餘%s元錢' % (threading.current_thread(), money, gMoney))
            gTimes += 1
            time.sleep(0.5)
            gLock.release()

class Consumer(threading.Thread):
    def run(self):
        global gMoney
        global gLock
        global gTimes
        while True:
            money = random.randint(100, 500)
            gLock.acquire()
            if gMoney > money:
                gMoney -= money
                print('%s當前取出%s元錢，剩餘%s元錢' % (threading.current_thread(), money, gMoney))
                time.sleep(0.5)
            else:
                # 若是錢不夠了，有多是已經超過了次數，這時候就判斷一下
                if gTimes >= 10:
                    gLock.release()
                    break
                print("%s當前想取%s元錢，剩餘%s元錢，不足！" % (threading.current_thread(),money,gMoney))
            gLock.release()

def main():
    for x in range(5):
        Consumer(name='消費者線程%d'%x).start()

    for x in range(5):
        Producer(name='生產者線程%d'%x).start()

if __name__ == '__main__':
    main()

Condition版的生產者與消費者模式：

Lock版本的生產者與消費者模式能夠正常的運行。可是存在一個不足，在消費者中，老是經過while True死循環而且上鎖的方式去判斷錢夠不夠。上鎖是一個很耗費CPU資源的行爲。所以這種方式不是最好的。還有一種更好的方式即是使用threading.Condition來實現。threading.Condition能夠在沒有數據的時候處於阻塞等待狀態。一旦有合適的數據了，還可使用notify相關的函數來通知其餘處於等待狀態的線程。這樣就能夠不用作一些無用的上鎖和解鎖的操做。能夠提升程序的性能。首先對threading.Condition相關的函數作個介紹，threading.Condition相似threading.Lock，能夠在修改全局數據的時候進行上鎖，也能夠在修改完畢後進行解鎖。如下將一些經常使用的函數作個簡單的介紹：

acquire：上鎖。
release：解鎖。
wait：將當前線程處於等待狀態，而且會釋放鎖。能夠被其餘線程使用notify和notify_all函數喚醒。被喚醒後會繼續等待上鎖，上鎖後繼續執行下面的代碼。
notify：通知某個正在等待的線程，默認是第1個等待的線程。
notify_all：通知全部正在等待的線程。notify和notify_all不會釋放鎖。而且須要在release以前調用。

Condition版的生產者與消費者模式代碼以下：

import threading
import random
import time

gMoney = 1000
gCondition = threading.Condition()
gTimes = 0
gTotalTimes = 5

class Producer(threading.Thread):
    def run(self):
        global gMoney
        global gCondition
        global gTimes
        while True:
            money = random.randint(100, 1000)
            gCondition.acquire()
            if gTimes >= gTotalTimes:
                gCondition.release()
                print('當前生產者總共生產了%s次'%gTimes)
                break
            gMoney += money
            print('%s當前存入%s元錢，剩餘%s元錢' % (threading.current_thread(), money, gMoney))
            gTimes += 1
            time.sleep(0.5)
            gCondition.notify_all()
            gCondition.release()

class Consumer(threading.Thread):
    def run(self):
        global gMoney
        global gCondition
        while True:
            money = random.randint(100, 500)
            gCondition.acquire()
            # 這裏要給個while循環判斷，由於等輪到這個線程的時候
            # 條件有可能又不知足了
            while gMoney < money:
                if gTimes >= gTotalTimes:
                    gCondition.release()
                    return
                print('%s準備取%s元錢，剩餘%s元錢，不足！'%(threading.current_thread(),money,gMoney))
                gCondition.wait()
            gMoney -= money
            print('%s當前取出%s元錢，剩餘%s元錢' % (threading.current_thread(), money, gMoney))
            time.sleep(0.5)
            gCondition.release()

def main():
    for x in range(5):
        Consumer(name='消費者線程%d'%x).start()

    for x in range(2):
        Producer(name='生產者線程%d'%x).start()

if __name__ == '__main__':
    main()

Queue線程安全隊列：

在線程中，訪問一些全局變量，加鎖是一個常常的過程。若是你是想把一些數據存儲到某個隊列中，那麼Python內置了一個線程安全的模塊叫作queue模塊。Python中的queue模塊中提供了同步的、線程安全的隊列類，包括FIFO（先進先出）隊列Queue，LIFO（後入先出）隊列LifoQueue。這些隊列都實現了鎖原語（能夠理解爲原子操做，即要麼不作，要麼都作完），可以在多線程中直接使用。可使用隊列來實現線程間的同步。相關的函數以下：

初始化Queue(maxsize)：建立一個先進先出的隊列。
qsize()：返回隊列的大小。
empty()：判斷隊列是否爲空。
full()：判斷隊列是否滿了。
get()：從隊列中取最後一個數據。
put()：將一個數據放到隊列中。

使用生產者與消費者模式多線程下載表情包：

import threading
import requests
from lxml import etree
from urllib import request
import os
import re
from queue import Queue

class Producer(threading.Thread):
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/62.0.3202.94 Safari/537.36'
    }
    def __init__(self,page_queue,img_queue,*args,**kwargs):
        super(Producer, self).__init__(*args,**kwargs)
        self.page_queue = page_queue
        self.img_queue = img_queue


    def run(self):
        while True:
            if self.page_queue.empty():
                break
            url = self.page_queue.get()
            self.parse_page(url)

    def parse_page(self,url):
        response = requests.get(url,headers=self.headers)
        text = response.text
        html = etree.HTML(text)
        imgs = html.xpath("//div[@class='page-content text-center']//a//img")
        for img in imgs:
            if img.get('class') == 'gif':
                continue
            img_url = img.xpath(".//@data-original")[0]
            suffix = os.path.splitext(img_url)[1]
            alt = img.xpath(".//@alt")[0]
            alt = re.sub(r'[，。？?,/\\·]','',alt)
            img_name = alt + suffix
            self.img_queue.put((img_url,img_name))

class Consumer(threading.Thread):
    def __init__(self,page_queue,img_queue,*args,**kwargs):
        super(Consumer, self).__init__(*args,**kwargs)
        self.page_queue = page_queue
        self.img_queue = img_queue

    def run(self):
        while True:
            if self.img_queue.empty() and self.page_queue.empty():
            　　break
            img_url,filename = self.img_queue.get()
            request.urlretrieve(url,'images/'+filename)
            print(filename+'  下載完成！')

def main():
    page_queue = Queue(100)
    img_queue = Queue(500)
    for x in range(1,101):
        url = "http://www.doutula.com/photo/list/?page=%d" % x
        page_queue.put(url)

    for x in range(5):
        t = Producer(page_queue,img_queue)
        t.start()

    for x in range(5):
        t = Consumer(page_queue,img_queue)
        t.start()

if __name__ == '__main__':
    main()

GIL全局解釋器鎖：

Python自帶的解釋器是CPython。CPython解釋器的多線程其實是一個假的多線程（在多核CPU中，只能利用一核，不能利用多核）。同一時刻只有一個線程在執行，爲了保證同一時刻只有一個線程在執行，在CPython解釋器中有一個東西叫作GIL（Global Intepreter Lock），叫作全局解釋器鎖。這個解釋器鎖是有必要的。由於CPython解釋器的內存管理不是線程安全的。固然除了CPython解釋器，還有其餘的解釋器，有些解釋器是沒有GIL鎖的，見下面：

Jython：用Java實現的Python解釋器。不存在GIL鎖。更多詳情請見：https://zh.wikipedia.org/wiki/Jython
IronPython：用.net實現的Python解釋器。不存在GIL鎖。更多詳情請見：https://zh.wikipedia.org/wiki/IronPython
PyPy：用Python實現的Python解釋器。存在GIL鎖。更多詳情請見：https://zh.wikipedia.org/wiki/PyPy
GIL雖然是一個假的多線程。可是在處理一些IO操做（好比文件讀寫和網絡請求）仍是能夠在很大程度上提升效率的。在IO操做上建議使用多線程提升效率。在一些CPU計算操做上不建議使用多線程，而建議使用多進程。

多線程下載百思不得姐段子：

import requests
from lxml import etree
import threading
from queue import Queue
import csv


class BSSpider(threading.Thread):
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/62.0.3202.94 Safari/537.36'
    }
    def __init__(self,page_queue,joke_queue,*args,**kwargs):
        super(BSSpider, self).__init__(*args,**kwargs)
        self.base_domain = 'http://www.budejie.com'
        self.page_queue = page_queue
        self.joke_queue = joke_queue

    def run(self):
        while True:
            if self.page_queue.empty():
                break
            url = self.page_queue.get()
            response = requests.get(url, headers=self.headers)
            text = response.text
            html = etree.HTML(text)
            descs = html.xpath("//div[@class='j-r-list-c-desc']")
            for desc in descs:
                jokes = desc.xpath(".//text()")
                joke = "\n".join(jokes).strip()
                link = self.base_domain+desc.xpath(".//a/@href")[0]
                self.joke_queue.put((joke,link))
            print('='*30+"第%s頁下載完成！"%url.split('/')[-1]+"="*30)

class BSWriter(threading.Thread):
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/62.0.3202.94 Safari/537.36'
    }

    def __init__(self, joke_queue, writer,gLock, *args, **kwargs):
        super(BSWriter, self).__init__(*args, **kwargs)
        self.joke_queue = joke_queue
        self.writer = writer
        self.lock = gLock

    def run(self):
        while True:
            try:
                joke_info = self.joke_queue.get(timeout=40)
                joke,link = joke_info
                self.lock.acquire()
                self.writer.writerow((joke,link))
                self.lock.release()
                print('保存一條')
            except:
                break

def main():
    page_queue = Queue(10)
    joke_queue = Queue(500)
    gLock = threading.Lock()
    fp = open('bsbdj.csv', 'a',newline='', encoding='utf-8')
    writer = csv.writer(fp)
    writer.writerow(('content', 'link'))

    for x in range(1,11):
        url = 'http://www.budejie.com/text/%d' % x
        page_queue.put(url)

    for x in range(5):
        t = BSSpider(page_queue,joke_queue)
        t.start()

    for x in range(5):
        t = BSWriter(joke_queue,writer,gLock)
        t.start()

if __name__ == '__main__':
    main()