Python3快速入門（九）——Python3併發編程
Python3快速入門（九）——Python3併發編程
1、Python線程模塊
一、線程簡介
一個標準的線程由線程ID，當前指令指針(PC），寄存器集合和堆棧組成。線程是進程中的一個實體，是被系統獨立調度和分派的基本單位，線程自己不擁有系統資源，與進程內的其它線程共享進程的全部資源。一個進程中至少有一個線程，並做爲程序的入口，即主線程，其它線程稱爲工做線程。
多線程，是指從軟件或者硬件上實現多個線程併發執行的技術。支持多線程能力的計算機因有硬件支持而可以在同一時間執行多個線程，進而提高總體處理性能。數據庫
二、線程狀態
線程有就緒、阻塞、運行三種基本狀態。就緒狀態是指線程具有運行的全部條件，在等待CPU執行； 運行狀態是指線程佔有CPU正在運行； 阻塞狀態是指線程在等待一個事件，邏輯上不可執行。
三種狀態的相互轉化以下圖所示：
編程
二、threading線程模塊
Python3 經過_thread 和 threading兩個模塊提供對線程的支持。_thread提供了低級別的、原始的線程以及簡單鎖，相比於 threading 模塊的功能比較有限，是對已經廢棄的thread模塊的兼容性支持方案。
threading 模塊除了包含_thread模塊中的全部方法外，還提供的以下方法：
threading.currentThread(): 返回當前的線程變量。
threading.enumerate(): 返回一個包含正在運行的線程的list。正在運行指線程啓動後、結束前，不包括啓動前和終止後的線程。
threading.activeCount(): 返回正在運行線程的數量，與len(threading.enumerate())有相同的結果。
Thread類提供方法以下:
run(): 用以表示線程活動的方法。
start():啓動線程活動。
join([time]): 等待至線程停止。阻塞調用線程直至線程的join() 方法被調用停止-正常退出或者拋出未處理的異常-或者是可選的超時發生。
isAlive(): 返回線程是否活動的。
getName(): 返回線程名。
setName(): 設置線程名。安全
三、multiprocessing模塊
multiprocessing模塊是跨平臺版本的多進程模塊，提供了一個Process類表明一個進程對象。建立子進程時，只須要傳入一個執行函數和函數的參數，建立一個Process實例。Process（self,group=None,target=None,name=None,args=(),kwargs=()）
group參數未使用，值始終爲None。
target表示調用的對象，子進程要執行的任務。
name能夠爲子進程命名。
args指定傳結target函數的位置參數，是一個元組形式，必須有逗號，如：args=(‘monicx’,)
kwargs指定傳結target函數的關鍵字參數，是一個字典，如kwargs={‘name’:‘monicx’,‘age’:18}
Process方法以下：
start():啓動進程，並調用子進程的run()方法。
run():進程啓動進運行的方法，在run內調用target指定的函數，子進程類中必定要實現run方法。
terminate():強制終止進程，不會進行任何清理操做，若是進程建立了子進程，子進程會變成殭屍進程；若是進程還保存了一個鎖，則不會釋放進程鎖，進而致使死鎖。
is_alive():判斷進程是不是「活着」的狀態。
join(timeout):讓主進程程等待某一子進程結束，才繼續執行主進程。timeout是可選的超時時間，超過一個時間主進程就不等待。多線程
四、全局解釋鎖GIL
Python並不支持真正意義上的多線程。Python中提供了多線程模塊，但若是想經過多線程提升代碼的速度，並不推薦使用多線程模塊。Python中有一個全局鎖Global Interpreter Lock（GIL），全局鎖會確保任什麼時候候多個線程中只有一個會被執行。線程的執行速度很是快，會誤覺得線程是並行執行的，但實際上都是輪流執行。通過GIL處理後，會增長線程執行的開銷。
全局鎖 GIL(Global interpreter lock) 並非 Python 的特性，而是在實現 Python 解析器(CPython)時所引入的一個概念。Python有CPython，PyPy，Psyco 等不一樣的 Python 執行環境，其中 JPython 沒有GIL。CPython 是大部分環境下默認的 Python 執行環境，GIL 並非 Python 的特性，Python 徹底能夠不依賴於 GIL。
GIL 限制了同一時刻只能有一個線程運行，沒法發揮多核 CPU 的優點。GIL 本質是互斥鎖，都是將併發運行變成串行，以此來控制同一時間內共享數據只能被一個任務所修改，進而保證數據安全。在一個 Python 的進程內，不只有主線程或者由主線程開啓的其它線程，還有解釋器開啓的垃圾回收等解釋器級別的線程。進程內，全部數據都是共享的，代碼做爲一種數據也會被全部線程共享，多個線程先訪問到解釋器的代碼，即拿到執行權限，而後將 target 的代碼交給解釋器的代碼去執行，解釋器的代碼是全部線程共享的，因此垃圾回收線程也可能訪問到解釋器的代碼而去執行，所以爲了保證數據安全須要加鎖處理，即 GIL。
因爲GIL 的存在，同一時刻同一進程中只有一個線程被執行。多核 CPU能夠並行完成計算，所以多核能夠提高計算性能，但 CPU 一旦遇到 I/O 阻塞，仍然須要等待，因此多核CPU對 I/O 密集型任務提高不明顯。根據執行任務是計算密集型仍是I/O 密集型，不一樣場景使用不一樣的方法，對於計算密集型任務，多進程佔優點，對於 I/O 密集型任務，多線程佔優點。
計算密集型任務-多進程方案：併發
# -*- coding:utf-8 -*-
from multiprocessing import Process
import os
import time
def work():
result = 0
for x in range(100000000):
result *= x
if __name__ == "__main__":
processes = []
print("CPU: ", os.cpu_count())
start = time.time()
for i in range(4):
p = Process(target=work)
processes.append(p)
p.start()
for p in processes:
p.join()
end = time.time()
print("計算密集型任務，多進程耗時 %s" % (end - start))
# output:
# CPU: 4
# 計算密集型任務，多進程耗時 9.485123872756958
計算密集型任務-多線程方案：app
# -*- coding:utf-8 -*-
from threading import Thread
import os, time
def work():
res = 0
for x in range(100000000):
res *= x
if __name__ == "__main__":
threads = []
print("CPU: ",os.cpu_count())
start = time.time()
for i in range(4):
thread = Thread(target=work) # 多進程
threads.append(thread)
thread.start()
for thread in threads:
thread.join()
end = time.time()
print("計算密集型任務，多線程耗時 %s" % (end - start))
# output:
# CPU: 4
# 計算密集型任務，多線程耗時 18.434288501739502
IO密集型任務-多進程方案：dom
# -*- coding:utf-8 -*-
from multiprocessing import Process
import os, time
def work():
time.sleep(2)
print("hello,Python----------------------------------------------------", file=open("tmp.txt", "w"))
if __name__ == "__main__":
processes = []
print("CPU: ", os.cpu_count())
start = time.time()
for i in range(400):
p = Process(target=work) # 多進程
processes.append(p)
p.start()
for p in processes:
p.join()
stop = time.time()
print("I/0密集型任務，多進程耗時 %s" % (stop - start))
# output:
# CPU: 4
# I/0密集型任務，多進程耗時 2.8894519805908203
IO密集型任務-多線程方案：異步
# -*- coding:utf-8 -*-
from threading import Thread
import os, time
def work():
time.sleep(2)
print("hello,Python----------------------------------------------------", file=open("tmp.txt", "w"))
if __name__ == "__main__":
threads = []
print("CPU: ", os.cpu_count())
start = time.time()
for x in range(400):
thread = Thread(target=work)
threads.append(thread)
thread.start()
for thread in threads:
thread.join()
end = time.time()
print("IO密集型任務，多線程耗時 %s" % (end - start))
# output:
# CPU: 4
# IO密集型任務，多線程耗時 2.044438362121582
2、建立線程
一、threading.Thread實例化
threading.Thread構造函數以下：ide
def __init__(self, group=None, target=None, name=None,
args=(), kwargs=None, *, daemon=None):
建立 threading.Thread 實例，調用其 start() 方法。函數
# -*- coding:utf-8 -*-
import time
import threading
def work_task(counter):
print("%s %s" % (threading.current_thread().name, time.ctime(time.time())))
n = counter;
while n > 0:
time.sleep(1)
n -= 1
if __name__ == "__main__":
print("main thread start：", time.strftime("%Y-%m-%d %H:%M:%S"))
threads = []
for x in range(10):
thread = threading.Thread(target=work_task, args=(x, ))
threads.append(thread)
for thread in threads:
thread.start()
for thread in threads:
thread.join()
print("main thread end：", time.strftime("%Y-%m-%d %H:%M:%S"))
# output:
# main thread start： 2019-07-03 21:49:58
# Thread-1 Wed Jul 3 21:49:58 2019
# Thread-2 Wed Jul 3 21:49:58 2019
# Thread-3 Wed Jul 3 21:49:58 2019
# Thread-4 Wed Jul 3 21:49:58 2019
# Thread-5 Wed Jul 3 21:49:58 2019
# Thread-6 Wed Jul 3 21:49:58 2019
# Thread-7 Wed Jul 3 21:49:58 2019
# Thread-8 Wed Jul 3 21:49:58 2019
# Thread-9 Wed Jul 3 21:49:58 2019
# Thread-10 Wed Jul 3 21:49:58 2019
# main thread end： 2019-07-03 21:50:07
二、threading.Thread子線程
能夠經過直接從 threading.Thread類繼承建立一個新的子類，在子類中重寫 run() 和 init() 方法，實例化後調用 start() 方法啓動新線程，start函數內部會調用線程的 run() 方法。
# -*- coding:utf-8 -*-
import threading
import time
class WorkThread(threading.Thread):
def __init__(self, thread_id, name):
threading.Thread.__init__(self)
self.thread_id = thread_id
self.name = name
def run(self):
print("start thread: ", self.name)
work(self.name, self.thread_id)
print("end thread: ", self.name)
def work(thread_name, thread_id):
print("%s %s %s" % (thread_name, thread_id, time.ctime(time.time())))
i = 0;
while i < 2:
i += 1
time.sleep(1)
if __name__ == '__main__':
thread1 = WorkThread(1, "Thread1")
thread2 = WorkThread(2, "Thread2")
thread1.start()
thread2.start()
thread1.join()
thread2.join()
print("exit main thread")
# output:
# start thread: Thread1
# Thread1 1 Tue Jul 2 20:39:42 2019
# start thread: Thread2
# Thread2 2 Tue Jul 2 20:39:42 2019
# end thread: end thread: Thread1
# Thread2
# exit main thread
若是須要從外部傳入函數，能夠將傳入參數做爲子線程實例屬性，在run實例方法內進行調用。
self.target = target
self.args = args
# -*- coding:utf-8 -*-
import threading
import time
class WorkThread(threading.Thread):
def __init__(self, target, args):
threading.Thread.__init__(self)
self.target = target
self.args = args
def run(self):
print("start thread: ", self.name)
self.target(*self.args)
print("end thread: ", self.name)
def work_task(counter):
time.sleep(1)
print("%s %s" % (threading.currentThread().name, time.ctime(time.time())))
i = counter;
while i > 0:
i -= 1
if __name__ == '__main__':
print("main thread start：", time.strftime("%Y-%m-%d %H:%M:%S"))
threads = []
for x in range(10):
thread = threading.Thread(target=work_task, args=(x,))
threads.append(thread)
for thread in threads:
thread.start()
for thread in threads:
thread.join()
print("main thread end：", time.strftime("%Y-%m-%d %H:%M:%S"))
# output:
# main thread start： 2019-07-03 22:02:32
# Thread-1 Wed Jul 3 22:02:33 2019Thread-5 Wed Jul 3 22:02:33 2019
# Thread-2 Wed Jul 3 22:02:33 2019
# Thread-3 Wed Jul 3 22:02:33 2019
# Thread-4 Wed Jul 3 22:02:33 2019
#
# Thread-7 Wed Jul 3 22:02:33 2019Thread-6 Wed Jul 3 22:02:33 2019
# Thread-10 Wed Jul 3 22:02:33 2019
# Thread-8 Wed Jul 3 22:02:33 2019
#
# Thread-9 Wed Jul 3 22:02:33 2019
# main thread end： 2019-07-03 22:02:33
三、start與run
import threading
import time
def work_task(counter):
n = counter
while n > 0:
n -= 1
print("thread name： %s， id: %s" % (threading.currentThread().name, threading.currentThread().ident))
if __name__ == "__main__":
print("main thread start")
thread1 = threading.Thread(target=work_task, args=(5,))
thread2 = threading.Thread(target=work_task, args=(5,))
thread1.start()
thread2.start()
print("main thread end")
# output:
# main thread start
# thread name： Thread-1， id: 139926959064832thread name： Thread-2， id: 139926880384768main thread end
#
#
# thread name： Thread-1， id: 139926959064832
# thread name： Thread-2， id: 139926880384768thread name： Thread-1， id: 139926959064832
#
# thread name： Thread-1， id: 139926959064832
# thread name： Thread-2， id: 139926880384768thread name： Thread-1， id: 139926959064832
#
# thread name： Thread-2， id: 139926880384768
# thread name： Thread-2， id: 139926880384768
使用start()方法啓動了兩個新的子線程並交替運行，每一個子進程ID也不一樣，啓動的線程名是定義線程對象時設置的name="xxxx"值，若是沒有設置name參數值，則會打印系統分配的Thread-x名稱。
import threading
import time
def work_task(counter):
n = counter
while n > 0:
n -= 1
print("thread name： %s， id: %s" % (threading.currentThread().name, threading.currentThread().ident))
if __name__ == "__main__":
print("main thread start")
thread1 = threading.Thread(target=work_task, args=(5,))
thread2 = threading.Thread(target=work_task, args=(5,))
thread1.run()
thread2.run()
print("main thread end")
# output:
# main thread start
# thread name： MainThread， id: 140683421988672
# thread name： MainThread， id: 140683421988672
# thread name： MainThread， id: 140683421988672
# thread name： MainThread， id: 140683421988672
# thread name： MainThread， id: 140683421988672
# thread name： MainThread， id: 140683421988672
# thread name： MainThread， id: 140683421988672
# thread name： MainThread， id: 140683421988672
# thread name： MainThread， id: 140683421988672
# thread name： MainThread， id: 140683421988672
# main thread end
用run()方法啓動線程，打印的線程名是MainThread，即主線程。兩個線程都用run()方法啓動，但卻先運行thread1.run()，運行完後才按順序運行thread2.run()，兩個線程都工做在主線程，沒有啓動新線程，所以，run()方法僅是普通函數調用。
四、join方法
當一個進程啓動後，會默認產生一個主線程，由於線程是程序執行流的最小單元，當設置多線程時，主線程會建立多個子線程。在Python中，默認狀況下主線程執行完本身的任務後，就會退出，此時子線程會繼續執行本身的任務，直到本身的任務結束。
import threading
import time
def work_task(counter):
n = counter
while n > 0:
n -= 1
print("thread name： %s， id: %s" % (threading.currentThread().name, threading.currentThread().ident))
if __name__ == "__main__":
print("main thread start")
threads = []
for x in range(5):
thread = threading.Thread(target=work_task, args=(5,))
threads.append(thread)
for thread in threads:
thread.start()
print("main thread end")
# output:
# main thread start
# thread name： Thread-1， id: 140306042726144thread name： Thread-2， id: 140306034333440
# thread name： Thread-2， id: 140306034333440
# thread name： Thread-2， id: 140306034333440
# thread name： Thread-2， id: 140306034333440
# thread name： Thread-2， id: 140306034333440
# thread name： Thread-3， id: 140306025940736
#
# thread name： Thread-3， id: 140306025940736
# thread name： Thread-3， id: 140306025940736
# thread name： Thread-1， id: 140306042726144thread name： Thread-3， id: 140306025940736
# thread name： Thread-3， id: 140306025940736
#
# thread name： Thread-1， id: 140306042726144
# thread name： Thread-1， id: 140306042726144
# thread name： Thread-1， id: 140306042726144
# thread name： Thread-4， id: 140306034333440
# thread name： Thread-4， id: 140306034333440
# thread name： Thread-5， id: 140306042726144thread name： Thread-4， id: 140306034333440
# main thread endthread name： Thread-4， id: 140306034333440
# thread name： Thread-4， id: 140306034333440
#
# thread name： Thread-5， id: 140306042726144
#
# thread name： Thread-5， id: 140306042726144
# thread name： Thread-5， id: 140306042726144
# thread name： Thread-5， id: 140306042726144
當使用setDaemon(True)方法，設置子線程爲守護線程時，主線程一旦執行結束，則所有線程所有被終止執行，可能會出現子線程的任務尚未徹底執行結束，就被迫中止。設置setDaemon必須在啓動子線程前進行設置。
import threading
import time
def work_task(counter):
n = counter
time.sleep(1)
while n > 0:
n -= 1
print("thread name： %s， id: %s" % (threading.currentThread().name, threading.currentThread().ident))
if __name__ == "__main__":
print("main thread start")
threads = []
for x in range(5):
thread = threading.Thread(target=work_task, args=(5,))
threads.append(thread)
for thread in threads:
thread.setDaemon(True)
thread.start()
print("main thread end")
# output:
# main thread start
# main thread end
join方法用於讓主線程等待子線執行完並返回結果後，再執行主線程剩下的內容，子線程不執行完，主線程就一直等待狀態。
import threading
import time
def work_task(counter):
n = counter
time.sleep(1)
while n > 0:
n -= 1
print("thread name： %s" % threading.currentThread().name)
if __name__ == "__main__":
print("main thread start")
threads = []
for x in range(5):
thread = threading.Thread(target=work_task, args=(5,))
threads.append(thread)
for thread in threads:
thread.setDaemon(True)
thread.start()
thread.join()
print("main thread end")
# output:
# main thread start
# thread name： Thread-1
# thread name： Thread-1
# thread name： Thread-1
# thread name： Thread-1
# thread name： Thread-1
# thread name： Thread-2
# thread name： Thread-2
# thread name： Thread-2
# thread name： Thread-2
# thread name： Thread-2
# thread name： Thread-3
# thread name： Thread-3
# thread name： Thread-3
# thread name： Thread-3
# thread name： Thread-3
# thread name： Thread-4
# thread name： Thread-4
# thread name： Thread-4
# thread name： Thread-4
# thread name： Thread-4
# thread name： Thread-5
# thread name： Thread-5
# thread name： Thread-5
# thread name： Thread-5
# thread name： Thread-5
# main thread end
join有一個timeout參數，當設置守護線程時，主線程對子線程等待timeout時間，給每一個子線程一個timeout時間，讓子線程執行，時間一到，無論任務有沒有完成，直接殺死。 若是有多個子線程，所有的等待時間是每一個子線程timeout的累加和。
import threading
import time
def work_task(counter):
print("thread name： %s work task start" % threading.currentThread().name)
n = counter
time.sleep(4)
while n > 0:
n -= 1
else:
print("thread name： %s work task end" % threading.currentThread().name)
if __name__ == "__main__":
print("main thread start")
threads = []
for x in range(5):
thread = threading.Thread(target=work_task, args=(5,))
threads.append(thread)
for x in range(5):
threads[x].setDaemon(True)
threads[x].start()
threads[x].join(1)
print("main thread end")
# output:
# main thread start
# thread name： Thread-1 work task start
# thread name： Thread-2 work task start
# thread name： Thread-3 work task start
# thread name： Thread-4 work task start
# thread name： Thread-5 work task start
# thread name： Thread-1 work task end
# main thread end
沒有設置守護線程時，主線程將會等待timeout的累加和的一段時間，時間一到，主線程結束，但並無殺死子線程，子線程依然能夠繼續執行，直到子線程所有結束，程序退出。
import threading
import time
def work_task(counter):
print("thread name： %s work task start" % threading.currentThread().name)
n = counter
time.sleep(4)
while n > 0:
n -= 1
else:
print("thread name： %s work task end" % threading.currentThread().name)
if __name__ == "__main__":
print("main thread start")
threads = []
for x in range(5):
thread = threading.Thread(target=work_task, args=(5,))
threads.append(thread)
for x in range(5):
threads[x].start()
threads[x].join(1)
print("main thread end")
# output:
# main thread start
# thread name： Thread-1 work task start
# thread name： Thread-2 work task start
# thread name： Thread-3 work task start
# thread name： Thread-4 work task start
# thread name： Thread-5 work task start
# thread name： Thread-1 work task end
# main thread end
# thread name： Thread-2 work task end
# thread name： Thread-3 work task end
# thread name： Thread-4 work task end
# thread name： Thread-5 work task end
3、線程同步
若是多個線程共同對某個數據修改，則可能出現不可預料的結果，爲了保證數據的正確性，須要對多個線程進行同步。
一、互斥鎖
threading.Thread 類的 Lock 鎖和 Rlock 鎖能夠實現簡單線程同步，Lock 鎖和 Rlock 鎖都有 acquire 方法和 release 方法，每次只容許一個線程操做的數據須要將其操做放到 acquire 和 release 方法之間。
# -*- coding:utf-8 -*-
import threading
import time
class WorkThread(threading.Thread):
def __init__(self, thread_id, name):
threading.Thread.__init__(self)
self.thread_id = thread_id
self.name = name
def run(self):
thread_locker.acquire()
print("start thread: ", self.name)
work(self.name, self.thread_id)
print("end thread: ", self.name)
thread_locker.release()
def work(thread_name, thread_id):
print("%s %s %s" % (thread_name, thread_id, time.ctime(time.time())))
i = 0;
while i < 2:
i += 1
time.sleep(1)
thread_locker = threading.Lock()
threads = []
if __name__ == '__main__':
thread1 = WorkThread(1, "Thread1")
thread2 = WorkThread(2, "Thread2")
thread1.start()
thread2.start()
threads.append(thread1)
threads.append(thread2)
for t in threads:
t.join()
print("exit main thread")
# output:
# start thread: Thread1
# Thread1 1 Tue Jul 2 20:48:05 2019
# end thread: Thread1
# start thread: Thread2
# Thread2 2 Tue Jul 2 20:48:07 2019
# end thread: Thread2
# exit main thread
二、信號量
互斥鎖同時只容許一個線程訪問共享數據，而信號量同時容許必定數量的線程訪問共享數據，如銀行櫃檯有 5 個窗口，則容許同時有 5 我的辦理業務，後面的人只能等待前面有人辦完業務後才能夠進入櫃檯辦理。
# -*- coding:utf-8 -*-
import threading
import time
semaphore = threading.BoundedSemaphore(5)
threads = []
def do_work(name):
semaphore.acquire()
time.sleep(2)
print(f"{time.strftime('%Y-%m-%d %H:%M:%S')} {threading.currentThread().name} is carrying on business")
semaphore.release()
if __name__ == '__main__':
print("main thread start：", time.strftime("%Y-%m-%d %H:%M:%S"))
for i in range(10):
t = threading.Thread(target=do_work, args=(i,))
threads.append(t)
for thread in threads:
thread.start()
for thread in threads:
thread.join()
print("main thread end：", time.strftime("%Y-%m-%d %H:%M:%S"))
# output:
# main thread start： 2019-07-03 22:31:06
# 2019-07-03 22:31:08 Thread-1 is carrying on business
# 2019-07-03 22:31:08 Thread-3 is carrying on business
# 2019-07-03 22:31:08 Thread-2 is carrying on business
# 2019-07-03 22:31:08 Thread-4 is carrying on business
# 2019-07-03 22:31:08 Thread-5 is carrying on business
# 2019-07-03 22:31:10 Thread-6 is carrying on business
# 2019-07-03 22:31:10 Thread-7 is carrying on business
# 2019-07-03 22:31:10 Thread-9 is carrying on business
# 2019-07-03 22:31:10 Thread-8 is carrying on business
# 2019-07-03 22:31:10 Thread-10 is carrying on business
# main thread end： 2019-07-03 22:31:10
三、條件變量
條件變量能讓一個線程 A 停下來，等待其它線程 B ，線程 B 知足了某個條件後通知（notify）線程 A 繼續運行。線程首先獲取一個條件變量鎖，若是條件不知足，則線程等待（wait）並釋放條件變量鎖；若是條件知足則執行線程，也能夠通知其它狀態爲 wait 的線程。其它處於 wait 狀態的線程接到通知後會從新判斷條件。
import threading
import time
class ThreadA(threading.Thread):
def __init__(self, con, name):
super(ThreadA, self).__init__()
self.cond = con
self.name = name
def run(self):
self.cond.acquire()
print(self.name + ": What can I do for you?")
self.cond.notify()
self.cond.wait()
print(self.name + ": Five yuan.")
self.cond.notify()
self.cond.wait()
print(self.name + ": You are welcome.")
self.cond.release()
class ThreadB(threading.Thread):
def __init__(self, con, name):
super(ThreadB, self).__init__()
self.cond = con
self.name = name
def run(self):
self.cond.acquire()
time.sleep(1)
print(self.name + ": A hot dog, please！")
self.cond.notify()
self.cond.wait()
print(self.name + ": Thanks.")
self.cond.notify()
self.cond.release()
if __name__ == "__main__":
cond = threading.Condition()
thread1 = ThreadA(cond, "ThreadA")
thread2 = ThreadB(cond, "ThreadB")
thread1.start()
thread2.start()
# output:
# ThreadA: What can I do for you?
# ThreadB: A hot dog, please！
# ThreadA: Five yuan.
# ThreadB: Thanks.
# ThreadA: You are welcome.
四、事件
事件用於線程間通訊。一個線程發出一個信號，其它一個或多個線程等待，調用 event 對象的 wait 方法，線程則會阻塞等待，直到其它線程 set 後，纔會被喚醒。
import threading
import time
class ThreadA(threading.Thread):
def __init__(self, _event, name):
super(ThreadA, self).__init__()
self.event = _event
self.name = name
def run(self):
print(self.name + ": What can I do for you?")
self.event.set()
time.sleep(0.5)
self.event.wait()
print(self.name + ": Five yuan.")
self.event.set()
time.sleep(0.5)
self.event.wait()
self.event.clear()
print(self.name + ": You are welcome!")
class ThreadB(threading.Thread):
def __init__(self, _event, name):
super(ThreadB, self).__init__()
self.event = _event
self.name = name
def run(self):
self.event.wait()
self.event.clear()
print(self.name + ": A hot dog, please!")
self.event.set()
time.sleep(0.5)
self.event.wait()
print(self.name + ": Thanks!")
self.event.set()
if __name__ == "__main__":
event = threading.Event()
thread1 = ThreadA(event, "ThreadA")
thread2 = ThreadB(event, "ThreadB")
thread1.start()
thread2.start()
# output:
# ThreadA: What can I do for you?
# ThreadB: A hot dog, please!
# ThreadA: Five yuan.
# ThreadB: Thanks!
# ThreadA: You are welcome!
五、線程優先級隊列
Python 的 Queue 模塊中提供了同步的、線程安全的隊列類，包括FIFO（先入先出)隊列Queue，LIFO（後入先出）隊列LifoQueue，優先級隊列 PriorityQueue。
Queue，LifoQueue，PriorityQueue都實現了鎖原語，可以在多線程中直接使用，可使用隊列來實現線程間的同步。
Queue 模塊中的經常使用方法:
Queue.qsize() 返回隊列的大小
Queue.empty() 若是隊列爲空，返回True，不然返回False
Queue.full() 若是隊列滿，返回True，不然返回False
Queue.get([block[, timeout]])獲取隊列，timeout等待時間
Queue.get_nowait() 至關Queue.get(False)
Queue.put(item) 寫入隊列，timeout等待時間
Queue.put_nowait(item) 至關Queue.put(item, False)
Queue.task_done() 在完成一項工做後，Queue.task_done()函數向任務已經完成的隊列發送一個信號
Queue.join() 阻塞直到隊列爲空，再執行別的操做
# -*- coding:utf-8 -*-
import threading
import time
import queue
exitFlag = 0
class WorkThread(threading.Thread):
def __init__(self, id, name, q):
threading.Thread.__init__(self)
self.thread_id = id
self.name = name
self.queue = q
def run(self):
work(self.name, self.queue)
def work(thread_name, q):
while not exitFlag:
thread_locker.acquire()
if not work_queue.empty():
data = q.get()
print("%s processing %s" % (thread_name, data))
thread_locker.release()
else:
thread_locker.release()
time.sleep(1)
thread_locker = threading.Lock()
thread_list = ["Thread1", "Thread2", "Thread3"]
work_queue = queue.Queue(10)
messages = ["one", "two", "three", "four", "five"]
threads = []
thread_id = 1
if __name__ == '__main__':
# 建立新線程
for name in thread_list:
thread = WorkThread(thread_id, name, work_queue)
thread.start()
threads.append(thread)
thread_id += 1
# 填充隊列
thread_locker.acquire()
for word in messages:
work_queue.put(word)
thread_locker.release()
# 等待隊列清空
while not work_queue.empty():
pass
# 通知線程是時候退出
exitFlag = 1
# 等待全部線程完成
for t in threads:
t.join()
print("exit main thread")
# output:
# Thread1 processing one
# Thread3 processing two
# Thread2 processing three
# Thread1 processing four
# Thread3 processing five
# exit main thread
六、線程死鎖
死鎖是指兩個或兩個以上的進程或線程在執行過程當中，因爭奪資源而形成的一種互相等待的現象。在線程間共享多個資源的時候，若是分別佔有一部分資源而且同時在等待對方的資源，就會形成死鎖。例如數據庫操做時A線程須要B線程的結果進行操做，B線程的須要A線程的結果進行操做，當A，B線程同時在進行操做尚未結果出來時，此時A，B線程將會一直處於等待對方結束的狀態。
import time
import threading
class Account:
def __init__(self, _id, balance, lock):
self.id = _id
self.balance = balance
self.lock = lock
def withdraw(self, amount):
self.balance -= amount
def deposit(self, amount):
self.balance += amount
def transfer(_from, to, amount):
if _from.lock.acquire():
_from.withdraw(amount)
time.sleep(1)
print('wait for lock...')
if to.lock.acquire():
to.deposit(amount)
to.lock.release()
_from.lock.release()
print('finish...')
if __name__ == "__main__":
a = Account('a', 1000, threading.Lock())
b = Account('b', 1000, threading.Lock())
threading.Thread(target=transfer, args=(a, b, 100)).start()
threading.Thread(target=transfer, args=(b, a, 200)).start()
解決死鎖問題的一種方案是爲程序中的每個鎖分配一個惟一的id，而後只容許按照升序規則來使用多個鎖。
4、線程池
一、線程池簡介
線程池在系統啓動時即建立大量空閒的線程，程序只要將一個任務函數提交給線程池，線程池就會啓動一個空閒的線程來執行它。當任務函數執行結束後，線程並不會死亡，而是再次返回到線程池中變成空閒狀態，等待執行下一個任務函數。
使用線程池能夠有效地控制系統中併發線程的數量。當系統中包含有大量的併發線程時，會致使系統性能急劇降低，甚至致使 Python 解釋器崩潰，而線程池的最大線程數參數能夠控制系統中併發線程的數量不超過此數。
concurrent.futures模塊中的 Executor是線程池的抽象基類，Executor 提供了兩個子類，即ThreadPoolExecutor 和ProcessPoolExecutor，其中 ThreadPoolExecutor 用於建立線程池，ProcessPoolExecutor 用於建立進程池。
Exectuor 提供了以下經常使用接口：submit(fn, *args, **kwargs)：將 fn 函數提交給線程池。args 表明傳給 fn 函數的參數，是元組類型，kwargs 表明以關鍵字參數的形式爲 fn 函數傳入參數，是字典類型。submit 方法會返回一個 Future 對象map(func, *iterables, timeout=None, chunksize=1)：map函數將會啓動多個線程，以異步方式當即對 iterables 執行 map 處理。
shutdown(wait=True)：關閉線程池。
Future 提供了以下方法：
cancel()：取消Future 表明的線程任務。若是任務正在執行，不可取消，則返回 False；不然，程序會取消任務，並返回 True。
cancelled()：返回Future表明的線程任務是否被成功取消。
running()：若是Future 表明的線程任務正在執行、不可被取消，則返回 True。
done()：若是Funture 表明的線程任務被成功取消或執行完成，則返回 True。
result(timeout=None)：獲取Future 表明的線程任務最後的返回結果。若是 Future 表明的線程任務還未完成，result方法將會阻塞當前線程，其中 timeout 參數指定最多阻塞多少秒。
exception(timeout=None)：獲取Future 表明的線程任務所引起的異常。若是任務成功完成，沒有異常，則該方法返回 None。
add_done_callback(fn)：爲Future 表明的線程任務註冊一個「回調函數」，當線程任務成功完成時，程序會自動觸發fn 函數。
使用用完一個線程池後，應該調用線程池的 shutdown() 方法， shutdown方法將啓動線程池的關閉序列。調用 shutdown() 方法後的線程池再也不接收新任務，但會將全部的已提交任務執行完成。當線程池中的全部任務都執行完成後，線程池中的全部線程都會死亡。
二、線程池
ThreadPoolExecutor(max_works)，若是未顯式指定max_works，默認線程池會建立CPU的數目*5數量的線程 。
# -*- coding:utf-8 -*-
from concurrent.futures import ThreadPoolExecutor
import threading
import time
import os
import string
class WorkThread(threading.Thread):
def __init__(self):
threading.Thread.__init__(self)
def run(self):
print('Process[%s]:%s start and run task' % (os.getpid(), threading.currentThread().getName()))
time.sleep(2)
return "Process[{}]:{} end".format(os.getpid(), threading.currentThread().getName())
def work_task(thread_name):
print('Process[%s]:%s start and run task' % (os.getpid(), threading.currentThread().getName()))
time.sleep(5)
return "Process[{}]:{} end".format(os.getpid(), threading.currentThread().getName())
def get_call_back(future):
print(future.result())
if __name__ == '__main__':
print('main thread start')
# create thread pool
thread_pool = ThreadPoolExecutor(5)
futures = []
for i in range(5):
thread = WorkThread()
future = thread_pool.submit(thread.run)
futures.append(future)
for i in range(5):
future = thread_pool.submit(work_task, i)
futures.append(future)
for future in futures:
future.add_done_callback(get_call_back)
# thread_pool.map(work_task, (2, 3, 4))
thread_pool.shutdown()
# output:
# main thread start
# Process[718]:ThreadPoolExecutor-0_0 start and run task
# Process[718]:ThreadPoolExecutor-0_1 start and run task
# Process[718]:ThreadPoolExecutor-0_2 start and run task
# Process[718]:ThreadPoolExecutor-0_3 start and run task
# Process[718]:ThreadPoolExecutor-0_4 start and run task
# Process[718]:ThreadPoolExecutor-0_3 end
# Process[718]:ThreadPoolExecutor-0_3 start and run task
# Process[718]:ThreadPoolExecutor-0_1 end
# Process[718]:ThreadPoolExecutor-0_1 start and run task
# Process[718]:ThreadPoolExecutor-0_2 end
# Process[718]:ThreadPoolExecutor-0_2 start and run task
# Process[718]:ThreadPoolExecutor-0_0 end
# Process[718]:ThreadPoolExecutor-0_0 start and run task
# Process[718]:ThreadPoolExecutor-0_4 end
# Process[718]:ThreadPoolExecutor-0_4 start and run task
# Process[718]:ThreadPoolExecutor-0_2 end
# Process[718]:ThreadPoolExecutor-0_3 end
# Process[718]:ThreadPoolExecutor-0_1 end
# Process[718]:ThreadPoolExecutor-0_4 end
# Process[718]:ThreadPoolExecutor-0_0 end
三、進程池
ProcessPoolExecutor(max_works)，若是未顯式指定max_works，默認進程池會建立CPU的數目*5數量的進程 。
進程池同步方案：
from concurrent.futures import ProcessPoolExecutor
import os
import time
import random
def work_task(n):
print('Process[%s] is running' % os.getpid())
time.sleep(random.randint(1,3))
return n**2
if __name__ == '__main__':
start = time.time()
pool = ProcessPoolExecutor()
for i in range(5):
obj = pool.submit(work_task, i).result()
pool.shutdown()
print('='*30)
print("time: ", time.time() - start)
# output;
# Process[7372] is running
# Process[7373] is running
# Process[7374] is running
# Process[7375] is running
# Process[7372] is running
# ==============================
# time: 10.023026466369629
進程池異步方案：
from concurrent.futures import ProcessPoolExecutor
import os
import time
import random
def work_task(n):
print('Process[%s] is running' % os.getpid())
time.sleep(random.randint(1, 3))
return n**2
if __name__ == '__main__':
start = time.time()
pool = ProcessPoolExecutor()
objs = []
for i in range(5):
obj = pool.submit(work_task, i)
objs.append(obj)
pool.shutdown()
print('='*30)
print([obj.result() for obj in objs])
print("time: ", time.time() - start)
# output;
# Process[8268] is running
# Process[8269] is running
# Process[8270] is running
# Process[8271] is running
# Process[8270] is running
# ==============================
# [0, 1, 4, 9, 16]
# time: 2.0124566555023193
5、生產者消費者模型
import threading
from queue import Queue
from urllib.request import urlopen
ips = ["www.baidu.com",
"www.taobao.com",
"www.huawei.com",
"www.alibaba.com",
"www.meituan.com",
"www.xiaomi.com"]
ports = [80, 443]
class Producer(threading.Thread):
def __init__(self, _queue):
super(Producer, self).__init__()
self.queue = _queue
def run(self):
urls = ["http://%s:%s" % (ip, port) for ip in ips for port in ports]
for url in urls:
self.queue.put(url)
class Consumer(threading.Thread):
def __init__(self, _queue):
super(Consumer, self).__init__()
self.queue = _queue
def run(self):
try:
url = self.queue.get()
urlopen(url)
except Exception as e:
print("%s is unknown url" % url)
else:
print("%s is ok" % url)
if __name__ == "__main__":
# 實例化一個隊列
queue = Queue()
for i in range(2):
producer = Producer(queue)
producer.start()
for i in range(30):
consumer = Consumer(queue)
consumer.start()
# output:
# http://www.taobao.com:443 is unknown url
# http://www.huawei.com:443 is unknown url
# http://www.huawei.com:443 is unknown urlhttp://www.taobao.com:443 is unknown url
#
# http://www.baidu.com:80 is ok
# http://www.baidu.com:443 is unknown url
# http://www.xiaomi.com:443 is unknown url
# http://www.baidu.com:80 is ok
# http://www.baidu.com:443 is unknown url
# http://www.xiaomi.com:443 is unknown url
# http://www.alibaba.com:443 is unknown url
# http://www.alibaba.com:443 is unknown url
# http://www.meituan.com:443 is unknown urlhttp://www.meituan.com:443 is unknown url
#
# http://www.huawei.com:80 is ok
# http://www.huawei.com:80 is ok
# http://www.xiaomi.com:80 is ok
# http://www.xiaomi.com:80 is ok
# http://www.taobao.com:80 is ok
# http://www.meituan.com:80 is ok
# http://www.meituan.com:80 is ok
# http://www.taobao.com:80 is ok
# http://www.alibaba.com:80 is ok
# http://www.alibaba.com:80 is ok
相關文章
- 1. Python3快速入門（九）Python3併發編程
- 2. Python3快速入門
- 3. Python3快速入門（十）Python3網絡編程
- 4. Python3快速入門（十）——Python3網絡編程
- 5. Python3快速入門（二）Python3基礎
- 6. Python3快速入門（五）——Python3函數
- 7. Python3快速入門（五）Python3函數
- 8. Python3快速入門（二）——Python3基礎
- 9. Python3快速入門（八）——Python3 JSON
- 10. Python3+RobotFramewok 快速入門（二）
- 更多相關文章...
- • Rust 併發編程 - RUST 教程
- • Memcached入門教程 - NoSQL教程
- • YAML 入門教程
- • Java Agent入門實戰（一）-Instrumentation介紹與使用
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
歡迎關注本站公眾號,獲取更多信息
