python多線程詳解

一、線程介紹

什么是線程

線程（Thread）也叫輕量級(jí)進(jìn)程，是操作系統(tǒng)能夠進(jìn)行運(yùn)算調(diào)度的最小單位，它被包涵在進(jìn)程之中，是進(jìn)程中的實(shí)際運(yùn)作單位。線程自己不擁有系統(tǒng)資源，只擁有一點(diǎn)兒在運(yùn)行中必不可少的資源，但它可與同屬一個(gè)進(jìn)程的其它線程共享進(jìn)程所擁有的全部資源。一個(gè)線程可以創(chuàng)建和撤消另一個(gè)線程，同一進(jìn)程中的多個(gè)線程之間可以并發(fā)執(zhí)行。

為什么要使用多線程

線程在程序中是獨(dú)立的、并發(fā)的執(zhí)行流。與分隔的進(jìn)程相比，進(jìn)程中線程之間的隔離程度要小，它們共享內(nèi)存、文件句柄和其他進(jìn)程應(yīng)有的狀態(tài)。

因?yàn)榫€程的劃分尺度小于進(jìn)程，使得多線程程序的并發(fā)性高。進(jìn)程在執(zhí)行過(guò)程中擁有獨(dú)立的內(nèi)存單元，而多個(gè)線程共享內(nèi)存，從而極大地提高了程序的運(yùn)行效率。

線程比進(jìn)程具有更高的性能，這是由于同一個(gè)進(jìn)程中的線程都有共性多個(gè)線程共享同一個(gè)進(jìn)程的虛擬空間。線程共享的環(huán)境包括進(jìn)程代碼段、進(jìn)程的公有數(shù)據(jù)等，利用這些共享的數(shù)據(jù)，線程之間很容易實(shí)現(xiàn)通信。

操作系統(tǒng)在創(chuàng)建進(jìn)程時(shí)，必須為該進(jìn)程分配獨(dú)立的內(nèi)存空間，并分配大量的相關(guān)資源，但創(chuàng)建線程則簡(jiǎn)單得多。因此，使用多線程來(lái)實(shí)現(xiàn)并發(fā)比使用多進(jìn)程的性能要高得多。

總結(jié)起來(lái)，使用多線程編程具有如下幾個(gè)優(yōu)點(diǎn)：

• 進(jìn)程之間不能共享內(nèi)存，但線程之間共享內(nèi)存非常容易。

• 操作系統(tǒng)在創(chuàng)建進(jìn)程時(shí)，需要為該進(jìn)程重新分配系統(tǒng)資源，但創(chuàng)建線程的代價(jià)則小得多。因此，使用多線程來(lái)實(shí)現(xiàn)多任務(wù)并發(fā)執(zhí)行比使用多進(jìn)程的效率高。

• Python 語(yǔ)言?xún)?nèi)置了多線程功能支持，而不是單純地作為底層操作系統(tǒng)的調(diào)度方式，從而簡(jiǎn)化了 Python 的多線程編程。

二、線程實(shí)現(xiàn)

threading模塊

普通創(chuàng)建方式

          
            import threading
import time

def run(n):
    print("task", n)
    time.sleep(1)
    print('2s')
    time.sleep(1)
    print('1s')
    time.sleep(1)
    print('0s')
    time.sleep(1)

if __name__ == '__main__':
    t1 = threading.Thread(target=run, args=("t1",))
    t2 = threading.Thread(target=run, args=("t2",))
    t1.start()
    t2.start()

----------------------------------

>>> task t1
>>> task t2
>>> 2s
>>> 2s
>>> 1s
>>> 1s
>>> 0s
>>> 0s
          
        

自定義線程

繼承threading.Thread來(lái)自定義線程類(lèi)，其本質(zhì)是重構(gòu)Thread類(lèi)中的run方法

          
            import threading
import time

class MyThread(threading.Thread):
    def __init__(self, n):
        super(MyThread, self).__init__()  # 重構(gòu)run函數(shù)必須要寫(xiě)
        self.n = n

    def run(self):
        print("task", self.n)
        time.sleep(1)
        print('2s')
        time.sleep(1)
        print('1s')
        time.sleep(1)
        print('0s')
        time.sleep(1)

if __name__ == "__main__":
    t1 = MyThread("t1")
    t2 = MyThread("t2")
    t1.start()
    t2.start()
    
----------------------------------

>>> task t1
>>> task t2
>>> 2s
>>> 2s
>>> 1s
>>> 1s
>>> 0s
>>> 0s
          
        

守護(hù)線程

我們看下面這個(gè)例子，這里使用setDaemon(True)把所有的子線程都變成了主線程的守護(hù)線程，因此當(dāng)主進(jìn)程結(jié)束后，子線程也會(huì)隨之結(jié)束。所以當(dāng)主線程結(jié)束后，整個(gè)程序就退出了。

          
            import threading
import time

def run(n):
    print("task", n)
    time.sleep(1)       #此時(shí)子線程停1s
    print('3')
    time.sleep(1)
    print('2')
    time.sleep(1)
    print('1')

if __name__ == '__main__':
    t = threading.Thread(target=run, args=("t1",))
    t.setDaemon(True)   #把子進(jìn)程設(shè)置為守護(hù)線程，必須在start()之前設(shè)置
    t.start()
    print("end")
    
----------------------------------

>>> task t1
>>> end
          
        

我們可以發(fā)現(xiàn)，設(shè)置守護(hù)線程之后，當(dāng)主線程結(jié)束時(shí)，子線程也將立即結(jié)束，不再執(zhí)行。

主線程等待子線程結(jié)束

為了讓守護(hù)線程執(zhí)行結(jié)束之后，主線程再結(jié)束，我們可以使用join方法，讓主線程等待子線程執(zhí)行。

          
            import threading
import time

def run(n):
    print("task", n)
    time.sleep(1)       #此時(shí)子線程停1s
    print('3')
    time.sleep(1)
    print('2')
    time.sleep(1)
    print('1')

if __name__ == '__main__':
    t = threading.Thread(target=run, args=("t1",))
    t.setDaemon(True)   #把子進(jìn)程設(shè)置為守護(hù)線程，必須在start()之前設(shè)置
    t.start()
    t.join() # 設(shè)置主線程等待子線程結(jié)束
    print("end")

----------------------------------

>>> task t1
>>> 3
>>> 2
>>> 1
>>> end
          
        

多線程共享全局變量

線程是進(jìn)程的執(zhí)行單元，進(jìn)程是系統(tǒng)分配資源的最小單位，所以在同一個(gè)進(jìn)程中的多線程是共享資源的。

          
            import threading
import time

g_num = 100

def work1():
    global g_num
    for i in range(3):
        g_num += 1
    print("in work1 g_num is : %d" % g_num)

def work2():
    global g_num
    print("in work2 g_num is : %d" % g_num)

if __name__ == '__main__':
    t1 = threading.Thread(target=work1)
    t1.start()
    time.sleep(1)
    t2 = threading.Thread(target=work2)
    t2.start()

----------------------------------

>>> in work1 g_num is : 103
>>> in work2 g_num is : 103
          
        

互斥鎖

由于線程之間是進(jìn)行隨機(jī)調(diào)度，并且每個(gè)線程可能只執(zhí)行n條執(zhí)行之后，當(dāng)多個(gè)線程同時(shí)修改同一條數(shù)據(jù)時(shí)可能會(huì)出現(xiàn)臟數(shù)據(jù)，所以，出現(xiàn)了線程鎖，即同一時(shí)刻允許一個(gè)線程執(zhí)行操作。線程鎖用于鎖定資源，你可以定義多個(gè)鎖, 像下面的代碼, 當(dāng)你需要獨(dú)占某一資源時(shí)，任何一個(gè)鎖都可以鎖這個(gè)資源，就好比你用不同的鎖都可以把相同的一個(gè)門(mén)鎖住是一個(gè)道理。

由于線程之間是進(jìn)行隨機(jī)調(diào)度，如果有多個(gè)線程同時(shí)操作一個(gè)對(duì)象，如果沒(méi)有很好地保護(hù)該對(duì)象，會(huì)造成程序結(jié)果的不可預(yù)期，我們也稱(chēng)此為“線程不安全”。

為了方式上面情況的發(fā)生，就出現(xiàn)了互斥鎖(Lock)

          
            from threading import Thread,Lock
import os,time
def work():
    global n
    lock.acquire()
    temp=n
    time.sleep(0.1)
    n=temp-1
    lock.release()
if __name__ == '__main__':
    lock=Lock()
    n=100
    l=[]
    for i in range(100):
        p=Thread(target=work)
        l.append(p)
        p.start()
    for p in l:
        p.join()
          
        

遞歸鎖

RLcok類(lèi)的用法和Lock類(lèi)一模一樣，但它支持嵌套，在多個(gè)鎖沒(méi)有釋放的時(shí)候一般會(huì)使用RLcok類(lèi)。

          
            import threading
import time

def Func(lock):
    global gl_num
    lock.acquire()
    gl_num += 1
    time.sleep(1)
    print(gl_num)
    lock.release()

if __name__ == '__main__':
    gl_num = 0
    lock = threading.RLock()
    for i in range(10):
        t = threading.Thread(target=Func, args=(lock,))
        t.start()
          
        

信號(hào)量（BoundedSemaphore類(lèi))

互斥鎖同時(shí)只允許一個(gè)線程更改數(shù)據(jù)，而Semaphore是同時(shí)允許一定數(shù)量的線程更改數(shù)據(jù) ，比如廁所有3個(gè)坑，那最多只允許3個(gè)人上廁所，后面的人只能等里面有人出來(lái)了才能再進(jìn)去。

          
            import threading
import time

def run(n, semaphore):
    semaphore.acquire()   #加鎖
    time.sleep(1)
    print("run the thread:%s\n" % n)
    semaphore.release()     #釋放

if __name__ == '__main__':
    num = 0
    semaphore = threading.BoundedSemaphore(5)  # 最多允許5個(gè)線程同時(shí)運(yùn)行
    for i in range(22):
        t = threading.Thread(target=run, args=("t-%s" % i, semaphore))
        t.start()
    while threading.active_count() != 1:
        pass  # print threading.active_count()
    else:
        print('-----all threads done-----')
          
        

事件（Event類(lèi)）

python線程的事件用于主線程控制其他線程的執(zhí)行，事件是一個(gè)簡(jiǎn)單的線程同步對(duì)象，其主要提供以下幾個(gè)方法：

• clear 將flag設(shè)置為“False”
• set 將flag設(shè)置為“True”
• is_set 判斷是否設(shè)置了flag
• wait 會(huì)一直監(jiān)聽(tīng)flag，如果沒(méi)有檢測(cè)到flag就一直處于阻塞狀態(tài)

事件處理的機(jī)制：全局定義了一個(gè)“Flag”，當(dāng)flag值為“False”，那么event.wait()就會(huì)阻塞，當(dāng)flag值為“True”，那么event.wait()便不再阻塞。

          
            #利用Event類(lèi)模擬紅綠燈
import threading
import time

event = threading.Event()


def lighter():
    count = 0
    event.set()     #初始值為綠燈
    while True:
        if 5 < count <=10 :
            event.clear()  # 紅燈，清除標(biāo)志位
            print("\33[41;1mred light is on...\033[0m")
        elif count > 10:
            event.set()  # 綠燈，設(shè)置標(biāo)志位
            count = 0
        else:
            print("\33[42;1mgreen light is on...\033[0m")

        time.sleep(1)
        count += 1

def car(name):
    while True:
        if event.is_set():      #判斷是否設(shè)置了標(biāo)志位
            print("[%s] running..."%name)
            time.sleep(1)
        else:
            print("[%s] sees red light,waiting..."%name)
            event.wait()
            print("[%s] green light is on,start going..."%name)

light = threading.Thread(target=lighter,)
light.start()

car = threading.Thread(target=car,args=("MINI",))
car.start()
          
        

三、GIL（Global Interpreter Lock）全局解釋器鎖

在非python環(huán)境中，單核情況下，同時(shí)只能有一個(gè)任務(wù)執(zhí)行。多核時(shí)可以支持多個(gè)線程同時(shí)執(zhí)行。但是在python中，無(wú)論有多少核，同時(shí)只能執(zhí)行一個(gè)線程。究其原因，這就是由于GIL的存在導(dǎo)致的。

GIL的全稱(chēng)是Global Interpreter Lock(全局解釋器鎖)，來(lái)源是python設(shè)計(jì)之初的考慮，為了數(shù)據(jù)安全所做的決定。某個(gè)線程想要執(zhí)行，必須先拿到GIL，我們可以把GIL看作是“通行證”，并且在一個(gè)python進(jìn)程中，GIL只有一個(gè)。拿不到通行證的線程，就不允許進(jìn)入CPU執(zhí)行。GIL只在cpython中才有，因?yàn)閏python調(diào)用的是c語(yǔ)言的原生線程，所以他不能直接操作cpu，只能利用GIL保證同一時(shí)間只能有一個(gè)線程拿到數(shù)據(jù)。而在pypy和jpython中是沒(méi)有GIL的。

Python多線程的工作過(guò)程：
python在使用多線程的時(shí)候，調(diào)用的是c語(yǔ)言的原生線程。

• 拿到公共數(shù)據(jù)
• 申請(qǐng)gil
• python解釋器調(diào)用os原生線程
• os操作cpu執(zhí)行運(yùn)算
• 當(dāng)該線程執(zhí)行時(shí)間到后，無(wú)論運(yùn)算是否已經(jīng)執(zhí)行完，gil都被要求釋放
• 進(jìn)而由其他進(jìn)程重復(fù)上面的過(guò)程
• 等其他進(jìn)程執(zhí)行完后，又會(huì)切換到之前的線程（從他記錄的上下文繼續(xù)執(zhí)行），整個(gè)過(guò)程是每個(gè)線程執(zhí)行自己的運(yùn)算，當(dāng)執(zhí)行時(shí)間到就進(jìn)行切換（context switch）。

python針對(duì)不同類(lèi)型的代碼執(zhí)行效率也是不同的：

1、CPU密集型代碼(各種循環(huán)處理、計(jì)算等等)，在這種情況下，由于計(jì)算工作多，ticks計(jì)數(shù)很快就會(huì)達(dá)到閾值，然后觸發(fā)GIL的釋放與再競(jìng)爭(zhēng)（多個(gè)線程來(lái)回切換當(dāng)然是需要消耗資源的），所以python下的多線程對(duì)CPU密集型代碼并不友好。
2、IO密集型代碼(文件處理、網(wǎng)絡(luò)爬蟲(chóng)等涉及文件讀寫(xiě)的操作)，多線程能夠有效提升效率(單線程下有IO操作會(huì)進(jìn)行IO等待，造成不必要的時(shí)間浪費(fèi)，而開(kāi)啟多線程能在線程A等待時(shí)，自動(dòng)切換到線程B，可以不浪費(fèi)CPU的資源，從而能提升程序執(zhí)行效率)。所以python的多線程對(duì)IO密集型代碼比較友好。

使用建議？

python下想要充分利用多核CPU，就用多進(jìn)程。因?yàn)槊總€(gè)進(jìn)程有各自獨(dú)立的GIL，互不干擾，這樣就可以真正意義上的并行執(zhí)行，在python中，多進(jìn)程的執(zhí)行效率優(yōu)于多線程(僅僅針對(duì)多核CPU而言)。

GIL在python中的版本差異：

1、在python2.x里，GIL的釋放邏輯是當(dāng)前線程遇見(jiàn)IO操作或者ticks計(jì)數(shù)達(dá)到100時(shí)進(jìn)行釋放。（ticks可以看作是python自身的一個(gè)計(jì)數(shù)器，專(zhuān)門(mén)做用于GIL，每次釋放后歸零，這個(gè)計(jì)數(shù)可以通過(guò)sys.setcheckinterval 來(lái)調(diào)整）。而每次釋放GIL鎖，線程進(jìn)行鎖競(jìng)爭(zhēng)、切換線程，會(huì)消耗資源。并且由于GIL鎖存在，python里一個(gè)進(jìn)程永遠(yuǎn)只能同時(shí)執(zhí)行一個(gè)線程(拿到GIL的線程才能執(zhí)行)，這就是為什么在多核CPU上，python的多線程效率并不高。
2、在python3.x中，GIL不使用ticks計(jì)數(shù)，改為使用計(jì)時(shí)器（執(zhí)行時(shí)間達(dá)到閾值后，當(dāng)前線程釋放GIL），這樣對(duì)CPU密集型程序更加友好，但依然沒(méi)有解決GIL導(dǎo)致的同一時(shí)間只能執(zhí)行一個(gè)線程的問(wèn)題，所以效率依然不盡如人意。