人人都能學(xué)會(huì)的 Python 多線程指南！

大家好，我是老表。

在 Python 中，多線程最常見的一個(gè)場(chǎng)景就是爬蟲，例如這樣一個(gè)需求，有多個(gè)結(jié)構(gòu)一樣的頁面需要爬取，例如下方的URL（豆瓣阿凡達(dá)影評(píng)，以10個(gè)為例）

  url_list = [
      'https://movie.douban.com/subject/1652587/reviews?sort=time&start=0',
      'https://movie.douban.com/subject/1652587/reviews?sort=time&start=20',
      'https://movie.douban.com/subject/1652587/reviews?sort=time&start=40',
      'https://movie.douban.com/subject/1652587/reviews?sort=time&start=60',
      'https://movie.douban.com/subject/1652587/reviews?sort=time&start=80',
      'https://movie.douban.com/subject/1652587/reviews?sort=time&start=100',
      'https://movie.douban.com/subject/1652587/reviews?sort=time&start=120',
      'https://movie.douban.com/subject/1652587/reviews?sort=time&start=140',
      'https://movie.douban.com/subject/1652587/reviews?sort=time&start=160',
      'https://movie.douban.com/subject/1652587/reviews?sort=time&start=180']

如果依次爬取，請(qǐng)求第一個(gè)頁面——得到返回?cái)?shù)據(jù)——解析數(shù)據(jù)——提取、存儲(chǔ)數(shù)據(jù)——請(qǐng)求第二個(gè)頁面，按照這樣的思路，那么大量時(shí)間都會(huì)浪費(fèi)在請(qǐng)求、返回?cái)?shù)據(jù)上，如果在等待第一個(gè)頁面返回?cái)?shù)據(jù)時(shí)去請(qǐng)求第二個(gè)頁面，就能有效的提高效率，多線程就可以實(shí)現(xiàn)這樣的功能。

在Python中實(shí)現(xiàn)多線程的方法也很多，我將基于 threading 模塊一點(diǎn)一點(diǎn)介紹，注意本文不會(huì)太注重于多線程背后的技術(shù)概念（面試常問），僅希望用最少的話教會(huì)大家如何實(shí)現(xiàn)。當(dāng)然會(huì)在最后介紹如何使用threading模塊來解決上面的爬蟲問題。

首先需要你的電腦安裝好了Python環(huán)境，并且安裝好了Python開發(fā)工具。

如果你還沒有安裝，可以參考以下文章：

如果僅用Python來處理數(shù)據(jù)、爬蟲、數(shù)據(jù)分析或者自動(dòng)化腳本、機(jī)器學(xué)習(xí)等，建議使用Python基礎(chǔ)環(huán)境+jupyter即可，安裝使用參考Windows/Mac 安裝、使用Python環(huán)境+jupyter notebook

如果想利用Python進(jìn)行web項(xiàng)目開發(fā)等，建議使用Python基礎(chǔ)環(huán)境+Pycharm，安裝使用參考：Windows下安裝、使用Pycharm教程，這下全了 和 Mac下玩轉(zhuǎn)Python-安裝&使用Python/PyCharm 。

threading基本使用

讓我們先從一個(gè)簡單的例子開始，定義do_something函數(shù)，執(zhí)行該函數(shù)需要消耗1秒

import time

start = time.perf_counter()

def do_something():
    print("-> 線程啟動(dòng)")
    time.sleep(1)
    print("-> 線程結(jié)束")

do_something()

finish = time.perf_counter()

print(f"全部任務(wù)執(zhí)行完成，耗時(shí) {round(finish - start,2)} 秒")

上面的代碼不難理解，執(zhí)行do_something并計(jì)算耗時(shí)，結(jié)果很明顯應(yīng)該是1s

-> 線程啟動(dòng)
-> 線程結(jié)束
全部任務(wù)執(zhí)行完成，耗時(shí) 1.01 秒

現(xiàn)在如果需要執(zhí)行兩次do_something，按照最基本的思路

import time

start = time.perf_counter()


def do_something():
    print("-> 線程啟動(dòng)")
    time.sleep(1)
    print("-> 線程結(jié)束")


do_something()
do_something()

finish = time.perf_counter()

print(f"全部任務(wù)執(zhí)行完成，耗時(shí) {round(finish - start,2)} 秒")

執(zhí)行上面代碼結(jié)果也很容易猜到是2秒

-> 線程啟動(dòng)
-> 線程結(jié)束
-> 線程啟動(dòng)
-> 線程結(jié)束
全部任務(wù)執(zhí)行完成，耗時(shí) 2.01 秒

這就是最常規(guī)的 同步 思路，在CPU執(zhí)行第一個(gè)函數(shù)，也就是等待1s的時(shí)間內(nèi)，什么也不干，等第一個(gè)函數(shù)執(zhí)行完畢后再執(zhí)行第二個(gè)函數(shù)

很明顯，這樣讓CPU干等著啥也不干并不是一個(gè)很好的選擇，而多線程就是解決這一問題的方法之一，讓CPU在等待某個(gè)任務(wù)完成時(shí)去執(zhí)行更多的操作，將整個(gè)過程簡化為下圖流程，這樣就能充分節(jié)省時(shí)間

現(xiàn)在使用threading來通過多線程的方式實(shí)現(xiàn)上面的過程，非常簡單，定義兩個(gè)線程并依次啟動(dòng)即可??

import time
import threading

start = time.perf_counter()


def do_something():
    print("-> 線程啟動(dòng)")
    time.sleep(1)
    print("-> 線程結(jié)束")


thread1 = threading.Thread(target=do_something)
thread2 = threading.Thread(target=do_something)

thread1.start()
thread2.start()

finish = time.perf_counter()

print(f"全部任務(wù)執(zhí)行完成，耗時(shí) {round(finish - start,2)} 秒")

執(zhí)行上面的代碼，結(jié)果如下

-> 線程啟動(dòng)
-> 線程啟動(dòng)
全部任務(wù)執(zhí)行完成，耗時(shí) 0.0 秒
-> 線程結(jié)束
-> 線程結(jié)束

可以看到，兩個(gè)子線程確實(shí)同時(shí)啟動(dòng)，但是主線程并未等待兩個(gè)子線程執(zhí)行完畢就直接結(jié)束。

為了解決這個(gè)問題，我們可以使用threading.join()方法，意思是在子線程完成運(yùn)行之前，這個(gè)子線程的父線程將一直被阻塞

換成人話就是讓主線程掛起，等待所有子線程結(jié)束再執(zhí)行，體現(xiàn)到代碼上也很簡單，只需要添加兩行即可

import time
import threading

start = time.perf_counter()


def do_something():
    print("-> 線程啟動(dòng)")
    time.sleep(1)
    print("-> 線程結(jié)束")


thread1 = threading.Thread(target=do_something)
thread2 = threading.Thread(target=do_something)

thread1.start() 
thread2.start()

thread1.join()
thread2.join()

finish = time.perf_counter()

print(f"全部任務(wù)執(zhí)行完成，耗時(shí) {round(finish - start,2)} 秒")

運(yùn)行結(jié)果如下，全部代碼在1秒內(nèi)運(yùn)行完畢

-> 線程啟動(dòng)
-> 線程啟動(dòng)
-> 線程結(jié)束
-> 線程結(jié)束
全部任務(wù)執(zhí)行完成，耗時(shí) 1.01 秒

至此，我們就得到了第一個(gè)有效的多線程代碼，相信你也能大致明白threading的基本使用流程。

傳遞參數(shù)

現(xiàn)在來看看如何在多線程之間傳遞參數(shù)，讓我們升級(jí)代碼：do_something函數(shù)來接受一個(gè)參數(shù)，控制他睡眠等待的時(shí)間

def do_something(num):
    print(f"-> 線程{num} 啟動(dòng)，睡眠 {num} 秒")
    time.sleep(num)
    print(f"-> 線程{num} 結(jié)束")

在 threading 中，創(chuàng)建線程時(shí)可以使用 args 來傳遞參數(shù)，例如現(xiàn)在接收一個(gè)參數(shù)，則上一小節(jié)的代碼可以如下修改

import time
import threading

start = time.perf_counter()

def do_something(num):
    print(f"-> 線程{num} 啟動(dòng)，睡眠 {num} 秒")
    time.sleep(num)
    print(f"-> 線程{num} 結(jié)束")

thread1 = threading.Thread(target=do_something,args = [1])
thread2 = threading.Thread(target=do_something,args = [2])

thread1.start()
thread2.start()

thread1.join()
thread2.join()

finish = time.perf_counter()

print(f"全部任務(wù)執(zhí)行完成，耗時(shí) {round(finish - start,2)} 秒")

這段代碼中，我分別讓兩個(gè)線程等待1、2秒，運(yùn)行結(jié)果顯然應(yīng)該是2秒

-> 線程1 啟動(dòng)，睡眠 1 秒
-> 線程2 啟動(dòng)，睡眠 2 秒
-> 線程1 結(jié)束
-> 線程2 結(jié)束
全部任務(wù)執(zhí)行完成，耗時(shí) 2.01 秒

如果你的線程函數(shù)需要更多的參數(shù)，只需要依次向args中追加即可。

簡化代碼

上面的案例中，我們僅開啟了兩個(gè)線程，如果是更多個(gè)線程的話，再依次重復(fù)定義、啟動(dòng)就會(huì)顯得十分繁瑣，此時(shí)我們可以使用循環(huán)來處理。

例如開啟10個(gè)線程，依次睡眠1-10秒，可以先創(chuàng)建一個(gè) list 用于存儲(chǔ)每個(gè)線程，接著利用循環(huán)依次創(chuàng)建線程，啟動(dòng)后追加到剛剛創(chuàng)建的 list 中，之后再依次等待每個(gè)線程執(zhí)行完畢，代碼如下

import time
import threading

start = time.perf_counter()

def do_something(num):
    print(f"-> 線程{num} 啟動(dòng)，睡眠 {num} 秒")
    time.sleep(num)
    print(f"-> 線程{num} 結(jié)束")

thread_list = []

for i in range(1,11):

    thread = threading.Thread(target=do_something, args=[i])
    thread.start()
    thread_list.append(thread)

for t in thread_list:

    t.join()

finish = time.perf_counter()

print(f"全部任務(wù)執(zhí)行完成，耗時(shí) {round(finish - start,2)} 秒")

結(jié)果是顯然的，雖然我們執(zhí)行了十次do_something，每次用時(shí)1-10秒，但總耗時(shí)應(yīng)該為10秒

-> 線程1 啟動(dòng)，睡眠 1 秒
-> 線程2 啟動(dòng)，睡眠 2 秒
-> 線程3 啟動(dòng)，睡眠 3 秒
-> 線程4 啟動(dòng)，睡眠 4 秒
-> 線程5 啟動(dòng)，睡眠 5 秒
-> 線程6 啟動(dòng)，睡眠 6 秒
-> 線程7 啟動(dòng)，睡眠 7 秒
-> 線程8 啟動(dòng)，睡眠 8 秒
-> 線程9 啟動(dòng)，睡眠 9 秒
-> 線程10 啟動(dòng)，睡眠 10 秒
-> 線程1 結(jié)束
-> 線程2 結(jié)束
-> 線程3 結(jié)束
-> 線程4 結(jié)束
-> 線程5 結(jié)束
-> 線程6 結(jié)束
-> 線程7 結(jié)束
-> 線程8 結(jié)束
-> 線程9 結(jié)束
-> 線程10 結(jié)束
全部任務(wù)執(zhí)行完成，耗時(shí) 10.01 秒

共享變量鎖的問題

現(xiàn)在，你應(yīng)該已經(jīng)了解 threading 最基本的用法，只需要將 do_somthing 函數(shù)進(jìn)行修改即可，但是如果你深入使用，還會(huì)有其他的問題出現(xiàn)，例如共享變量的問題，讓我們繼續(xù)探討。

多線程很常見的一個(gè)應(yīng)用就是爬蟲，回到開頭的爬蟲問題，如果我們希望爬取10個(gè)網(wǎng)頁的評(píng)論，可能會(huì)先定一個(gè)空dataframe，然后使用多線程都往這個(gè)dataframe中寫入數(shù)據(jù)，但由于多個(gè)線程同時(shí)操作這一個(gè)變量，可能會(huì)導(dǎo)致評(píng)論并不是按照順序?qū)懭氲摹?/p>

例如第一個(gè)頁面有10條評(píng)論，第一個(gè)線程寫入了2條后，第二個(gè)線程將第二個(gè)頁面的前兩條寫入，最終導(dǎo)致十個(gè)頁面的評(píng)論是亂序存儲(chǔ)！

讓我們把這個(gè)問題抽象出來，還是之前的代碼，稍微修改一下

我們先定義了一個(gè)空list，線程函數(shù)會(huì)將傳入的數(shù)字添加到該list中，在未加鎖的情況下，由于線程競(jìng)爭(zhēng)，雖然我們線程是按照順序開啟，但是最終數(shù)字并不是按照順序?qū)懭搿?/p>

有沒有辦法解決呢？當(dāng)然有，很自然的想法就是當(dāng)?shù)谝粋€(gè)線程操作該變量時(shí)，其他線程等著，寫完了再釋放，這就是鎖！

先看代碼

在上面的代碼中，我們使用 threding.Lock 創(chuàng)建了一個(gè)線程鎖，之后在線程函數(shù)操作 result 前，首先使用 lock.acquire() 加上鎖，之后操作 results ，在修改完后使用 lock.relese() 釋放，此時(shí)其他線程若想操作 results 則會(huì)阻塞，等該線程釋放后才能拿走操作中，這樣我們就保證了線程是“安全的”！

最基本的線程鎖用法就如上面代碼所示，定義鎖 --> 上鎖 --> 解鎖，但是一定要注意，lock.acquire() 和 lock.relese()，如果加了鎖但是沒有釋放，后面的線程將會(huì)全部阻塞！

限制線程數(shù)量

最后還有一個(gè)常見的問題，上面我們需要執(zhí)行幾次線程函數(shù)就開了幾個(gè)線程，但是如果需要爬成千上萬個(gè)網(wǎng)頁，開這么多線程cpu一定不同意，代碼也會(huì)在開啟的線程達(dá)到一定數(shù)量后報(bào)錯(cuò)。

所以如何讓程序只啟動(dòng)我們指定的線程數(shù)量，例如一次開五個(gè)線程，結(jié)束一個(gè)再添加一個(gè)，直到全部任務(wù)完成？

還是鎖！在 threading 模塊中有一個(gè) BoundedSemaphore（信號(hào)量）類，我們可以給他一個(gè)初始的信號(hào)量（最大線程數(shù)），之后每次有線程獲得信號(hào)量的時(shí)候（即 acquire() ）計(jì)數(shù)器－1，釋放信號(hào)量時(shí)候(release())計(jì)數(shù)器＋1，計(jì)數(shù)器為0的時(shí)候其它線程就被阻塞無法獲得信號(hào)量。當(dāng)計(jì)數(shù)器為設(shè)定好的上限的時(shí)候 BoundedSemaphore 就無法進(jìn)行 release() 操作了。

體現(xiàn)到代碼上則比較簡單，還是基于上面的例子修改

總共需要運(yùn)行十次，我們定義最大線程數(shù)為3，并在線程啟動(dòng)前調(diào)用acquire方法增加一個(gè)計(jì)數(shù)，在線程最后釋放。

此時(shí)程序一次只能啟動(dòng)三個(gè)線程，如圖中所示，首先啟動(dòng)123，之后完成123，啟動(dòng)456，當(dāng)?shù)谒膫€(gè)線程結(jié)束啟動(dòng)第七個(gè)線程······直到全部線程結(jié)束。

這里我們同時(shí)使用了上一節(jié)說的線程鎖來保護(hù)變量，用 BoundedSemaphore 鎖來控制最大線程數(shù)，在實(shí)際寫代碼時(shí)就需要小心檢查鎖是否正確釋放，否則就會(huì)報(bào)錯(cuò)！

一個(gè)真實(shí)的多線程爬蟲案例

至此，threading 模塊最常見的用法就介紹完畢，現(xiàn)在讓我們回到本文一開始的問題，有多個(gè)（以十個(gè)為例）URL需要爬取，既然每個(gè)頁面需要執(zhí)行的操作一樣，如果等待一個(gè)頁面爬取完畢再爬第二頁面就太浪費(fèi)時(shí)間了。這時(shí)就可以仿照上面的思路去使用多線程加速。

我們只需要將上面的do_something函數(shù)修改為對(duì)也面的爬取操作，之后的創(chuàng)建啟動(dòng)線程操作不變即可，代碼如下

import time
import threading
import requests
import pandas as pd
from faker import Faker
from bs4 import BeautifulSoup


def craw_url(url):
    global df
    fake = Faker()
    headers = {'User-Agent': fake.user_agent()}
    r = requests.get(url, headers=headers)
    soup = BeautifulSoup(r.content, 'html.parser')
    review_list = soup.find_all(class_="main review-item")

    for i in range(len(review_list)):

        rank = review_list[i].select('span')[0].get('title')
        time1 = review_list[i].select('span')[1].get('content')
        title = review_list[i].select('h2>a')[0].text
        df = df.append({'時(shí)間': time1,
                        '評(píng)分': rank,
                        '標(biāo)題': title, }, ignore_index=True)

    print("-> 爬取完成")


if __name__ == '__main__':

    start = time.perf_counter()
    df = pd.DataFrame(columns=['時(shí)間', '評(píng)分', '標(biāo)題'])

    url_list = [
        'https://movie.douban.com/subject/1652587/reviews?sort=time&start=0',
        'https://movie.douban.com/subject/1652587/reviews?sort=time&start=20',
        'https://movie.douban.com/subject/1652587/reviews?sort=time&start=40',
        'https://movie.douban.com/subject/1652587/reviews?sort=time&start=60',
        'https://movie.douban.com/subject/1652587/reviews?sort=time&start=80',
        'https://movie.douban.com/subject/1652587/reviews?sort=time&start=100',
        'https://movie.douban.com/subject/1652587/reviews?sort=time&start=120',
        'https://movie.douban.com/subject/1652587/reviews?sort=time&start=140',
        'https://movie.douban.com/subject/1652587/reviews?sort=time&start=160',
        'https://movie.douban.com/subject/1652587/reviews?sort=time&start=180']
    thread_list = []
    for i in url_list:

        thread = threading.Thread(target=craw_url, args=[i])
        thread.start()
        thread_list.append(thread)

    for t in thread_list:

        t.join()

    finish = time.perf_counter()
    
    print(f"全部任務(wù)執(zhí)行完成，耗時(shí) {round(finish - start,2)} 秒")

執(zhí)行這段代碼，差不多僅用了1秒就將全部內(nèi)容爬取并存儲(chǔ)到 dataframe 中，比同步的代碼塊了近十倍！如果感興趣的話可以自己嘗試一下。

至此，有關(guān) Python 多線程模塊 threading 的基本用法與需要注意的幾點(diǎn)就介紹完畢，如果全部認(rèn)真看完的話，我相信你一定能照貓畫虎寫出第一個(gè)多線程爬蟲腳本。

當(dāng)然有關(guān) Python 多線程還有很多飽受詬病的爭(zhēng)議（GIL），多線程的實(shí)現(xiàn)方法也遠(yuǎn)不止 threading 模塊，例如更常見的寫法是通過 concurrent.futures 模塊以及多進(jìn)程、協(xié)程，這些都留在本系列后續(xù)文章中再進(jìn)一步討論！

點(diǎn)贊+留言+轉(zhuǎn)發(fā)，就是對(duì)我最大的支持啦～