threading 模塊的 Thread 類(lèi)的使用

1. 多線程的基本概念

程序要完成兩個(gè)任務(wù)：

• 任務(wù) 1 進(jìn)行一項(xiàng)復(fù)雜的計(jì)算，需要 1 秒才能完成。
• 任務(wù) 2 讀取磁盤(pán)，需要 1 秒才能完成。

我們可以串行的執(zhí)行這兩項(xiàng)任務(wù)，先執(zhí)行任務(wù) 1，再執(zhí)行任務(wù) 2，完成這兩項(xiàng)任務(wù)總共需要 2 秒，如下圖所示：

我們可以并行的執(zhí)行這兩項(xiàng)任務(wù)，同時(shí)執(zhí)行這兩項(xiàng)任務(wù)，完成這兩項(xiàng)任務(wù)只需要 1 秒，如下圖所示：

顯然，并行執(zhí)行的時(shí)間小于串行執(zhí)行的時(shí)間。很多場(chǎng)景下，我們希望程序能夠同時(shí)執(zhí)行多個(gè)任務(wù)，操作系統(tǒng)提供了多線程的機(jī)制用于實(shí)現(xiàn)并行執(zhí)行多個(gè)任務(wù)。在操作系統(tǒng)中，線程是一個(gè)可以獨(dú)立執(zhí)行的任務(wù)。程序執(zhí)行時(shí)至少包含一個(gè)線程，可以使用線程相關(guān)的 API 創(chuàng)建新的線程。

Python 的 threading 模塊提供了類(lèi) Thread，用戶通過(guò)新建一個(gè)類(lèi) Thread 創(chuàng)建新的線程，本文描述了類(lèi) Thread 的基本使用。

2. 多線程的基本使用

Python 的 threading 模塊中提供了類(lèi) Thread 用于實(shí)現(xiàn)多線程，用戶有兩種使用多線程的方式:

• 在線程構(gòu)造函數(shù)中指定線程的入口函數(shù)。
• 自定義一個(gè)類(lèi)，該類(lèi)繼承類(lèi) Thread，在自定義的類(lèi)中實(shí)現(xiàn) run 方法。

2.1 線程的構(gòu)造函數(shù)和重要的成員方法

本節(jié)介紹 Thread 相關(guān)的三個(gè)函數(shù)的功能：

• 類(lèi) Thread 的構(gòu)造函數(shù)
• 類(lèi) Thread 的 start 方法
• 類(lèi) Thread 的 join 方法

2.1.1 類(lèi)Thread的構(gòu)造函數(shù)

Thread(group = None, target = None, name = None, args = (), kwargs = {}) 

參數(shù)的含義如下：

• group: 線程組，目前還沒(méi)有實(shí)現(xiàn)，在此處必須是 None。
• target: 線程的入口函數(shù)，線程從該函數(shù)開(kāi)始執(zhí)行。
• name: 線程名。
• args: 線程的入口函數(shù)的參數(shù)，以元組的形式傳入。
• kwargs: 線程的入口函數(shù)的參數(shù)，以字典的形式傳入。

使用 Thread 構(gòu)造一個(gè)新線程時(shí)，必須指定 target 和 args 兩個(gè)參數(shù)，target 為線程的入口，args 為線程入口函數(shù)的參數(shù)。

2.1.2 類(lèi) Thread 的 start 方法

start()

在線程對(duì)象的構(gòu)造函數(shù)中 target 指定了線程入口函數(shù)，args 指定了線程入口函數(shù)的參數(shù)。線程對(duì)象的 start 方法使新線程開(kāi)始執(zhí)行，執(zhí)行函數(shù) target(args)。

2.1.3 類(lèi) Thread 的 join 方法

join()

調(diào)用線程對(duì)象的 start 方法后，新線程開(kāi)始執(zhí)行函數(shù) target(args)。調(diào)用線程對(duì)象的 join 方法，主線程阻塞，等待新線程執(zhí)行完畢。

2.2 指定線程的入口函數(shù)

下面通過(guò)一個(gè)具體的例子，說(shuō)明通過(guò)指定線程的入口函數(shù)的方式使用多線程。

import time
import threading

def thread_entry(begin, end):
    for i in range(begin, end):
        time.sleep(1)
        print(i)

t0 = threading.Thread(target = thread_entry, args = (1, 4))
t1 = threading.Thread(target = thread_entry, args = (101, 104))
t0.start()
t1.start()
t0.join()
t1.join()

• 在第 9 行和第 10 行，通過(guò)調(diào)用 Thread 的構(gòu)造函數(shù)創(chuàng)建了兩個(gè)線程。

• 在第 9 行，設(shè)定線程的入口函數(shù)為 thread_entry，傳遞給入口函數(shù)兩個(gè)參數(shù)：1 和 4，新的線程將執(zhí)行 thread_entry(1, 4)，變量 t0 指向新創(chuàng)建的線程對(duì)象。

• 在第 10 行，設(shè)定線程的入口函數(shù)為 thread_entry，傳遞給入口函數(shù)兩個(gè)參數(shù)：101 和 104，新的線程將執(zhí)行 thread_entry(101, 104)，變量 t1 指向新創(chuàng)建的線程對(duì)象。

• 在第 4 行到第 7 行，定義了線程入口函數(shù)，該函數(shù)的功能是打印在 [begin, end) 區(qū)間的整數(shù)，每打印一個(gè)整數(shù)，調(diào)用 time.sleep(1) 睡眠 1 秒鐘。

• 在第 11 行，調(diào)用 start 方法啟動(dòng)線程 t0，t0 開(kāi)始執(zhí)行 thread_entry(1, 4)。

• 在第 12 行，調(diào)用 start 方法啟動(dòng)線程 t1，t1 開(kāi)始執(zhí)行 thread_entry(101, 104)。

• 在第 13 行和第 14 行，調(diào)用 join 方法，等待線程 t0 和 t1 執(zhí)行完畢。

程序的運(yùn)行結(jié)果如下：

1
101
2
102
3
103

線程 t0 的輸出結(jié)果為 1、2、3，線程 t1 的輸出結(jié)果為 101、102、103。由于兩者是并發(fā)執(zhí)行的，所以結(jié)果交織在一起。

2.3 繼承 Thread

下面通過(guò)一個(gè)具體的例子，說(shuō)明通過(guò)繼承 Thread 的方式使用多線程。

import time
import threading

class MyThread(threading.Thread):
    def __init__(self, begin, end):
        threading.Thread.__init__(self)
        self.begin = begin
        self.end = end

    def run(self):
        for i in range(self.begin, self.end):
            time.sleep(1)
            print(i)

t0 = MyThread(1, 4)
t1 = MyThread(101, 104)
t0.start()
t1.start()
t0.join()
t1.join()

• 在第 4 行，定義類(lèi) MyThread，繼承 threading.Thread。

• 在第 5 行，定義了構(gòu)造函數(shù) __init__，首先調(diào)用父類(lèi) thread.Thread.__init__ 初始化 Thread 對(duì)象，然后將參數(shù) begin 和 end 保存在 MyThread 的成員變量中。

• 在第 10 行，定義了方法 run，當(dāng)線程開(kāi)始運(yùn)行時(shí)，run 方法會(huì)被調(diào)用。在 run 方法中，打印在 [begin, end) 區(qū)間的整數(shù)，每打印一個(gè)整數(shù)，調(diào)用 time.sleep(1) 睡眠 1 秒鐘。

• 在第 15 行和第 16 行，通過(guò)調(diào)用 MyThread 的構(gòu)造函數(shù)創(chuàng)建了兩個(gè)線程。

• 在第 17 行，調(diào)用 start 方法啟動(dòng)線程 t0，t0 開(kāi)始執(zhí)行 MyThread 的方法 run()。

• 在第 18 行，調(diào)用 start 方法啟動(dòng)線程 t1，t1 開(kāi)始執(zhí)行 MyThread 的方法 run()。

• 在第 19 行和第 20 行，調(diào)用 join 方法，等待線程 t0 和 t1 執(zhí)行完畢。

程序的運(yùn)行結(jié)果如下：

1
101
2
102
3
103

線程 t0 執(zhí)行 thread_entry(1, 4)，輸出結(jié)果為 1、2、3，線程 t1 執(zhí)行 thread_entry(101, 104)，輸出結(jié)果為 101、102、103。由于兩者是并發(fā)執(zhí)行的，所以結(jié)果交織在一起。

2.4 常見(jiàn)的錯(cuò)誤

2.4.1 自定義的類(lèi)的 __init__ 方法忘記調(diào)用父類(lèi) Thread 的 __init__ 方法

通過(guò)自定義類(lèi)繼承 Thread 的方式實(shí)現(xiàn)線程時(shí)，要求自定義的類(lèi)的 __init__ 方法調(diào)用父類(lèi) Thread 的 __init__ 方法，如果忘記調(diào)用 Thread 的 __init__ 方法，則會(huì)報(bào)錯(cuò)。編寫(xiě) forget_init.py，其內(nèi)容如下：

import time
import threading;

class MyThread(threading.Thread):
    def __init__(self, id):
        # 在此處沒(méi)有調(diào)用父類(lèi) threading.Thread.__init__ 方法
        self.id = id

    def run(self):
        for i in range(3):
            print('This is thread %s' % self.id)
            time.sleep(3)

t1 = MyThread(0)
t1 = MyThread(1)
t0.start()
t1.start()
t0.join()
t1.join()

運(yùn)行 forget_init.py，程序輸出如下：

Traceback (most recent call last):
  File "forget_init.py", line 14, in <module>
    t0 = MyThread(0)
  File "forget_init.py", line 7, in __init__
    self.id = id
  File "/usr/lib/python3.6/threading.py", line 1089, in name
    assert self._initialized, "Thread.__init__() not called"
AssertionError: Thread.__init__() not called

以上錯(cuò)誤信息顯示，Thread.__init__ 沒(méi)有被調(diào)用。

2.4.2 只有一個(gè)線程參數(shù)時(shí)，使用 (arg) 表示線程參數(shù)

元組只包含一個(gè)元素時(shí)，必須加一個(gè)逗號(hào)，在下面的定義中，變量 tuple 表示的是一個(gè)元組，該元組包含了一個(gè)元素 123。

>>> tuple = (123,)
>>> tuple
(123,)

在下面的定義中，忘記加逗號(hào)，則變量 expression 表示的是一個(gè)整數(shù)類(lèi)型的表達(dá)式，變量 expression 是一個(gè)整數(shù) 123，而不是元組 (123,)。

>>> expression = (123)
>>> expression
123

通過(guò)指定線程入口函數(shù)的方式實(shí)現(xiàn)線程時(shí)，使用元組傳遞線程參數(shù)，如果只有一個(gè)線程參數(shù) arg，使用 (arg) 表示線程參數(shù)時(shí)，則會(huì)報(bào)錯(cuò)。編寫(xiě)程序 not_tuple.py，內(nèi)容如下：

import time
import threading;

def run(id):
    for i in range(3):
        print('This is thread %d' % id)
        time.sleep(3)

t0 = threading.Thread(target = run, args = (0)) # 此處錯(cuò)誤，應(yīng)為(0,)
t1 = threading.Thread(target = run, args = (1)) # 此處錯(cuò)誤，應(yīng)為(1,)
t0.start()
t1.start()
t0.join()
t1.join()

運(yùn)行 not_tuple.py，程序輸出如下：

Exception in thread Thread-1:
Traceback (most recent call last):
  File "/usr/lib/python3.6/threading.py", line 916, in _bootstrap_inner
    self.run()
  File "/usr/lib/python3.6/threading.py", line 864, in run
    self._target(*self._args, **self._kwargs)
TypeError: run() argument after * must be an iterable, not int

以上顯示錯(cuò)誤信息 “TypeError: run() argument after * must be an iterable, not int”，初學(xué)者很難看明白這段錯(cuò)誤信息，這段錯(cuò)誤信息表示 run() 的 arguments 必須是可以遍歷的(iterable)。線程入口參數(shù)是一個(gè)元組，而參數(shù) (0) 表示的是一個(gè)整數(shù)而不是元組 (0,)。

3. 使用多線程進(jìn)行并行 IO 操作

本節(jié)通過(guò)實(shí)例說(shuō)明 Python 多線程的使用場(chǎng)景?，F(xiàn)在需要編寫(xiě)程序獲取 baidu.com、taobao.com、qq.com 首頁(yè)，程序包括 3 個(gè)任務(wù)：

• 獲取 baidu.com 的首頁(yè)
• 獲取 taobao.com 的首頁(yè)
• 獲取 qq.com 的首頁(yè)

本節(jié)需要使用到 python 的 requests 模塊，requests 模塊的用于 http 請(qǐng)求，requests 模塊提供了 get 方法用于獲取網(wǎng)頁(yè)。

在 3.1 小節(jié)演示串行執(zhí)行這 3 個(gè)任務(wù)，并記錄串行完成 3 個(gè)任務(wù)總共所需要的時(shí)間；在 3.2 小節(jié)演示并行執(zhí)行這 3 個(gè)任務(wù)，并記錄并行完成 3 個(gè)任務(wù)總共所需要的時(shí)間。

3.1 串行獲取 baidu.com、taobao.com、qq.com 首頁(yè)

編寫(xiě)程序 serial.py，該程序以串行的方式獲取 baidu、taobao、qq 的首頁(yè)，內(nèi)容如下：

from datetime import datetime
import requests
import threading

def fetch(url):
    response = requests.get(url)
    print('Get %s: %s' % (url, response))

time0 = datetime.now()

fetch("https://www.baidu.com/")
fetch("https://www.taobao.com/")
fetch("https://www.qq.com/")

time1 = datetime.now()
time = time1 - time0
print(time.microseconds)

• 在第 5 行，定義了函數(shù) fetch，函數(shù) fetch 獲取指定 url 的網(wǎng)頁(yè)。

• 在第 6 行，調(diào)用 requests 模塊的 get 方法獲取獲取指定 url 的網(wǎng)頁(yè)。

• 在第 9 行，記錄執(zhí)行的開(kāi)始時(shí)間。

• 在第 11 行到第 13 行，串行執(zhí)行獲取 baidu、taobao、qq 的首頁(yè)。

• 在第 15 行到第 17 行，記錄執(zhí)行的結(jié)束時(shí)間，并計(jì)算總共花費(fèi)的時(shí)間，time.micoseconds 表示完成需要的時(shí)間（微秒）。

執(zhí)行 serial.py，輸出如下：

Get https://www.baidu.com/: <Response [200]>
Get https://www.taobao.com/: <Response [200]>
Get https://www.qq.com/: <Response [200]>
683173

在輸出中，<Response [200]> 是服務(wù)器返回的狀態(tài)碼，表示獲取成功。成功獲取了 baidu、taobao、qq 的首頁(yè)，總共用時(shí)為 683173 微秒。

3.2 并行獲取 baidu.com、taobao.com、qq.com 首頁(yè)

編寫(xiě)程序 parallel.py，該程序以并行的方式獲取 baidu、taobao、qq 的首頁(yè)，內(nèi)容如下：

from datetime import datetime
import requests
import threading

def fetch(url):
    response = requests.get(url)
    print('Get %s: %s' % (url, response))

time0 = datetime.now()

t0 = threading.Thread(target = fetch, args = ("https://www.baidu.com/",))
t1 = threading.Thread(target = fetch, args = ("https://www.taobao.com/",))
t2 = threading.Thread(target = fetch, args = ("https://www.qq.com/",))
t0.start()
t1.start()
t2.start()
t0.join()
t1.join()
t2.join()

time1 = datetime.now()
time = time1 - time0
print(time.microseconds)

• 在第 5 行，定義了函數(shù) fetch，函數(shù) fetch 獲取指定 url 的網(wǎng)頁(yè)。

• 在第 6 行，調(diào)用 requests 模塊的 get 方法獲取獲取指定 url 的網(wǎng)頁(yè)。

• 在第 9 行，記錄執(zhí)行的開(kāi)始時(shí)間。

• 在第 11 行到第 13 行，創(chuàng)建了 3 個(gè)線程，分別執(zhí)行獲取 baidu、taobao、qq 的首頁(yè)。

• 在第 14 行到第 16 行，啟動(dòng)這 3 個(gè)線程，這 3 個(gè)線程并行執(zhí)行。

• 在第 17 行到第 19 行，等待這 3 個(gè)線程執(zhí)行完畢。

• 在第 21 行到第 23 行，記錄執(zhí)行的結(jié)束時(shí)間，并計(jì)算總共花費(fèi)的時(shí)間，time.micoseconds 表示完成需要的時(shí)間（微秒）。

執(zhí)行 parallel.py，輸出如下：

Get https://www.baidu.com/: <Response [200]>
Get https://www.qq.com/: <Response [200]>
Get https://www.taobao.com/: <Response [200]>
383800

在輸出中，<Response [200]> 是服務(wù)器返回的狀態(tài)碼，表示獲取成功。成功獲取了 baidu、taobao、qq的首頁(yè)，總共用時(shí)為 383800 微秒。相比執(zhí)行，串行執(zhí)行總共用時(shí)為 683173 微秒，因此使用多線程加快了程序的執(zhí)行速度。

4. 獲取線程的返回值

在繼承 Thread 實(shí)現(xiàn)多線程的方式中，將線程的返回值保存在線程對(duì)象中，使用一個(gè)成員變量保存線程的返回值。下面通過(guò)一個(gè)具體的例子，說(shuō)明如何獲取線程的返回值。使用多線程技術(shù)計(jì)算 1+2+3 … + 100 的累加和，算法思路如下：

• 主程序創(chuàng)建 2 個(gè)線程：

• 線程 1，計(jì)算前 50 項(xiàng)的累加和，即 1+2+3 … + 50，保存計(jì)算結(jié)果。

• 線程 2，計(jì)算后 50 項(xiàng)的累加和，即 51+52+53 … + 100，保存計(jì)算結(jié)果。

• 主程序等待線程 1 和線程 2 執(zhí)行完畢，獲取它們各自的計(jì)算結(jié)果，并相加得到最終的計(jì)算結(jié)果。

編寫(xiě)程序 get_return_value.py，其內(nèi)容如下：

import threading

class MyThread(threading.Thread):
    def __init__(self, begin, end):
        threading.Thread.__init__(self)
        self.begin = begin
        self.end = end

    def run(self):
        self.result = 0
        for i in range(self.begin, self.end):
            self.result += i

t0 = MyThread(1,51)
t1 = MyThread(51,101)
t0.start()
t1.start()
t0.join()
t1.join()
print(t0.result)
print(t1.result)
print(t0.result + t1.result)

• 在第 14 行，創(chuàng)建第一個(gè)線程，計(jì)算區(qū)間 [1, 51) 內(nèi)的累加和。

• 在第 15 行，創(chuàng)建第二個(gè)線程，計(jì)算區(qū)間 [51, 101) 內(nèi)的累加和。

• 在第 4 行，函數(shù) __init__ 將線程參數(shù) begin 和 end 保存到線程對(duì)象中。

• 在第 9 行，線程啟動(dòng)后執(zhí)行函數(shù) run。

• 在第 10 行到第 12 行，使用 self.result 保存線程的計(jì)算結(jié)果。

• 在第 16 行到第 19 行，啟動(dòng)線程進(jìn)行計(jì)算，主程序等待子線程計(jì)算結(jié)束。

• 在第 20 行到第 22 行，從 t0.result 中獲取線程 t0 的計(jì)算結(jié)果，從 t1.result 中獲取線程 t1 的計(jì)算結(jié)果，將兩者相加，打印最終的結(jié)果。

運(yùn)行程序 get_return_value.py，輸出如下：

1275
3775
5050

線程 t0 計(jì)算前 50 項(xiàng)，計(jì)算結(jié)果為 1275；線程 t1 計(jì)算后 50 項(xiàng)，計(jì)算結(jié)果為 3775；主程序?qū)烧呦嗉?，得到最終的計(jì)算結(jié)果為 5050。