python模塊之threading

threading在低級的_thread模塊上構建了更高級的線程接口。

threading模塊基於Java線程模型設計。不過Java中鎖和條件變量是每一個對象的基本行爲，在python中倒是單獨的對象。python的Thread類行爲是Java的Thread類行爲的子集，目前尚不支持優先級、線程組，線程沒法銷燬、中止、暫停、恢復或中斷。Java中Thread類的靜態方法在Python中映射爲模塊級的函數。

模塊級函數

threading.active_count()
返回當前活動的Thread對象的數量，與enumerate()函數返回的列表元素個數相同

threading.current_thread()
返回當前Thread對象，對應調用者的控制線程(thread of control)。若是調用者的控制線程不是經過threading模塊建立，返回一個功能受限的啞線程對象(dummy thread object)

threading.get_ident()
返回一個非零整數，表明當前線程的"線程標識符"。這個值意在做爲魔術cookie使用，例如做爲索引從特定於線程的字典對象獲取數據。當一個線程退出，新的線程建立，線程標識符可能被回收使用

threading.enumerate()
返回當前活動Thread對象的列表。該列表包含守護線程、current_thread()函數建立的啞線程，以及主線程，不包含已終止的線程和未啓動的線程。

threading.main_thread()
返回主線程對象。一般來講，主線程就是啓動python解釋器的線程。

threading.settrace(func)
爲啓動自threading模塊的全部線程設置一個trace函數。在每一個線程的run()方法調用前，傳遞func參數給sys.settrace()

threading.setprofile(func)
爲啓動自threading模塊的全部線程設置一個profile函數。在每一個線程的run()方法調用前，傳遞func參數給sys.setprofile()

threading.stack_size([size])
返回建立新線程使用的線程堆棧大小。

可選參數size指定後續建立的線程的堆棧大小，必須是0(表示使用平臺或配置的默認值)或大於等於32768(32KiB)的正整數。若是未指定，默認size爲0.

若是不支持改動線程堆棧大小，拋出RuntimeError異常。若是size不合法，拋出ValueError異常，堆棧大小保持不變。

32KiB是目前能保證解釋器堆棧空間充足的最小值。某些平臺可能對堆棧大小作了特殊的限制，好比要求最小堆棧大小在32KiB以上，或要求以系統內存頁大小的倍數分配。

Windows系統及使用POSIX線程的系統可用

常量

threading.TIMEOUT_MAX
阻塞函數(Lock.acquire(), RLock.acquire(), Condition.wait()等)的timeout參數可接受的最大值。超出該值將拋出OverflowError異常。

Thread-Local Data

Thread-local數據的值是特定於線程的。管理Thread-local數據，只須要建立local或其子類的實例並在該實例上存儲屬性：

mydata = threading.local()
mydata.x = 1

不一樣的線程，實例的值也會不一樣。

class threading.local

表示thread-local數據的類。

Thread

Thread類表明在單獨的控制線程中運行的活動，有兩種方式指定：傳遞可調用對象到構造器的target參數，或重寫子類的run()方法。除了__int__()方法和run()方法，Thread子類不該該重寫除此以外的其餘方法。

建立的線程對象，必須使用start()方法啓動，start()在一個單獨的控制線程調用run()方法。這時該線程被認爲是"活動的"。當run()方法結束(正常執行完成或拋出了未處理的異常)時，線程對象再也不是"活動的"。is_alive()方法可用於檢查線程是否處於活動狀態。

調用線程對象的join()方法將致使線程阻塞，直到調用join()方法的線程執行結束。

線程擁有名字，能夠傳遞給構造器。經過name屬性讀取或修改。

主線程：對應python程序的初始控制線程。主線程不是守護線程。

守護線程：當沒有非守護線程處於活動狀態時，整個python程序將退出。經過daemon屬性或構造器參數，能夠標記一個線程爲守護線程。daemon屬性的初始值繼承自建立該線程的線程

啞線程：對應"外部線程"alien thread，即在threading模塊以外(好比C代碼)啓動的控制線程。啞線程具備有限的功能，老是認爲是活動的和守護的，不能調用join()方法。它們永遠不會被刪除，由於不能檢測外部線程的結束狀況。

Note：守護線程將在程序關閉時直接中止。相關資源(好比打開的文件、數據庫事務等)可能不會被妥善地釋放。若是想要線程優雅地中止，將線程設置爲非守護線程，並使用合適的信號機制好比Event

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

• group：None。做爲未來實現ThreadGroup類後的保留參數。
• target：可調用對象，將被run()方法調用
• name：線程名稱。默認構建Thread-N形式的惟一名稱。
• args：target調用須要接收的位置參數，元組形式
• kwargs：target調用須要接收的關鍵字參數，字典形式
• daemon：傳遞一個布爾值，標記該線程是否爲守護線程。None表示繼承建立該線程的當前線程的daemon屬性。

若是子類繼承Thread並重寫構造器，必須確保在執行線程的其餘操做前在構造器中調用Thread.__init__()

start()
開啓線程。每一個線程最多隻能調用一次，不然拋出RuntimeError異常。它將在一個單獨的控制線程調用線程對象的run()方法。

run()
定義線程功能的方法，一般在子類中重寫。標準的run()方法調用傳入構造器的可調用對象target(存在的話)，並使用args和kwargs分別做爲target的位置參數和關鍵字參數。

# 建立Thread的實例，傳給它一個函數

from threading import Thread
from time import sleep, ctime

sleep_time = [4, 2]


def task(task_tag, sleep_tag):
    print("task", task_tag, "started at:", ctime())
    sleep(sleep_tag)
    print("task", task_tag, "done at:", ctime())


def main():
    print("Main thread started at:", ctime())
    threads = []
    nloops = range(len(sleep_time))  # [0, 1]

    for i in nloops:
        t = Thread(target=task, args=(i, sleep_time[i]))
        threads.append(t)

    for i in nloops:
        threads[i].start()  # 啓動線程

    for i in nloops:
        threads[i].join()  # 主線程阻塞，直至調用join()方法的線程終止

    print("Main thread done at:", ctime())


if __name__ == '__main__':
    main()

# 派生Thread的子類，並建立子類的實例

from threading import Thread
from time import sleep, ctime

sleep_time = [4, 2]


class MyThread(Thread):
    # 重寫run()方法
    def run(self):
        print(self.name, "started at:", ctime())
        self._target(self._args)
        print(self.name, "done at:", ctime())


def task(sleep_tag):
    sleep(sleep_tag)


def main():
    print("Main thread started at:", ctime())
    threads = []
    nloops = range(len(sleep_time))

    for i in nloops:
        t = MyThread(target=task, args=sleep_time[i], name=task.__name__ + str(i))
        threads.append(t)

    for i in nloops:
        threads[i].start()

    for i in nloops:
        threads[i].join()

    print("Main thread done at:", ctime())


if __name__ == '__main__':
    main()

join(timeout=None)
阻塞主線程直到調用join方法的線程終止(多是正常執行完成，也多是拋出了未處理的異常)或達到timeout設定的時間。可屢次調用。

• timeout：阻塞時間(秒)。若是爲None，表示一直阻塞直至調用join方法的線程終止；若是不爲None，表示阻塞的時間，達到該時間後，無論調用join()方法的線程是否執行完成，繼續執行主線程或其餘啓動的線程。

若是線程調用join()方法可能致使死鎖，或在調用start()以前調用join()，拋出RuntimeError異常。

name
獲取或設置線程名稱。多個線程可能名稱相同，初始值由構造器設置。

ident
線程標識符，若是爲None說明該線程未啓動。當一個線程退出，新的線程建立，線程標識符可能被回收使用。即便線程退出，該標識符仍可用。

is_alive()
判斷線程是否處於活動狀態。

daemon
布爾標誌，表示這個線程是不是守護線程。必須在調用start()以前設置，不然拋出RuntimeError異常。初始值繼承自建立該線程的線程。主線程不是守護線程，所以在主線程中建立的線程daemon屬性默認值爲False

CPython實現細節：在CPython中，因爲GIL的緣由，一次只有一個線程可以執行python代碼(即便某些面向性能的庫能克服這個限制???)。想要python程序更好地利用多核機器的計算機資源(計算密集型)，建議使用multiprocessing或concurrent.futures.ProcessPoolExecutor。若是是同時運行多個I/O密集型任務，threading仍然不失爲一個合適的模塊

Lock

原語鎖，是同步原語的一種，當它處於"locked"狀態時不屬於特定線程。在python中，這是目前可用的最低級的同步原語，實現自_thread擴展模塊。

原語鎖有兩種狀態：locked(鎖定)或unlocked(未鎖定)。建立時爲未鎖定狀態。
原語鎖有兩種方法：acquire()和release()。當鎖處於未鎖定狀態時，acquire()改變其爲鎖定狀態。當鎖處於鎖定狀態時，調用acquire()方法將致使線程阻塞，直到其餘線程調用release()釋放鎖。

class threading.Lock

acquire(blocking=True, timeout=-1)
獲取鎖。成功返回True，獲取返回False。

• blocking：默認爲True，在獲取到鎖以前阻塞線程；反之即便沒有獲取到鎖也不會阻塞線程。
• timeout：指定線程阻塞的最長時間，單位爲秒；-1表示無限制等待。當blocking爲False時，禁止指定timeout參數

release()
釋放鎖。任何線程均可以調用，不僅是獲取了鎖的線程。

鎖更改成未上鎖狀態後，對於調用了acquire()方法而致使阻塞的線程，將由系統決定哪一個線程獲取到鎖。

release()方法只能在上鎖狀態調用，不然將拋出RuntimeError異常。

RLock

重入鎖，同步原語的一種，可由同一線程屢次獲取已持有的鎖。除了原語鎖的上鎖/解鎖狀態，重入鎖還使用了owning thread和recursion level的概念。在上鎖狀態，可能有多個線程擁有鎖；在解鎖狀態，沒有線程擁有鎖。

acquire()/release()必須成對出現，能夠嵌套，只有最後一個release(即最外層的release)調用纔會最終釋放鎖。

class threading.RLock

acquire(blocking=True, timeout=-1)
使用默認參數調用時，若是當前線程已經擁有鎖，增長1次遞歸深度並當即返回；若是是其餘線程擁有鎖，阻塞當前線程直到鎖被釋放。一旦鎖釋放(遞歸深度爲0，此時鎖不屬於任何線程)，各個線程爭奪鎖，並設置遞歸深度爲1。

release()
釋放鎖且遞歸深度減1。若是調用後遞歸深度爲0，重置鎖爲未鎖定狀態(不屬於任何線程)，由其餘線程爭奪鎖。若是調用後遞歸深度非0，鎖仍爲上鎖狀態，屬於當前線程。

只能由已經獲取了鎖的線程調用，不然拋出RuntimeError異常。

Condition

condition變量老是與某種鎖相聯繫：傳入或者默認建立的鎖對象。傳入鎖對象適用於多個condition變量須要共享同一個鎖的場景。鎖是condition對象的一部分，不須要對鎖單獨進行追蹤。

condition對象遵循上下文管理協議：使用with語句在封閉塊內獲取關聯的鎖對象，在condition對象上調用acquire和release實際上調用的是關聯鎖的對應方法。

class threading.Condition(lock=None)

條件變量容許一個或多個線程等待，直到接收到另外一個線程的通知。

lock參數必須是Lock或RLock對象，做爲底層的鎖使用。默認使用RLock

acquire(*args)
調用底層lock對象的acquire()方法獲取鎖

release()
調用底層lock對象的release()方法釋放鎖

wait(timeout=None)
釋放鎖並阻塞當前線程直到被另一個線程調用notify()或notify_all()喚醒，或者達到設置的timeout時間，任意一種狀況都將從新獲取鎖並返回。
只能由已獲取到鎖的線程調用，不然拋出RuntimeError異常。

3.2版本前該方法始終返回None，3.2版本開始除非超時會返回False，其餘狀況都返回True

wait_for(predicate, timeout=None)
阻塞當前線程直到可調用對象predicate返回值爲True或bool()判斷爲True。

wait_for方法將不斷調用wait()方法直到超時或知足predicate返回值爲True或bool()判斷爲True。

返回值爲最後一次執行predicate的返回值，若是超時返回False。

只能由已獲取到鎖的線程調用，不然拋出RuntimeError異常。

notify(n=1)
喚醒wait()或wait_for()狀態下的某個線程。只能由已獲取到鎖的線程調用，不然拋出RuntimeError異常。

notify_all()
喚醒wait()或wait_for()狀態下的全部線程。只能由已獲取到鎖的線程調用，不然拋出RuntimeError異常。

notify()和notify_all()並不釋放鎖。意思是調用wait()方法的線程不會當即返回，須要等到調用notify()和notify_all()的線程釋放鎖以後才返回。

# 生產者-消費者模式中Condition的用法

# 消費者：
with cv:
    while not an_item_is_available():
        cv.wait()
    get_an_available_item()

# 生產者：
with cv:
    make_an_item_available()
    cv.notify()

# 消費者(使用wait_for改進)：
with cv:
    cv.wait_for(an_item_is_available)
    get_an_available_item()

Semaphore Objects

信號量對象管理一個內部計數器，隨着調用acquire()減1，release()調用加1，但必定不會小於0。當調用acquire()時若是計數器等於0將會阻塞線程直到某個線程調用release()方法。支持上下文管理器協議

class threading.Semaphore(value=1)

指定初始計數器的信號量，每調用一次release()加1，每調用一次acquire()減1。

acquire(blocking=True, timeout=None)
獲取信號量。

使用默認參數調用時：

1. 若是計數器大於0，減1並當即返回True
2. 若是計數器等於0，阻塞直到某個線程調用release()喚醒，喚醒後計數器減1並返回True

release()
釋放信號量。

class threading.BoundedSemaphore(value=1)

邊界信號量，計數器值不能超過設置的最大邊界。經常使用於限制資源佔用的場景好比數據庫鏈接。

Event Objects

事件是最簡單的線程間通訊機制。事件對象管理一個內部標誌，調用set()時該標誌爲True，調用clear()時該標誌爲False，調用wait()時線程阻塞直到標誌爲True

class threading.Event

is_set()
若是事件標誌爲True，返回True

set()
設置事件標誌爲True。將喚醒全部調用了wait()而阻塞的線程。

clear()
重置事件標誌爲False。將阻塞全部調用了wait()的線程。

wait(timeout=None)
阻塞線程直到事件標誌爲True或超時。

Timer Objects

Timer繼承自Thread，表示通過必定時間後要運行的任務。

class threading.Timer(interval, function, args=None, kwargs=None)

建立定時器，在interval時間後運行function任務。

cancel()
終止定時器並結束任務(僅對待執行狀態中的任務有效)。

---

相關文章

Python 多線程： threading.local類