摘要：限制同時(shí)運(yùn)行線程數(shù)使用類就行，在內(nèi)部管理著一個(gè)計(jì)數(shù)器。當(dāng)計(jì)數(shù)器到時(shí)，再調(diào)用就會(huì)阻塞，直到其他線程來(lái)調(diào)用，這樣就限制了同時(shí)運(yùn)行線程的數(shù)量。

事前最好了解一下什么是進(jìn)程，什么是線程，什么是GIL，本文不再贅述，直接介紹模塊的使用：

推薦1，推薦2，推薦3，更多自尋

普通的python爬蟲(chóng)是單進(jìn)程單線程的，這樣在遇到大量重復(fù)的操作時(shí)就只能逐個(gè)進(jìn)行，我們就很難過(guò)了。舉個(gè)栗子：你有1000個(gè)美圖的鏈接，逐個(gè)喂給下載器（函數(shù)），看著圖片只能一個(gè)個(gè)蹦出來(lái)，你不心急嗎？于是我們想，能不能同時(shí)跑多個(gè)下載器，實(shí)現(xiàn)多圖同時(shí)下載？——答案是可以的，使用多進(jìn)程/多線程，把每個(gè)帶著不同參數(shù)下載器分給每個(gè)進(jìn)程/線程就，然后同時(shí)跑多個(gè)進(jìn)程/線程就行了。

本文就介紹如何用多線程和多進(jìn)程給爬蟲(chóng)加速

補(bǔ)充主線程與子線程（進(jìn)程同理）：

一個(gè)py程序就有一個(gè)主線程，主線程負(fù)責(zé)整個(gè)py程序的代碼，當(dāng)主線程處理到啟用多線程的代碼時(shí)，就會(huì)創(chuàng)建若干個(gè)子線程，這里就有選擇了，主線程是等待子線程的結(jié)束再繼續(xù)處理還是直接繼續(xù)處理讓子線程在外頭跑

Python標(biāo)準(zhǔn)庫(kù)原本有threading和multiprocessing模塊編寫(xiě)相應(yīng)的多線程/多進(jìn)程代碼。但從Python3.2開(kāi)始，標(biāo)準(zhǔn)庫(kù)為我們提供了concurrent.futures模塊，它提供了ThreadPoolExecutor和ProcessPoolExecutor兩個(gè)類，實(shí)現(xiàn)了對(duì)threading和multiprocessing的更高級(jí)的抽象，對(duì)編寫(xiě)線程池/進(jìn)程池提供了直接的支持。多進(jìn)程我們介紹futures的ProcessPoolExecutor
注：python 2.7 請(qǐng)安裝future模塊，pip install future

ProcessPoolExecutor類是Executor類的子類，實(shí)例化ProcessPoolExecutor類以創(chuàng)建進(jìn)程池，在實(shí)例化的過(guò)程中應(yīng)指定同時(shí)運(yùn)行的最大進(jìn)程數(shù)

from concurrent.futures import  ProcessPoolExecutor
pool = ProcessPoolExecutor(max_workers=4) # 運(yùn)行最大進(jìn)程數(shù)4
#進(jìn)程池的操作...
pool.shutdown(wait=True) # 關(guān)閉進(jìn)程池，默認(rèn)等待所有進(jìn)程的完成。
print("Deep") # 有shutdown的情況下所有進(jìn)程完成后才會(huì)運(yùn)行下面的print，沒(méi)有的話會(huì)馬上運(yùn)行

"創(chuàng)建進(jìn)程也可用with,這時(shí)會(huì)自帶shutdown功能
with ProcessPoolExecutor(4) as pool:
    #進(jìn)程池的操作...
"

該類有兩種方法對(duì)進(jìn)程池提交任務(wù)建立進(jìn)程(函數(shù)及一組參數(shù)構(gòu)成一個(gè)任務(wù))，分別是submit()和map()，如果單純想多開(kāi)進(jìn)程別無(wú)他想，用哪個(gè)都行，但submit()會(huì)有更靈活的用法

fn：函數(shù)

*iterables：函數(shù)每個(gè)參數(shù)的集合，N個(gè)參數(shù)就接N個(gè)集合

可以理解這是python自帶map()的多進(jìn)程版，他返回的是一個(gè)迭代器，包含每個(gè)任務(wù)對(duì)應(yīng)的返回值（有序的），下面用例子來(lái)分析

from concurrent.futures import  ProcessPoolExecutor
import time


def test(x):
    time.sleep(x) # 時(shí)間阻塞
    print(str(x)+"s")
    return x

if __name__ == "__main__":
    with ProcessPoolExecutor(4) as pool:
        p = pool.map(test,[2,3,10,5,6])
        for i in p:
            print(i)

輸出

2s
2
3s
3
5s
6s
10s
10
5
6

分析（下面以參數(shù)代替某個(gè)進(jìn)程）：

帶s的是函數(shù)輸出的，進(jìn)程池最大允許4個(gè)進(jìn)程同時(shí)運(yùn)行，所以參數(shù) 2，3，10，5 首先一起進(jìn)去。2最快完成，馬上讓給6進(jìn)去，2+6<10 ,所以后進(jìn)6完成得比10快，最后輸出順序就是 2s,3s,5s,6s,10s

不帶s的是for循環(huán)打印迭代器中的結(jié)果，由輸出可見(jiàn)，i的值分配是會(huì)等待進(jìn)程完成返回值的，等2的完成返回2，等3的完成返回3，等10的完成返回10，由于10完成前5和6早就完成了，所以返回10后緊接著返回5和6，最后輸出順序?yàn)?b>2,3,10,5,6,是有序的，對(duì)應(yīng)各任務(wù)的返回值

在爬蟲(chóng)中，上面代碼中的時(shí)間阻塞會(huì)對(duì)應(yīng)著網(wǎng)絡(luò)I/O阻塞，任務(wù)中往往包含著網(wǎng)絡(luò)請(qǐng)求。比如你有很多個(gè)圖片鏈接，就寫(xiě)一個(gè)下載圖片的函數(shù)（接收一個(gè)圖片鏈接的參數(shù)），把函數(shù)和圖片鏈接的集合喂給map()就實(shí)現(xiàn)多進(jìn)程了加速了。

fn：函數(shù)

*arg：函數(shù)的參數(shù)

該方法是往進(jìn)程池中提交可回調(diào)的任務(wù)，并返回一個(gè)future實(shí)例。提交多個(gè)任務(wù)可用循環(huán)實(shí)現(xiàn)，返回的future實(shí)例用列表存起來(lái)，每個(gè)future代表一個(gè)進(jìn)程。關(guān)于future對(duì)象有許多方法：

future.running()：判斷某個(gè)future（進(jìn)程）是否運(yùn)行中

future.done()：判斷某個(gè)future（進(jìn)程）是否正常結(jié)束

future.cancel()：終止某個(gè)future（進(jìn)程），終止失敗返回False，成功返回True

future.result()：獲取future對(duì)應(yīng)任務(wù)返回的結(jié)果。如果future還沒(méi)完成就會(huì)去等待

future.add_done_callback(fn)：接收函數(shù)fn，將fn綁定到future對(duì)象上。當(dāng)future對(duì)象被終止或完成時(shí)，fn將會(huì)被調(diào)用并接受該future對(duì)象

as_completed(fs)：接收f(shuō)utures列表，futures列表中一旦有某個(gè)future（進(jìn)程）完成就將該future對(duì)象yield回來(lái)，是個(gè)迭代器

from concurrent.futures import ProcessPoolExecutor,as_completed
import time

def test(x):
    time.sleep(x)
    print(str(x)+"s")
    return x

if __name__ == "__main__":
    with ProcessPoolExecutor(4) as pool:
        futures = [pool.submit(test,i) for i in [2,3,10,5,6]]

        """for j in futures:
            print(j.result()) # 對(duì)應(yīng)接收參數(shù)有序輸出，輸出2，3，10，5，6
        """
        for j in as_completed(futures):
            print(j.result()) # 對(duì)應(yīng)進(jìn)程完成順序輸出，輸出2，3，5，6，10

建議小心使用，雖然多線程能實(shí)現(xiàn)高并發(fā)，但由于線程資源共享的特性，某個(gè)線程操作這些共享的資源時(shí)可能操到一半就停止讓給另一個(gè)線程操作，導(dǎo)致錯(cuò)亂的發(fā)生。為避免此情況發(fā)生對(duì)某些操作需要加鎖，所以這里介紹對(duì)鎖有支持的threading模塊，python自帶直接導(dǎo)入。
如果你確信這些操作不會(huì)發(fā)生錯(cuò)亂，可以直接使用concurrent.future 的 ThreadPoolExecutor，方法什么的和ProcessPoolExecutor的一樣

創(chuàng)建線程有兩種方法：

實(shí)例化 threading.Thread 類，target接收函數(shù)，arg以可迭代形式接收參數(shù)。這種方法最簡(jiǎn)單

import threading
     import time
     
     def test(x):
        time.sleep(x)
        print(str(x)+"s")
        return x
        
     t1 = threading.Thread(target=test, args=(1,)) # 創(chuàng)建線程
     t2 = threading.Thread(target=test, args=(3,))
     t1.start() # 啟動(dòng)線程
     t2.start()

繼承threading.Thread 類，重寫(xiě)run方法，把函數(shù)及參數(shù)接收寫(xiě)進(jìn)自己寫(xiě)的多線程類中。這種方法更靈活，threading.Thread 類并沒(méi)有供獲取線程調(diào)用函數(shù)返回值的方法，如果需要函數(shù)返回值就需要繼承該類自己實(shí)現(xiàn)

import threading
import time

class TestThread(threading.Thread):
    def __init__(self,x):
        threading.Thread.__init__(self)
        self.x = x  # 參數(shù)接收

    def run(self):
        time.sleep(self.x)  # 原來(lái)的函數(shù)寫(xiě)到run中
        print(str(self.x)+"s")

    def result(self):  # 實(shí)現(xiàn)獲取調(diào)用函數(shù)的返回值的方法
        return self.x

t1 = TestThread(1)  #創(chuàng)建線程
t2 = TestThread(3)
t1.start()  # 啟動(dòng)線程
t2.start()
t1.join()  # 等待線程結(jié)束
t2.join()
print(t1.result(),t2.result())

線程相關(guān)方法和屬性：

Thread.start()：啟動(dòng)線程

Thread.join()：等待線程的結(jié)束，沒(méi)有join的話會(huì)接著運(yùn)行join下面的代碼

Thread.is_alive()：判斷線程是否在運(yùn)行，線程未開(kāi)啟/結(jié)束時(shí)返回 False

Thread.name：返回線程的名字，默認(rèn)線程名是Thread-N，N指第N個(gè)開(kāi)啟的線程

Thread.setName(str)：給線程命名

Thread.setDaemon(True/False)：設(shè)置子線程是否會(huì)隨主線程結(jié)束而結(jié)束，原本所有子線程默認(rèn)是不會(huì)隨主線程結(jié)束而結(jié)束的

線程間資源共享，如果多個(gè)線程共同對(duì)某個(gè)數(shù)據(jù)修改，可能會(huì)出現(xiàn)錯(cuò)誤，為了保證數(shù)據(jù)的正確性，需要對(duì)多個(gè)線程進(jìn)行同步。這時(shí)就需要引入鎖了（利用GIL），鎖只有一個(gè)，一個(gè)線程在持有鎖的狀態(tài)下對(duì)某些數(shù)據(jù)進(jìn)行操作，其他線程就無(wú)法對(duì)該數(shù)據(jù)進(jìn)行操作，直至該線程釋放鎖讓其他線程搶，誰(shuí)搶到誰(shuí)就有權(quán)修改。

threading提供Lock和RLock兩類鎖，前者一個(gè)線程只能獲取獲取一次鎖，后者允許一個(gè)線程能重復(fù)獲取鎖。如果某個(gè)線程對(duì)全局?jǐn)?shù)據(jù)的操作是割裂的（分塊的），那就使用RLock。

acquire()：獲取鎖

release()：釋放鎖

有數(shù)據(jù)操作放在acquire 和 release 之間，就不會(huì)出現(xiàn)多個(gè)線程修改同一個(gè)數(shù)據(jù)的風(fēng)險(xiǎn)了

acquire 和 release 必須成對(duì)存在，如果一個(gè)線程只拿不放，其他線程沒(méi)有鎖能搶就只能永遠(yuǎn)阻塞（停止）

一個(gè)錯(cuò)亂的例子及鎖的使用：

import time, threading

lock = threading.Lock() # rlock = threading.RLock()
balance = [0]

def test(n):
    for i in range(100000): # 理想的情況是執(zhí)行了+n，-n操作后才讓另一個(gè)線程處理，結(jié)果永0
        #lock.acquire()
        balance[0] = balance[0] + n  # 某個(gè)線程可能處理到這里就終止讓給另一個(gè)線程處理了，循環(huán)一大，結(jié)果可能錯(cuò)亂不為0
        balance[0] = balance[0] - n
        #lock.release()
t1 = threading.Thread(target=test, args=(5,))
t2 = threading.Thread(target=test, args=(8.0,))
t1.start()
t2.start()
t1.join()
t2.join()
print(balance[0])

在不加鎖的情況下多跑幾次，你會(huì)的到不同的結(jié)果。但是加了鎖之后，+n,-n兩個(gè)操作完整執(zhí)行，不會(huì)中途中斷，結(jié)果永0。

使用 threading.Semaphore 類就行，Semaphore 在內(nèi)部管理著一個(gè)計(jì)數(shù)器。調(diào)用 acquire() 會(huì)使這個(gè)計(jì)數(shù)器減1，release() 則是加1。計(jì)數(shù)器的值永遠(yuǎn)不會(huì)小于 0。當(dāng)計(jì)數(shù)器到 0 時(shí)，再調(diào)用 acquire() 就會(huì)阻塞，直到其他線程來(lái)調(diào)用release()，這樣就限制了同時(shí)運(yùn)行線程的數(shù)量。

使用上非常簡(jiǎn)單，實(shí)例化Semaphore并指定線程數(shù)后，給函數(shù)的頭加個(gè)acquire()，尾加個(gè)release()就行。

import threading, time

def test(x):
    semaphore.acquire()
    time.sleep(x)
    print(x)
    semaphore.release()
    
semaphore = threading.Semaphore(4) # 最大4個(gè)線程同時(shí)進(jìn)行
ts = [threading.Thread(target=test,args=(i,)) for i in [2,3,5,10,6]]
[t.start() for t in ts]

"輸出：2，3，5，6，10 
（原理和上面多進(jìn)程的那個(gè)差不多）"

關(guān)于threading的其他高級(jí)用法本文并未提及，以上都是些常用的用法，如果有更高級(jí)的需要，可以參考這文章

講了這么多，都是模塊的用法，沒(méi)怎么提到爬蟲(chóng)。那么最后大概的講下如何把多進(jìn)程/多線程運(yùn)用到爬蟲(chóng)中，并給個(gè)代碼實(shí)例用作參考。

如果爬蟲(chóng)需要重復(fù)進(jìn)行某個(gè)操作（如下載一張圖片，爬取一張網(wǎng)頁(yè)的源碼，破解一次加密【加密耗cpu最好多進(jìn)程】），那把這個(gè)操作抽象成一個(gè)接收相應(yīng)參數(shù)的函數(shù)，把函數(shù)喂給進(jìn)程/線程即可。

沒(méi)了，大概就這么用?? ?

下面給個(gè)多進(jìn)程/多線程結(jié)合的網(wǎng)易云音樂(lè)評(píng)論下載器（下載某首音樂(lè)的多頁(yè)評(píng)論），包含加密算法，如不清楚可看之前的文章，我們用多進(jìn)程加速加密過(guò)程，用多線程加速爬取過(guò)程。
本代碼較長(zhǎng)，長(zhǎng)到高亮效果都沒(méi)有了，因此該長(zhǎng)代碼分為兩部分，前半部分是之前文章提到的加密方法，后半部分是本文的多進(jìn)程多線程重點(diǎn)代碼：

import json, re, base64, random, requests, binascii, threading
from Crypto.Cipher import AES#新的加密模塊只接受bytes數(shù)據(jù)，否者報(bào)錯(cuò)，密匙明文什么的要先轉(zhuǎn)碼
from concurrent.futures import ProcessPoolExecutor
from math import ceil

secret_key = b"0CoJUm6Qyw8W8jud"#第四參數(shù)，aes密匙
pub_key ="010001"#第二參數(shù)，rsa公匙組成
modulus = "00e0b509f6259df8642dbc35662901477df22677ec152b5ff68ace615bb7b725152b3ab17a876aea8a5aa76d2e417629ec4ee341f56135fccf695280104e0312ecbda92557c93870114af6c9d05c4f7f0c3685b7a46bee255932575cce10b424d813cfe4875d3e82047b97ddef52741d546b8e289dc6935b3ece0462db0a22b8e7"#第三參數(shù)，rsa公匙組成
headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/65.0.3325.146 Safari/537.36"}

def random_16():
    return bytes("".join(random.sample("1234567890DeepDarkFantasy",16)),"utf-8")

#aes加密
def aes_encrypt(text,key):
    pad = 16 - len(text)%16#對(duì)長(zhǎng)度不是16倍數(shù)的字符串進(jìn)行補(bǔ)全，然后在轉(zhuǎn)為bytes數(shù)據(jù)
    try:                    #如果接到bytes數(shù)據(jù)（如第一次aes加密得到的密文）要解碼再進(jìn)行補(bǔ)全
        text = text.decode()
    except:
        pass
    text = text + pad * chr(pad)
    try:
        text = text.encode()
    except:
        pass
    encryptor = AES.new(key,AES.MODE_CBC,b"0102030405060708")
    ciphertext = encryptor.encrypt(text)
    ciphertext = base64.b64encode(ciphertext)#得到的密文還要進(jìn)行base64編碼
    return ciphertext

#rsa加密
def rsa_encrypt(ran_16,pub_key,modulus):
    text = ran_16[::-1]#明文處理，反序并hex編碼
    rsa = int(binascii.hexlify(text), 16) ** int(pub_key, 16) % int(modulus, 16)
    return format(rsa, "x").zfill(256)

#返回加密后內(nèi)容
def encrypt_data(data):
    ran_16 = random_16()
    text = json.dumps(data)
    params = aes_encrypt(text,secret_key)
    params = aes_encrypt(params,ran_16)
    encSecKey = rsa_encrypt(ran_16,pub_key,modulus)
    return  {"params":params.decode(),
             "encSecKey":encSecKey  }

class OnePageComment(threading.Thread):  # 下載一頁(yè)評(píng)論的線程類
    def __init__(self,post_url, enc_data):
        threading.Thread.__init__(self)
        self.post_url = post_url
        self.enc_data = enc_data
        self.comment = "" # 創(chuàng)建一個(gè)comment變量?jī)?chǔ)存爬到的數(shù)據(jù)

    def run(self):
        semaphore.acquire()
        content = requests.post(self.post_url, headers = headers, data = self.enc_data ).json()
        if "hotComments" in content:
            if content["hotComments"]:
                self.comment +=  "*************精彩評(píng)論

"
                self.common(content, "hotComments")

            self.comment += "

*************最新評(píng)論

"
            self.common(content, "comments")
        else:
            self.common(content, "comments")
        semaphore.release()

    def common(self, content,c_type):
        for each in content[c_type]:
            if each ["beReplied"]:
                if each["beReplied"][0]["content"]:
                    self.comment += each["content"] + "
	回復(fù)：
	" + each["beReplied"][0]["content"] + "
" + "-" * 60 + "
"
            else:
                self.comment += each["content"] + "
" + "-" * 60 + "
"

    def get_comment(self):  # 選擇返回評(píng)論而不是直接寫(xiě)入文件，因?yàn)槎鄠€(gè)線程同時(shí)操作一個(gè)文件有風(fēng)險(xiǎn)，應(yīng)先返回，后統(tǒng)一寫(xiě)入
        return self.comment


def get_enc_datas(pages, max_workers=4): # 多進(jìn)程加密
    raw_datas = []
    for i in range(pages):
        if i == 0:
            raw_datas.append({"rid":"", "offset":"0", "total":"true", "limit":"20", "csrf_token":""})
        else:
            raw_datas.append({"rid":"", "offset":str(i*20), "total":"false", "limit":"20", "csrf_token":""})
    with ProcessPoolExecutor(max_workers) as pool:  # 多進(jìn)程適合計(jì)算密集型任務(wù)，如加密
        result = pool.map(encrypt_data,raw_datas)
    return list(result)

def one_song_comment(id_): # 爬取一首歌的評(píng)論并寫(xiě)入txt，網(wǎng)絡(luò)I/O密集使用多線程
    post_url = "http://music.163.com/weapi/v1/resource/comments/R_SO_4_" + str(id_) + "?csrf_token="
    ts = [OnePageComment(post_url,i) for i in enc_datas]
    [i.start() for i in ts]
    [i.join() for i in ts]
    comments = [i.get_comment() for i in ts]
    with open(id_ + ".txt", "w", encoding="utf-8") as f:
        f.writelines(comments)


if __name__ == "__main__":
    semaphore = threading.Semaphore(4) # 最大線程4
    enc_datas = get_enc_datas(10)  # 獲取加密后的數(shù)據(jù),對(duì)所有歌曲都是通用的，這里有十頁(yè)的加密數(shù)據(jù)，對(duì)應(yīng)爬十頁(yè)評(píng)論
    one_song_comment("29498682")

效果提升驚人！！不信你跑一下上面的程序，然后和自己寫(xiě)的單線程/單進(jìn)程比較

cpu和網(wǎng)絡(luò)都跑到了峰值，網(wǎng)絡(luò)峰值在cpu峰值之后，因?yàn)槭窍榷噙M(jìn)程加密數(shù)據(jù)，后多線程爬取