一、多線程

二、多線程編程之threading模塊

2.1、使用threading進行多線程操作有兩種方法：

三、多線程同步之Lock（互斥鎖）

四、多線程同步之Semaphore（信號量）

五、多線程同步之Condition

六、多線程同步之Event

七、線程優(yōu)先級隊列（queue）

八、多線程之線程池pool

九、總結

---

一、多線程

線程（Thread）也稱輕量級進程，是操作系統(tǒng)能夠運行調度的最小單位，被包含在進程中，是進程中的實際運作單位。

線程自身不擁有系統(tǒng)資源，只擁有一些在運行中必不可少的資源，但可與同屬一個進程的其他線程共享所擁有的全部資源

一個線程可以創(chuàng)建和撤銷另一個線程，同一進程中的多個線程之間可以并發(fā)執(zhí)行

• 就緒狀態(tài)指線程具備運行的所有狀態(tài)，邏輯上可以運行，在等待處理機
• 運行狀態(tài)指線程占有處理機正在運行

• 阻塞狀態(tài)指線程在等待一個事件，邏輯上不可執(zhí)行

盡管GIL（全局鎖 ）限制了CPython多線程在CPU密集型任務中的并行性，但Python的多線程在I/O密集型任務中依然能夠發(fā)揮多核CPU的優(yōu)勢，通過在I/O等待期間執(zhí)行其他任務來提升程序的運行效率。

實例1：計算密集型任務-多進程！！多進程！！多進程！！

?from multiprocessing import Processimport os,time?# 計算密集型任務def work():res = 0for i in range(100000000):res *= i?if __name__ == '__main__':l = []print("本機為",os.cpu_count(),"核 CPU")start = time.time()for i in range(4):p = Process(target=work) ? ?# 多進程l.append(p)p.start()for p in l:p.join()stop = time.time()print("計算密集型任務，多進程耗時 %s" % (stop - start))'''本機為 8 核 CPU計算密集型任務，多進程耗時 5.117187976837158'''

實例1：計算密集型任務-多線程！！多線程！！多線程！！

?import os,timefrom threading import Thread?# 計算密集型任務def work():res = 0for i in range(100000000):res *= i?if __name__ == '__main__':l = []print("本機為",os.cpu_count(),"核 CPU")start = time.time()for i in range(4):p = Thread(target=work) ? ?# 多線程l.append(p)p.start()for p in l:p.join()stop = time.time()print("計算密集型任務，多線程耗時 %s" % (stop - start))'''本機為 8 核 CPU計算密集型任務，多線程耗時 14.287675857543945'''

實例2：I/O密集型任務-多進程！！多進程！！多進程！！

?from multiprocessing import Processimport os,time?# I/O密集型任務def work():time.sleep(2)print("===>",file=open("tmp.txt","w"))??if __name__ == '__main__':l = []print("本機為", os.cpu_count(), "核 CPU")start = time.time()for i in range(400):p = Process(target=work) ?# 多進程l.append(p)p.start()for p in l:p.join()stop = time.time()print("I/O密集型任務，多進程耗時 %s" % (stop - start))'''本機為 8 核 CPUI/O密集型任務，多進程耗時 11.03010869026184'''

實例2：I/O密集型任務-多線程！！多線程！！多線程！！

?import os,timefrom threading import Thread?# I/O密集型任務def work():time.sleep(2)print("===>",file=open("tmp.txt","w"))??if __name__ == '__main__':l = []print("本機為", os.cpu_count(), "核 CPU")start = time.time()for i in range(400):p = Thread(target=work) ?# 多線程l.append(p)p.start()for p in l:p.join()stop = time.time()print("I/O密集型任務，多線程耗時 %s" % (stop - start))'''本機為 8 核 CPUI/O密集型任務，多線程耗時 2.0814177989959717'''

結論：在Python中，對于密集型任務，多進程占優(yōu)勢；對于I/O密集型任務，多線程占優(yōu)勢。

二、多線程編程之threading模塊

Python提供多線程編程的模塊有兩個：thread和threading。thread模塊提供低級別的基本功能來支持，提供簡單的鎖來確保同步（不推薦）。threading模塊對_thread進行了封裝，提供了更高級別，功能更強。

2.1、使用threading進行多線程操作有兩種方法：

方法一：創(chuàng)建threading.Thread類的實例，調用其start()方法

?import timeimport threading?def task_thread(counter):print(f'線程名稱：{threading.current_thread().name} 參數：{counter} 開始時間：{time.strftime("%Y-%m-%d %H:%M:%S")}')num = counterwhile num:time.sleep(3)num -= 1print(f'線程名稱：{threading.current_thread().name} 參數：{counter} 結束時間：{time.strftime("%Y-%m-%d %H:%M:%S")}')?if __name__ == '__main__':print(f'主線程開始時間:{time.strftime("%Y-%m-%d %H:%M:%S")}')?# 初始化三個線程，傳遞不同的參數t1 = threading.Thread(target=task_thread,args=(3,))t2 = threading.Thread(target=task_thread,args=(2,))t3 = threading.Thread(target=task_thread,args=(1,))# 開啟三個線程t1.start();t2.start();t3.start()# 等待運行結束t1.join();t2.join();t3.join()?print(f'主線程結束時間:{time.strftime("%Y-%m-%d %H:%M:%S")}')'''程序實例化了三個Thread類的實例，并任務函數傳遞不同的參數，使他們運行不同的時間后結束，start()方法開啟線程，join()方法阻塞主線程，等待當前線程運行結束。'''

方法二：繼承Thread類，在子類中重寫run()和init()方法*（了解---）

?import timeimport threading?class MyThread(threading.Thread):def __init__(self,counter):super().__init__()self.counter = counter?def run(self):print(f'線程名稱：{threading.current_thread().name} 參數：{self.counter} 開始時間：{time.strftime("%Y-%m-%d %H:%M:%S")}')counter = self.counterwhile counter:time.sleep(3)counter -= 1print(f'線程名稱：{threading.current_thread().name} 參數：{self.counter} 結束時間：{time.strftime("%Y-%m-%d %H:%M:%S")}')if __name__ == '__main__':print(f'主線程開始時間:{time.strftime("%Y-%m-%d %H:%M:%S")}')?# 初始化三個線程，傳遞不同的參數t1 = MyThread(3)t2 = MyThread(2)t3 = MyThread(1)# 開啟三個線程t1.start();t2.start();t3.start()# 等待運行結束t1.join();t2.join();t3.join()?print(f'主線程結束時間:{time.strftime("%Y-%m-%d %H:%M:%S")}')

如果繼承Thread類，想要調用外部函數：

?import timeimport threading??def task_thread(counter):print(f'線程名稱：{threading.current_thread().name} 參數：{counter} 開始時間：{time.strftime("%Y-%m-%d %H:%M:%S")}')num = counterwhile num:time.sleep(3)num -= 1print(f'線程名稱：{threading.current_thread().name} 參數：{counter} 結束時間：{time.strftime("%Y-%m-%d %H:%M:%S")}')??class MyThread(threading.Thread):def __init__(self,target,args):super().__init__()self.args = argsself.target = target?def run(self):self.target(*self.args)?if __name__ == '__main__':print(f'主線程開始時間:{time.strftime("%Y-%m-%d %H:%M:%S")}')?# 初始化三個線程，傳遞不同的參數t1 = MyThread(target=task_thread,args=(3,))t2 = MyThread(target=task_thread,args=(2,))t3 = MyThread(target=task_thread,args=(1,))# 開啟三個線程t1.start();t2.start();t3.start()# 等待運行結束t1.join();t2.join();t3.join()?print(f'主線程結束時間:{time.strftime("%Y-%m-%d %H:%M:%S")}')

三、多線程同步之Lock（互斥鎖）

如果多個線程共同對某個數據修改，則可能出現不可預料的結果，這個時候就需要使用互斥鎖來進行同步。例如：在三個線程對共同變量num進行100萬次加減操作后，其num的結果不為0

不加鎖的意外情況：

?import time,threading?num = 0?def task_thread(n):global ?numfor i in range(1000000):num = num + nnum = num - n?t1 = threading.Thread(target=task_thread,args=(6,))t2 = threading.Thread(target=task_thread,args=(17,))t3 = threading.Thread(target=task_thread,args=(11,))t1.start();t2.start();t3.start()t1.join();t2.join();t3.join()print(num)'''6'''

使用互斥鎖對多個線程進行同步，限制當一個線程正在訪問數據時，其他只能等待，直到前一線程釋放鎖。

使用threading.Thread對象的Lock和Rlock可以實現簡單的線程同步，都有acquire和release方法。

?# 加互斥鎖后運行結果始終一致import time,threading?num = 0lock = threading.Lock()def task_thread(n):global num# 獲取鎖，用于線程同步lock.acquire()for i in range(1000000):num = num + nnum = num - n# 釋放鎖，開啟下一個線程lock.release()??t1 = threading.Thread(target=task_thread,args=(6,))t2 = threading.Thread(target=task_thread,args=(17,))t3 = threading.Thread(target=task_thread,args=(11,))t1.start();t2.start();t3.start()t1.join();t2.join();t3.join()print(num)

四、多線程同步之Semaphore（信號量）

互斥鎖是只允許一個線程訪問共享數據，而信號量是同時運行一定數量的線程訪問共享數據，比如銀行柜臺有5個窗口，運行同時有5個人辦理業(yè)務，后面的人只能等待其完成。

?# 使用信號量控制并發(fā)import threadingimport time?# 銀行柜臺窗口數量NUM_WINDOWS = 5?# 用于控制窗口訪問的信號量semaphore = threading.Semaphore(NUM_WINDOWS)?# 客戶辦理業(yè)務的函數def customer(name, service_time):# 嘗試獲取信號量semaphore.acquire()print(f"{time.ctime()}: {name} 開始辦理業(yè)務")time.sleep(service_time) ?# 模擬辦理業(yè)務的時間print(f"{time.ctime()}: {name} 辦理業(yè)務完成")semaphore.release() ?# 釋放信號量?# 創(chuàng)建客戶線程列表customers = []for i in range(12):name = f"客戶{i+1}"service_time = 3 ?# 假設每個客戶辦理業(yè)務需要1秒時間thread = threading.Thread(target=customer, args=(name, service_time))customers.append(thread)?# 啟動所有客戶線程for customer in customers:customer.start()?# 等待所有客戶完成業(yè)務for customer in customers:customer.join()?print("所有客戶都辦理完業(yè)務了。")

上述代碼實現了同一時刻只有5個線程獲得資源運行

五、多線程同步之Condition

條件對象Condition能讓一個線程A停下來，等待其他線程B，線程B滿足了某個條件后通知線程A繼續(xù)運行。步驟：

線程首先獲取一個條件變量鎖，如果條件不足，則該線程等待（wait）并釋放條件變量鎖；如果條件滿足，就繼續(xù)執(zhí)行線程，執(zhí)行完成后可以通知（notify）其他狀態(tài)為wait的線程執(zhí)行。其他處于wait狀態(tài)的線程接到通知后會重新判斷條件來確定是否繼續(xù)執(zhí)行

?import threading?class Boy(threading.Thread):def __init__(self,cond,name):super(Boy,self).__init__()self.cond = condself.name = name?def run(self):self.cond.acquire()print(self.name + ":嫁給我吧！？")self.cond.notify() ?# 喚醒一個掛起的線程，讓hanmeimei表態(tài)self.cond.wait() ? ?# 釋放內部所占用的鎖，同時線程被掛起，直至接收到通知被喚醒或超時，等待heimeimei回答print(self.name + ": 我單膝下跪，送上戒指！")self.cond.notify()self.cond.wait()print(self.name + ": lI太太，你的選擇太明智了。")self.cond.release()class Girl(threading.Thread):def __init__(self,cond,name):super(Girl,self).__init__()self.cond = condself.name = name?def run(self):self.cond.acquire()self.cond.wait() ? ?# 等待Lilei求婚print(self.name + ": 沒有情調，不夠浪漫，不答應")self.cond.notify()self.cond.wait()print(self.name + ": 好吧，答應你了")self.cond.notify()self.cond.release()?cond = threading.Condition()boy = Boy(cond,"LiLei")girl = Girl(cond,"HanMeiMei")girl.start()boy.start()

六、多線程同步之Event

事件用于線程之間的通信。一個線程發(fā)出一個信號，其他一個或多個線程等待，調用Event對象的wait方法，線程則會阻塞等待，直到別的線程set之后才會被喚醒。與上述類似

?import threadingimport time??class Boy(threading.Thread):def __init__(self,cond,name):super(Boy,self).__init__()self.cond = condself.name = name?def run(self):print(self.name + ":嫁給我吧！？")self.cond.set() ?# 喚醒一個掛起的線程，讓hanmeimei表態(tài)time.sleep(0.5)self.cond.wait() ? ?# 釋放內部所占用的鎖，同時線程被掛起，直至接收到通知被喚醒或超時，等待heimeimei回答print(self.name + ": 我單膝下跪，送上戒指！")self.cond.set()time.sleep(0.5)self.cond.wait()self.cond.clear()print(self.name + ": lI太太，你的選擇太明智了。")class Girl(threading.Thread):def __init__(self,cond,name):super(Girl,self).__init__()self.cond = condself.name = name?def run(self):self.cond.wait() ? ?# 等待Lilei求婚self.cond.clear()print(self.name + ": 沒有情調，不夠浪漫，不答應")self.cond.set()time.sleep(0.5)self.cond.wait()print(self.name + ": 好吧，答應你了")self.cond.set()?cond = threading.Event()boy = Boy(cond,"LiLei")girl = Girl(cond,"HanMeiMei")girl.start()boy.start()

七、線程優(yōu)先級隊列（queue）

Python的queue模塊中提供了同步的、線程安全的隊列類，包括先進先出隊列Queue、后進先出隊列LifoQueue和優(yōu)先級隊列PriorityQueue。這些隊列都實現了鎖原語，可以直接使用來實現線程之間的同步。

?'''有一小冰箱用來存放冷飲，假如只能放5瓶冷飲，A不停地放冷飲，B不停的取冷飲，A和B的放取速度不一致，如何保持同步呢？'''import threading,timeimport queue?# 先進先出q = queue.Queue(maxsize=5)# q = LifoQuere(maxsize=3)# q = PriorityQueue(maxsize=3)?def ProducerA():count = 1while True:q.put(f"冷飲 {count}")print(f"{time.strftime('%H:%M:%S')} A 放入：[冷飲 {count}]")count += 1time.sleep(2)?def ConsumerB():while True:print(f"{time.strftime('%H:%M:%S')} B 取出：{q.get()}")time.sleep(5)?p = ?threading.Thread(target=ProducerA)c = threading.Thread(target=ConsumerB)p.start()c.start()

八、多線程之線程池pool

將 任務添加到線程池中，線程池會自動指定一個空閑的線程去執(zhí)行任務，當超過線程池的最大線程數時，任務需要等待有新的空閑線程后才會被執(zhí)行。

使用threading模塊及queue模塊定制線程池，可以使用multiprocessing。

• from multiprocessing import Pool導入的是進程池

• from multiprocessing.dummy import Pool導入的是線程池

?'''模擬一個耗時2秒的任務，比較其順序執(zhí)行5次和線程池（并發(fā)數為5）執(zhí)行的耗時。'''from multiprocessing.dummy import Pool as ThreadPoolimport time?def fun(n):time.sleep(2)?start = time.time()for i in range(5):fun(i)print("單線程順序執(zhí)行耗時：",time.time() - start)?start2 = time.time()# 開8個worker，沒有參數時默認是cpu的核心數pool = ThreadPool(processes=5)# 在線程中執(zhí)行urllib2.urlopen(url)并返回執(zhí)行結果results2 = pool.map(fun,range(5))pool.close()pool.join()print("線程池（5）并發(fā)執(zhí)行耗時：",time.time() - start2)'''單線程順序執(zhí)行耗時： 10.041245937347412線程池（5）并發(fā)執(zhí)行耗時： 2.0453202724456787'''

九、總結

• Python多線程適合用在I/O密集型任務中。
• 對于I/O密集型任務來說，較少時間用在CPU計算上，較多時間用在I/O上，如文件讀寫、web請求、數據庫請求等

• 對于計算密集型任務，應該使用多進程