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

Python中線程有兩種方式：函數(shù)或者用類來包裝線程對(duì)象。threading模塊中包含了豐富的多線程支持功能：

• threading.currentThread(): 返回當(dāng)前線程；
• threading.enumerate(): 返回包含正在運(yùn)行的線程列表；
• threading.activeCount(): 返回正在運(yùn)行的線程數(shù)量，與len(threading.enumerate())等價(jià)。

Thread類

通過Thread類來處理線程，類中提供的一些方法：

• run(): 用以表示線程執(zhí)行的方法（可重載實(shí)現(xiàn)實(shí)際功能）；
• start(): 啟動(dòng)線程；
• join([time]): 等待線程中止（或者超時(shí)）；
• isAlive(): 返回線程是否活動(dòng)；
• getName(): 返回線程名；
• setName(): 設(shè)置線程名；
• setDaemon(True)：設(shè)置為后臺(tái)進(jìn)程（必須在start調(diào)用前設(shè)定）。

函數(shù)方式

通過Thread直接構(gòu)造線程，然后通過start方法啟動(dòng)線程：

threading.Thread(group=None, target=None, name=None, args=(), kwargs=None, *,daemon=None)

各參數(shù)說明：

• group：指定線程隸屬的線程組（當(dāng)前忽略）；
• target：指定線程要調(diào)度的目標(biāo)方法（即實(shí)現(xiàn)功能的函數(shù)）；
• args：傳遞給目標(biāo)方法的參數(shù)（以元組的方式）；
• kwargs：傳遞給目標(biāo)方法的參數(shù)（以字典的方式）；
• daemon：指定線程是否為后臺(tái)線程。

def simpleRoutine(name, delay):
    print(f"routine {name} starting...")
    time.sleep(delay)
    print(f"routine {name} finished")
if __name__ == '__main__':
    thrOne = threading.Thread(target=simpleRoutine, args=("First", 1))
    thrTwo = threading.Thread(target=simpleRoutine, args=("Two", 2))
    thrOne.start()
    thrTwo.start()
    thrOne.join()
    thrTwo.join()

繼承方式

直接繼承Thread，創(chuàng)建一個(gè)新的子類（主要實(shí)現(xiàn)run方法）：

class SimpleThread (threading.Thread):
    def __init__(self, name, delay):
        # threading.Thread.__init__(self)
        super().__init__()
        self.name = name
        self.delay = delay
    def run(self):
        print(f"thread {self.name} starting...")
        time.sleep(self.delay)
        print(f"thread {self.name} finished")
if __name__ == '__main__':
    thrOne = SimpleThread("First", 2)
    thrTwo = SimpleThread("Second", 2)
    thrOne.start()
    thrTwo.start()
    thrOne.join()
    thrTwo.join()

同步機(jī)制

當(dāng)多個(gè)線程同時(shí)修改同一條數(shù)據(jù)時(shí)可能會(huì)出現(xiàn)臟數(shù)據(jù)；所以，就需要線程鎖，即同一時(shí)刻只允許一個(gè)線程執(zhí)行操作。

同步鎖Lock

threading提供了Lock和RLock（可重入鎖）兩個(gè)類，它們都提供了如下兩個(gè)方法來加鎖和釋放鎖：

• acquire(blocking=True, timeout=-1)：加鎖，其中 timeout 參數(shù)指定加鎖多少秒。
• release()：釋放鎖。

兩種使用鎖的方式：

gCount = 0
def PlusOne(locker):
    global gCount
      with locker:
          gCount += 1、
def MinusOne(locker):
    global gCount
      if locker.acquire():
          gCount -= 1
          locker.release()

條件變量Condition

Condition對(duì)象內(nèi)部維護(hù)了一個(gè)鎖（構(gòu)造時(shí)可傳遞一個(gè)Lock/RLock對(duì)象，否則內(nèi)部會(huì)自行創(chuàng)建一個(gè)RLock)和一個(gè)waiting池：

• 通過acquire獲得Condition對(duì)象；
• 當(dāng)調(diào)用wait方法時(shí)，線程會(huì)釋放Condition內(nèi)部的鎖并進(jìn)入blocked狀態(tài)，同時(shí)在waiting池中記錄這個(gè)線程；
• 當(dāng)調(diào)用notify方法時(shí)，Condition對(duì)象會(huì)從waiting池中挑選一個(gè)線程，通知其調(diào)用acquire方法嘗試取到鎖。

Condition對(duì)象：

__init__(self,lock=None)：Condition類總是與一個(gè)鎖相關(guān)聯(lián)（若不指定lock參數(shù)，會(huì)自動(dòng)創(chuàng)建一個(gè)與之綁定的RLock對(duì)象）；

acquire(timeout)：調(diào)用關(guān)聯(lián)鎖的acquire()方法；

release()：調(diào)用關(guān)聯(lián)鎖的release()方法

wait(timeout)：線程掛起，直到收到一個(gè)notify通知或超時(shí)才會(huì)被喚醒；必須在已獲得鎖的前提下調(diào)用；

notify(n=1)：?jiǎn)拘褀aiting池中的n個(gè)正在等待的線程并通知它：

• 收到通知的線程將自動(dòng)調(diào)用acquire()方法嘗試加鎖；
• 若waiting池中有多個(gè)線程，隨機(jī)選擇n個(gè)喚醒；
• 必須在已獲得鎖的前提下調(diào)用，否則將引發(fā)錯(cuò)誤。

notify_all()：通知所有線程。

class Producer(threading.Thread):
    def __init__(self, cond, storage):
        threading.Thread.__init__(self)
        self.cond = cond
        self.storage = storage
    def run(self):
        label = 1
        while True:
            with self.cond:
                if len(self.storage) < 10:
                    self.storage.append(label)
                    print(f"<- Produce {label} product")
                    label += 1
                    self.cond.notify(2)
                else:
                    print(f"<- storage full: Has Produced {label - 1} product")
                    self.cond.notify_all()
                    self.cond.wait()
                time.sleep(0.4)
class Consumer(threading.Thread):
    def __init__(self, name, cond, storage):
        threading.Thread.__init__(self)
        self.name = name
        self.cond = cond
        self.storage = storage
    def run(self):
        while True:
            if self.cond.acquire():
                if len(self.storage) > 1:
                    pro = self.storage.pop(0)
                    print(f"-> {self.name} consumed {pro}")
                    self.cond.notify()
                else:
                    print(f"-> {self.name} storage empty: no product to consume")
                    self.cond.wait()
                self.cond.release()
                time.sleep(1)

信號(hào)量Semaphore

信號(hào)量對(duì)象內(nèi)部維護(hù)一個(gè)計(jì)數(shù)器：

• acquire(blocking=True,timeout=None)時(shí)減1，當(dāng)計(jì)數(shù)為0就阻塞請(qǐng)求的線程；
• release()時(shí)加1，當(dāng)計(jì)數(shù)大于0恢復(fù)被阻塞的線程；

threading中有Semaphore和BoundedSemaphore兩個(gè)信號(hào)量；BoundedSemaphore限制了release的次數(shù)，任何時(shí)候計(jì)數(shù)器的值，都不不能大于初始值（release時(shí)會(huì)檢測(cè)計(jì)數(shù)器的值，若大于等于初始值，則拋出ValueError異常）。

通過Semaphore維護(hù)生產(chǎn)（release一個(gè)）、消費(fèi)（acquire一個(gè)）量：

# products = threading.Semaphore(0)
def produceOne(label, sem: threading.Semaphore):
    sem.release()
    print(f"{label} produce one")
def consumeOne(label, sem: threading.Semaphore):
    sem.acquire()
    print(f"{label} consume one")

通過BoundedSemaphore來控制并發(fā)數(shù)量（最多有Semaphore初始值數(shù)量的線程并發(fā)）：

# runner = threading.BoundedSemaphore(3)
def runBound(name, sem: threading.BoundedSemaphore):
    with sem:
        print(f"{name} is running")
        time.sleep(1)
        print(f"{name} finished")

事件Event

事件對(duì)象內(nèi)部有個(gè)標(biāo)志字段，用于線程等待事件的發(fā)生：

• isSet()：返回event的狀態(tài)值；
• wait()：狀態(tài)為False時(shí)，一直阻塞；否則立即返回；
• set()： 設(shè)置狀態(tài)值為True，激活所有被阻塞的線程；
• clear()：恢復(fù)狀態(tài)值為False。

多線程等待事件發(fā)生，然后開始執(zhí)行：

def waiters(name, evt: threading.Event):
    evt.wait()
    print(f"{name} is running")
    time.sleep(1)
    print(f"{name} finished")
def starting(evt: threading.Event):
    evt.set()
    print("event is set")

屏障Barrier

屏障用于設(shè)定等待線程數(shù)量，當(dāng)數(shù)量達(dá)到指定值時(shí)，開始執(zhí)行：

threading.Barrier(parties, action=None, timeout=None)

屏障屬性與方法：

• wait(timeout=None)：等待通過屏障；線程被阻塞，直到阻塞的數(shù)量達(dá)到parties時(shí)，被阻塞的線程被同時(shí)全部釋放；
• reset()：重置屏障到默認(rèn)的空狀態(tài)；
• abort()：將障礙置為斷開狀態(tài)；導(dǎo)致等待的線程引發(fā)BrokenBarrierError異常；
• partier()：通過障礙所需的線程數(shù)；
• n_waiting()：當(dāng)前在屏障中等待的線程數(shù)；
• broken()：如果屏障處于斷開狀態(tài)，則返回True。

def waitBarrier(name, barr: threading.Barrier):
    print(f"{name} waiting for open")
    try:
        barr.wait()
        print(f"{name} running")
        time.sleep(5)
    except threading.BrokenBarrierError:
        print(f"{name} exception")
    print(f"{name} finished")

GIL全局解釋器鎖

GIL（Global Interpreter Lock，全局解釋器鎖）；cpython中，某個(gè)線程想要執(zhí)行，必須先拿到GIL（可以把GIL看作是“通行證”）。每次釋放GIL鎖，線程都要進(jìn)行鎖競(jìng)爭(zhēng)，切換線程，會(huì)消耗資源。

由于GIL鎖的存在，python里一個(gè)進(jìn)程永遠(yuǎn)只能同時(shí)執(zhí)行一個(gè)線程(拿到GIL的線程)，這就是為什么在多核CPU上，python的多線程效率并不高：

• CPU密集型代碼：由于計(jì)算工作多，會(huì)很快用完時(shí)間片，然后觸發(fā)GIL的釋放與再競(jìng)爭(zhēng)；
• IO密集型代碼(文件處理、網(wǎng)絡(luò)爬蟲等)：多線程能夠有效提升效率(單線程下有IO操作會(huì)進(jìn)行IO等待，造成不必要的時(shí)間浪費(fèi)，而開啟多線程能在線程A等待時(shí)，自動(dòng)切換到線程B，可以不浪費(fèi)CPU的資源，從而能提升程序執(zhí)行效率)。

python在使用多線程的時(shí)候，調(diào)用的是c語言的原生線程：

• 拿到公共數(shù)據(jù)
• 申請(qǐng)GIL
• python解釋器調(diào)用os原生線程
• os操作cpu執(zhí)行運(yùn)算
• 當(dāng)線程執(zhí)行時(shí)間到后，就進(jìn)行切換（context switch）

到此這篇關(guān)于Python多線程與同步機(jī)制淺析的文章就介紹到這了,更多相關(guān)Python多線程內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論