Python3快速入門（九）——Python3并發編程

發布時間：2020-08-06 09:48:34 來源：網絡閱讀：3916 作者：天山老妖S 欄目：編程語言

Python3快速入門（九）——Python3并發編程

一、Python線程模塊

1、線程簡介

一個標準的線程由線程ID，當前指令指針(PC），寄存器集合和堆棧組成。線程是進程中的一個實體，是被系統獨立調度和分派的基本單位，線程本身不擁有系統資源，與進程內的其它線程共享進程的所有資源。一個進程中至少有一個線程，并作為程序的入口，即主線程，其它線程稱為工作線程。
???? 多線程，是指從軟件或者硬件上實現多個線程并發執行的技術。支持多線程能力的計算機因有硬件支持而能夠在同一時間執行多個線程，進而提升整體處理性能。

2、線程狀態

線程有就緒、阻塞、運行三種基本狀態。就緒狀態是指線程具備運行的所有條件，在等待CPU執行；?運行狀態是指線程占有CPU正在運行；?阻塞狀態是指線程在等待一個事件，邏輯上不可執行。
三種狀態的相互轉化如下圖所示：
Python3快速入門（九）——Python3并發編程

2、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():?設置線程名。

3、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是可選的超時時間，超過一個時間主進程就不等待。

4、全局解釋鎖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

計算密集型任務-多線程方案：

# -*- 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密集型任務-多進程方案：

# -*- 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

二、創建線程

1、threading.Thread實例化

threading.Thread構造函數如下：

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

2、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

3、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()方法僅是普通函數調用。

4、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

三、線程同步

如果多個線程共同對某個數據修改，則可能出現不可預料的結果，為了保證數據的正確性，需要對多個線程進行同步。

1、互斥鎖

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

2、信號量

互斥鎖同時只允許一個線程訪問共享數據，而信號量同時允許一定數量的線程訪問共享數據，如銀行柜臺有 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

3、條件變量

條件變量能讓一個線程 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.

4、事件

事件用于線程間通信。一個線程發出一個信號，其它一個或多個線程等待，調用 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!

5、線程優先級隊列

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

6、線程死鎖

死鎖是指兩個或兩個以上的進程或線程在執行過程中，因爭奪資源而造成的一種互相等待的現象。在線程間共享多個資源的時候，如果分別占有一部分資源并且同時在等待對方的資源，就會造成死鎖。例如數據庫操作時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，然后只允許按照升序規則來使用多個鎖。

四、線程池

1、線程池簡介

線程池在系統啟動時即創建大量空閑的線程，程序只要將一個任務函數提交給線程池，線程池就會啟動一個空閑的線程來執行它。當任務函數執行結束后，線程并不會死亡，而是再次返回到線程池中變成空閑狀態，等待執行下一個任務函數。
使用線程池可以有效地控制系統中并發線程的數量。當系統中包含有大量的并發線程時，會導致系統性能急劇下降，甚至導致?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() 方法后的線程池不再接收新任務，但會將所有的已提交任務執行完成。當線程池中的所有任務都執行完成后，線程池中的所有線程都會死亡。

2、線程池

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

3、進程池

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

五、生產者消費者模型

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

向AI問一下細節

91超碰碰碰碰久久久久久综合_超碰av人澡人澡人澡人澡人掠_国产黄大片在线观看画质优化_txt小说免费全本

Python3快速入門（九）——Python3并發編程