為什么要使用多線程？

使用多線程，可以同時(shí)進(jìn)行多項(xiàng)任務(wù)，可以使用戶界面更友好，還可以后臺(tái)執(zhí)行某些用時(shí)長的任務(wù)，同時(shí)具有易于通信的優(yōu)點(diǎn)。

python3中多線程的實(shí)現(xiàn)使用了threading模塊，它允許同一進(jìn)程中運(yùn)行多個(gè)線程。

如何創(chuàng)建和執(zhí)行一個(gè)線程

一般我們有兩種方法來創(chuàng)建線程，一種是以某個(gè)函數(shù)來作為起點(diǎn)，另一種是繼承Thread類。

方法一

獲取一個(gè)Thread對(duì)象，構(gòu)造參數(shù)中target是起點(diǎn)函數(shù)，注意不要加括號(hào)。假如起點(diǎn)函數(shù)有參數(shù)，則可以通過args輸入元組參數(shù)或者kwargs輸入字典參數(shù)。

#! -*-conding=: UTF-8 -*-
# 2023/5/6 15:53
import time
from threading import Thread


def task():
    print("另外開始一個(gè)子線程做任務(wù)啦")
    time.sleep(1)  # 用time.sleep模擬任務(wù)耗時(shí)
    print("子線程任務(wù)結(jié)束啦")


if __name__ == '__main__':
    print("這里是主線程")
    # 創(chuàng)建線程對(duì)象
    t1 = Thread(target=task)
    # 啟動(dòng)
    t1.start()
    time.sleep(0.3)
    print("主線程依然可以干別的事")

輸出結(jié)果為：

這里是主線程
另外開始一個(gè)子線程做任務(wù)啦
主線程依然可以干別的事
子線程任務(wù)結(jié)束啦

方法二

#! -*-conding=: UTF-8 -*-
# 2023/5/6 15:53
import time
from threading import Thread


class NewThread(Thread):
    def __init__(self):
        Thread.__init__(self)  # 必須步驟

    def run(self):  # 入口是名字為run的方法
        print("開始新的線程做一個(gè)任務(wù)啦")
        time.sleep(1)  # 用time.sleep模擬任務(wù)耗時(shí)
        print("這個(gè)新線程中的任務(wù)結(jié)束啦")


if __name__ == '__main__':
    print("這里是主線程")
    # 創(chuàng)建線程對(duì)象
    t1 = NewThread()
    # 啟動(dòng)
    t1.start()
    time.sleep(0.3)  # 這里如果主線程結(jié)束，子線程會(huì)立刻退出，暫時(shí)先用sleep規(guī)避
    print("主線程依然可以干別的事")

正式介紹threading模塊

docs.python.org/3/library/t…

關(guān)于線程信息的函數(shù)：

• threading.active_count()：返回當(dāng)前存活的Thread對(duì)象數(shù)量。
• threading.current_thread()：返回當(dāng)前線程的Thread對(duì)象。
• threading.enumerate()：列表形式返回所有存活的Thread對(duì)象。
• threading.main_thread()：返回主Thread對(duì)象。

Thread對(duì)象的方法及屬性：

• Thread.name：線程的名字，沒有語義，可以相同名稱。
• Thread.ident：線程標(biāo)識(shí)符，非零整數(shù)。
• Thread.Daemon：是否為守護(hù)線程。
• Thread.is_alive()：是否存活。
• Thread.start()：開始線程活動(dòng)。若多次調(diào)用拋出RuntimeError。
• Thread.run()：用來重載的，
• Thread.join(timeout=None)：等待直到線程正?；虍惓＝Y(jié)束。尚未開始拋出RuntimeError
• Thread(group=None, target=None, name=None, args=(), kwargs={}, *, deamon=None)：構(gòu)造函數(shù)。

守護(hù)線程 Daemon

在Python 3中，守護(hù)線程（daemon thread）是一種特殊的線程，它在程序運(yùn)行時(shí)在后臺(tái)運(yùn)行，不會(huì)阻止程序的退出。當(dāng)主線程退出時(shí)，守護(hù)線程也會(huì)自動(dòng)退出，而不需要等待它執(zhí)行完畢。

方法一

在創(chuàng)建線程對(duì)象時(shí)，可以通過設(shè)置daemon屬性為True來創(chuàng)建守護(hù)線程，例如：

import threading
import time

def worker():
    while True:
        print('Worker thread running')
        time.sleep(1)

# 創(chuàng)建守護(hù)線程
t = threading.Thread(target=worker, daemon=True)
# 啟動(dòng)線程
t.start()

# 主線程執(zhí)行一些操作
print('Main thread running')
time.sleep(5)
print('Main thread finished')

在這個(gè)示例中，我們創(chuàng)建了一個(gè)守護(hù)線程worker()，并將daemon屬性設(shè)置為True。在主線程中，我們執(zhí)行了一些操作，并休眠5秒鐘。由于守護(hù)線程的存在，即使主線程已經(jīng)結(jié)束，守護(hù)線程仍會(huì)在后臺(tái)運(yùn)行。

方法二

設(shè)置守護(hù)線程用Thread.setDaemon(bool)

#! -*-conding=: UTF-8 -*-
# 2023/5/6 16:06


import time
from threading import Thread


def task1():
    print("開始子線程1做任務(wù)1啦")
    time.sleep(1)  # 用time.sleep模擬任務(wù)耗時(shí)
    print("子線程1中的任務(wù)1結(jié)束啦")


def task2():
    print("開始子線程2做任務(wù)2啦")
    for i in range(5):
        print("任務(wù)2-{}".format(i))
        time.sleep(1)
    print("子線程2中的任務(wù)2結(jié)束啦")


if __name__ == '__main__':
    print("這里是主線程")
    # 創(chuàng)建線程對(duì)象
    t1 = Thread(target=task1)
    t2 = Thread(target=task2)
    t2.setDaemon(True)  # 設(shè)置為守護(hù)進(jìn)程，必須在start之前
    # 啟動(dòng)
    t1.start()
    t2.start()
    time.sleep(1)
    print("主線程結(jié)束了")

輸出結(jié)果為：

這里是主線程
開始子線程1做任務(wù)1啦
開始子線程2做任務(wù)2啦
任務(wù)2-0
主線程結(jié)束了
子線程1中的任務(wù)1結(jié)束啦任務(wù)2-1

守護(hù)線程的作用在于，當(dāng)我們需要在程序運(yùn)行時(shí)執(zhí)行一些后臺(tái)任務(wù)，但是不想讓這些任務(wù)阻止程序的正常退出時(shí)，可以使用守護(hù)線程。
例如，在一個(gè)Web應(yīng)用程序中，我們可能需要啟動(dòng)一個(gè)守護(hù)線程來定期清理緩存或者執(zhí)行一些后臺(tái)任務(wù)。

需要注意的是，守護(hù)線程無法完全控制其執(zhí)行過程，因此不能用于一些必須在程序退出之前完成的任務(wù)。同時(shí)，守護(hù)線程不能訪問一些主線程資源，例如共享內(nèi)存或者打開的文件，因?yàn)檫@些資源可能會(huì)在主線程結(jié)束時(shí)被釋放。

讓主線程等待子線程結(jié)束 join

假如要讓主線程等子線程結(jié)束，那么可以使用Thread.join()方法。

當(dāng)調(diào)用線程對(duì)象的join()方法時(shí)，主線程將被阻塞，直到該線程執(zhí)行完成或者超時(shí)。

以下是一個(gè)簡單的示例：

import threading
import time

def worker():
    print('Worker thread started')
    time.sleep(2)
    print('Worker thread finished')

# 創(chuàng)建線程對(duì)象
t = threading.Thread(target=worker)
# 啟動(dòng)線程
t.start()

# 等待線程結(jié)束
t.join()

# 主線程繼續(xù)執(zhí)行
print('Main thread finished')

在這個(gè)示例中，我們創(chuàng)建了一個(gè)子線程worker()，并使用start()方法啟動(dòng)線程。在主線程中，我們調(diào)用了線程對(duì)象的join()方法，讓主線程等待子線程執(zhí)行完畢。在子線程執(zhí)行完畢后，主線程繼續(xù)執(zhí)行。

需要注意的是，join()方法還可以設(shè)置超時(shí)時(shí)間，以避免無限期等待線程的執(zhí)行。例如：

import threading
import time

def worker():
    print('Worker thread started')
    time.sleep(2)
    print('Worker thread finished')

# 創(chuàng)建線程對(duì)象
t = threading.Thread(target=worker)
# 啟動(dòng)線程
t.start()

# 等待線程結(jié)束，最多等待3秒鐘
t.join(3)

# 主線程繼續(xù)執(zhí)行
print('Main thread finished')

在這個(gè)示例中，我們?cè)O(shè)置了join()方法的超時(shí)時(shí)間為3秒鐘，即使子線程沒有執(zhí)行完成，主線程也會(huì)在3秒鐘后繼續(xù)執(zhí)行。

線程共享資源可能引起什么問題？

在線程編程中，多個(gè)線程可能同時(shí)訪問和修改同一個(gè)共享資源，例如全局變量、共享內(nèi)存、文件等。如果沒有進(jìn)行適當(dāng)?shù)耐讲僮?，就可能?huì)引發(fā)以下問題：

競(jìng)態(tài)條件（Race Condition）：當(dāng)多個(gè)線程同時(shí)訪問和修改同一個(gè)共享資源時(shí)，就可能會(huì)發(fā)生競(jìng)態(tài)條件。這種情況下，由于線程執(zhí)行順序的不確定性，可能會(huì)導(dǎo)致資源被錯(cuò)誤地讀取或?qū)懭耄瑥亩l(fā)程序的錯(cuò)誤或崩潰。

死鎖（Deadlock）：當(dāng)多個(gè)線程都在等待另一個(gè)線程釋放某個(gè)資源時(shí)，就可能會(huì)發(fā)生死鎖。這種情況下，程序會(huì)永久地阻塞在這個(gè)狀態(tài)下，無法繼續(xù)執(zhí)行。

活鎖（Livelock）：多個(gè)線程相互協(xié)作，但是由于某些原因無法前進(jìn)，導(dǎo)致它們不斷重試，最終導(dǎo)致系統(tǒng)陷入死循環(huán)?；铈i是一種比死鎖更難以診斷和解決的問題。

為了避免以上問題，我們可以使用線程同步機(jī)制來保護(hù)共享資源的訪問。

例如，可以使用鎖（Lock）、信號(hào)量（Semaphore）、條件變量（Condition）等機(jī)制來限制同時(shí)訪問共享資源的線程數(shù)量，從而避免競(jìng)態(tài)條件。同時(shí)，也可以使用一些算法和策略來避免死鎖和活鎖等問題的發(fā)生。

下面是一些具體的例子，說明在多線程程序中共享資源可能引發(fā)的問題：

競(jìng)態(tài)條件

import threading

x = 0

def increment():
    global x
    x += 1

threads = []
for i in range(1000):
    t = threading.Thread(target=increment)
    threads.append(t)
    t.start()

for t in threads:
    t.join()

print('x =', x)

在這個(gè)例子中，我們創(chuàng)建了1000個(gè)線程來執(zhí)行increment()函數(shù)，這個(gè)函數(shù)會(huì)對(duì)全局變量x進(jìn)行加1操作。由于多個(gè)線程同時(shí)訪問和修改x變量，就會(huì)產(chǎn)生競(jìng)態(tài)條件，導(dǎo)致x的最終值可能小于1000。

死鎖

import threading

lock1 = threading.Lock()
lock2 = threading.Lock()

def worker1():
    print('Worker 1 acquiring lock 1')
    lock1.acquire()
    print('Worker 1 acquired lock 1')
    print('Worker 1 acquiring lock 2')
    lock2.acquire()
    print('Worker 1 acquired lock 2')
    lock2.release()
    lock1.release()

def worker2():
    print('Worker 2 acquiring lock 2')
    lock2.acquire()
    print('Worker 2 acquired lock 2')
    print('Worker 2 acquiring lock 1')
    lock1.acquire()
    print('Worker 2 acquired lock 1')
    lock1.release()
    lock2.release()

t1 = threading.Thread(target=worker1)
t2 = threading.Thread(target=worker2)
t1.start()
t2.start()
t1.join()
t2.join()

print('Main thread finished')

在這個(gè)例子中，我們創(chuàng)建了兩個(gè)線程worker1()和worker2()，它們都需要同時(shí)獲取lock1和lock2兩個(gè)鎖來執(zhí)行操作。由于worker1()先獲取lock1，然后嘗試獲取lock2，而worker2()先獲取lock2，然后嘗試獲取lock1，就可能會(huì)產(chǎn)生死鎖的情況。

活鎖

import threading

class Account:
    def __init__(self, balance):
        self.balance = balance
        self.lock = threading.Lock()

    def withdraw(self, amount):
        with self.lock:
            if self.balance < amount:
                print('Withdraw failed: not enough balance')
                return False
            print(f'Withdraw {amount} from account')
            self.balance -= amount
            return True

    def transfer(self, target, amount):
        while True:
            if self.withdraw(amount):
                if target.deposit(amount):
                    return True
                else:
                    self.deposit(amount)
            else:
                return False

    def deposit(self, amount):
        with self.lock:
            print(f'Deposit {amount} to account')
            self.balance += amount
            return True

def worker1(acc1, acc2):
    while True:
        acc1.transfer(acc2, 100)
        print('Worker 1: transfer complete')

def worker2(acc1, acc2):
    while True:
        acc2.transfer(acc1, 100)
        print('Worker 2: transfer complete')

acc1 = Account(1000)
acc2 = Account(1000)
t1 = threading.Thread(target=worker1, args=(acc1, acc2))
t2 = threading.Thread(target=worker2, args=(acc1, acc2))
t1.start()
t2.start()
t1.join()
t2.join()

在這個(gè)例子中，我們創(chuàng)建了兩個(gè)賬戶acc1和acc2，并創(chuàng)建了兩個(gè)線程worker1()和worker2()，它們不斷地在這兩個(gè)賬戶之間轉(zhuǎn)賬。

由于transfer()方法中需要獲取鎖來修改賬戶余額，但是兩個(gè)線程的執(zhí)行順序可能會(huì)導(dǎo)致它們同時(shí)等待對(duì)方釋放鎖，從而無法前進(jìn)，最終導(dǎo)致系統(tǒng)陷入活鎖的狀態(tài)。

具體來說，假設(shè)worker1()執(zhí)行了acc1.transfer(acc2, 100)，然后進(jìn)入了transfer()方法中的if self.withdraw(amount)分支，在等待acc1的鎖。

此時(shí)，worker2()執(zhí)行了acc2.transfer(acc1, 100)，然后也進(jìn)入了transfer()方法中的if self.withdraw(amount)分支，在等待acc2的鎖。由于acc1和acc2之間的轉(zhuǎn)賬是相互依賴的，因此這兩個(gè)線程無法前進(jìn)，會(huì)一直重試，最終導(dǎo)致系統(tǒng)陷入活鎖的狀態(tài)。

多線程的鎖機(jī)制

在Python3中，鎖機(jī)制是一種線程同步機(jī)制，它用于協(xié)調(diào)多個(gè)線程的并發(fā)訪問共享資源，以避免競(jìng)態(tài)條件的發(fā)生。

Python 3中的多線程鎖機(jī)制主要是通過threading模塊中的Lock、RLock和Semaphore等類來實(shí)現(xiàn)的。

Lock類是最基本的鎖，它提供了兩個(gè)基本方法acquire()和release()，用于獲取鎖和釋放鎖。當(dāng)一個(gè)線程調(diào)用acquire()方法時(shí)，如果該鎖沒有被其他線程獲取，則該線程獲取到該鎖并進(jìn)入臨界區(qū)，否則該線程就會(huì)被阻塞，直到該鎖被其他線程釋放為止。

RLock類是可重入鎖，它允許同一個(gè)線程多次獲取該鎖，每次獲取都必須有對(duì)應(yīng)的釋放操作。如果一個(gè)線程已經(jīng)獲取到該鎖，它可以再次獲取該鎖而不被阻塞，這就是可重入的特性。RLock類提供了acquire()和release()方法，與Lock類相同。

Semaphore類是信號(hào)量，它與鎖類似，但可以允許多個(gè)線程同時(shí)訪問某個(gè)資源，而不是像鎖一樣只允許一個(gè)線程訪問。它提供了acquire()和release()方法，用于獲取和釋放資源。

下面是一個(gè)使用Lock類的示例代碼：

import threading

counter = 0
lock = threading.Lock()

def increment():
    global counter
    for i in range(100000):
        lock.acquire()
        counter += 1
        lock.release()

threads = []
for i in range(10):
    t = threading.Thread(target=increment)
    threads.append(t)
    t.start()

for t in threads:
    t.join()

print(counter)

上面的代碼中，我們定義了一個(gè)全局變量counter和一個(gè)Lock對(duì)象lock。increment()函數(shù)用于在循環(huán)中對(duì)counter進(jìn)行100000次加1操作，而在每次加1之前，我們首先獲取lock，加1操作完成之后再釋放lock。這樣保證了多個(gè)線程同時(shí)對(duì)counter進(jìn)行操作時(shí)，不會(huì)產(chǎn)生競(jìng)爭條件。

另外，還需要注意到，對(duì)于每個(gè)獲取鎖的線程，一定要記得在合適的地方釋放鎖，否則就會(huì)出現(xiàn)死鎖的情況。

在多線程環(huán)境中，多個(gè)線程可能同時(shí)訪問某個(gè)共享資源，這可能導(dǎo)致競(jìng)態(tài)條件的發(fā)生，從而導(dǎo)致程序出現(xiàn)不可預(yù)測(cè)的結(jié)果。為了避免這種情況的發(fā)生，我們可以使用鎖機(jī)制來控制對(duì)共享資源的訪問。在使用鎖機(jī)制時(shí)，需要注意以下幾點(diǎn)：

• 鎖是一種互斥機(jī)制，即同一時(shí)刻只能有一個(gè)線程持有鎖，其他線程必須等待該線程釋放鎖后才能繼續(xù)執(zhí)行。

• 在訪問共享資源前，線程需要先獲取鎖。如果鎖已經(jīng)被其他線程持有，則線程會(huì)被阻塞，直到其他線程釋放鎖。

• 在訪問共享資源后，線程需要釋放鎖，以便其他線程可以獲取鎖并訪問共享資源。

• 在使用鎖時(shí)，需要保證所有線程都使用同一個(gè)鎖對(duì)象。

鎖機(jī)制可以用于解決多線程程序中的競(jìng)態(tài)條件、死鎖和活鎖等問題。

下面我們分別通過例子來說明鎖是如何解決這些問題的。

競(jìng)態(tài)條件

競(jìng)態(tài)條件指的是多個(gè)線程對(duì)共享資源的競(jìng)爭，導(dǎo)致結(jié)果的正確性取決于線程的執(zhí)行順序。

比如，在一個(gè)多線程程序中，多個(gè)線程同時(shí)對(duì)同一個(gè)變量進(jìn)行加減操作，結(jié)果可能取決于每個(gè)線程的執(zhí)行順序，這就是一個(gè)典型的競(jìng)態(tài)條件。

通過使用鎖，可以保證在任何時(shí)刻只有一個(gè)線程能夠訪問共享資源，從而避免競(jìng)態(tài)條件的出現(xiàn)。下面的例子演示了如何使用鎖來解決競(jìng)態(tài)條件：

import threading

class Counter:
    def __init__(self):
        self.count = 0
        self.lock = threading.Lock()

    def increment(self):
        with self.lock:
            self.count += 1

def worker(counter, num_iters):
    for i in range(num_iters):
        counter.increment()

counter = Counter()
num_threads = 10
num_iters = 10000
threads = [threading.Thread(target=worker, args=(counter, num_iters)) for _ in range(num_threads)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print(counter.count)

在這個(gè)例子中，多個(gè)線程對(duì) Counter 對(duì)象的 count 屬性進(jìn)行加 1 操作，這可能會(huì)導(dǎo)致競(jìng)態(tài)條件。

為了避免這種情況，我們使用了一個(gè)鎖，通過 with self.lock 來獲取鎖，這樣在任何時(shí)刻只有一個(gè)線程能夠修改 count 屬性。

這樣，我們就避免了競(jìng)態(tài)條件的出現(xiàn)。

死鎖

死鎖是指兩個(gè)或多個(gè)線程在等待彼此釋放資源，從而形成僵局的情況。為了解決死鎖問題，可以使用鎖機(jī)制來協(xié)調(diào)線程對(duì)共享資源的訪問。具體來說，當(dāng)一個(gè)線程獲得鎖時(shí)，其他線程必須等待該線程釋放鎖之后才能訪問共享資源，從而避免多個(gè)線程同時(shí)訪問同一個(gè)共享資源而產(chǎn)生死鎖。

例如，考慮一個(gè)簡單的場(chǎng)景，其中有兩個(gè)線程，分別需要獲取兩個(gè)共享資源才能繼續(xù)執(zhí)行。假設(shè)這兩個(gè)線程在獲取資源時(shí)的順序不同，可能會(huì)出現(xiàn)死鎖的情況。

import threading

resource_a = threading.Lock()
resource_b = threading.Lock()

def thread_a():
    resource_a.acquire()
    resource_b.acquire()
    print("Thread A acquired resource A and resource B")
    resource_b.release()
    resource_a.release()

def thread_b():
    resource_b.acquire()
    resource_a.acquire()
    print("Thread B acquired resource A and resource B")
    resource_a.release()
    resource_b.release()

thread1 = threading.Thread(target=thread_a)
thread2 = threading.Thread(target=thread_b)

thread1.start()
thread2.start()

thread1.join()
thread2.join()

上述代碼中，thread_a和thread_b分別獲取resource_a和resource_b，但是它們的獲取順序不同。因此，如果這兩個(gè)線程同時(shí)運(yùn)行，就有可能發(fā)生死鎖的情況，導(dǎo)致程序卡住。

為了避免死鎖，可以使用鎖機(jī)制。修改上述代碼，如下所示：

import threading

resource_a = threading.Lock()
resource_b = threading.Lock()

def thread_a():
    resource_a.acquire()
    resource_b.acquire()
    print("Thread A acquired resource A and resource B")
    resource_b.release()
    resource_a.release()

def thread_b():
    resource_a.acquire()
    resource_b.acquire()
    print("Thread B acquired resource A and resource B")
    resource_b.release()
    resource_a.release()

thread1 = threading.Thread(target=thread_a)
thread2 = threading.Thread(target=thread_b)

thread1.start()
thread2.start()

thread1.join()
thread2.join()

在這個(gè)示例中，每個(gè)線程都按照相同的順序獲取鎖，這樣就避免了死鎖的情況。

活鎖

活鎖是多線程程序中的一種常見問題，它是指線程在嘗試協(xié)調(diào)其操作時(shí)一直重試，但最終沒有達(dá)到進(jìn)展的狀態(tài)。一個(gè)常見的例子是兩個(gè)線程互相等待對(duì)方釋放其持有的資源。

使用鎖是解決活鎖問題的一種常見方式。當(dāng)線程需要訪問共享資源時(shí)，必須獲得相應(yīng)的鎖。如果鎖已經(jīng)被其他線程持有，線程將阻塞直到獲得鎖為止。這樣，當(dāng)多個(gè)線程嘗試同時(shí)訪問同一共享資源時(shí)，只有一個(gè)線程能夠獲取鎖，其他線程將被阻塞。

下面是一個(gè)使用鎖解決活鎖問題的例子。假設(shè)有兩個(gè)線程A和B，它們需要同時(shí)訪問兩個(gè)共享資源x和y，但由于資源x和y的訪問順序不同，線程A需要先獲得x再獲得y，而線程B需要先獲得y再獲得x。如果兩個(gè)線程嘗試同時(shí)獲取它們需要的資源，就會(huì)出現(xiàn)活鎖問題。

使用鎖可以解決這個(gè)問題。假設(shè)每個(gè)線程都先獲取x的鎖，然后再獲取y的鎖，這樣就可以保證每個(gè)線程都按照相同的順序獲取資源，避免了死鎖和活鎖的問題。

下面是使用鎖解決活鎖問題的代碼示例：

import threading

class Resource:
    def __init__(self):
        self.lock1 = threading.Lock()
        self.lock2 = threading.Lock()

    def get_x(self):
        self.lock1.acquire()
        return "x"

    def get_y(self):
        self.lock2.acquire()
        return "y"

    def release_x(self):
        self.lock1.release()

    def release_y(self):
        self.lock2.release()

def thread_a(resource):
    while True:
        x = resource.get_x()
        y = resource.get_y()
        print("Thread A got resources x and y")
        resource.release_x()
        resource.release_y()

def thread_b(resource):
    while True:
        y = resource.get_y()
        x = resource.get_x()
        print("Thread B got resources y and x")
        resource.release_y()
        resource.release_x()

if __name__ == "__main__":
    resource = Resource()
    a = threading.Thread(target=thread_a, args=(resource,))
    b = threading.Thread(target=thread_b, args=(resource,))
    a.start()
    b.start()

在這個(gè)例子中，每個(gè)線程都使用相同的鎖順序來獲取資源x和y，這樣就避免了活鎖的問題。

使用鎖可能導(dǎo)致執(zhí)行速度慢，但是保證了線程安全
無論是Lock還是RLock，acquire和release都要成對(duì)出現(xiàn)

多線程的通信

Python3 中多線程之間的通信方式有以下幾種：

隊(duì)列

在 Python 3 中，可以使用隊(duì)列（Queue）實(shí)現(xiàn)多線程之間的通信。隊(duì)列是線程安全的數(shù)據(jù)結(jié)構(gòu)，可以實(shí)現(xiàn)線程之間的同步和協(xié)作，避免競(jìng)爭條件和死鎖問題。

Python 內(nèi)置了 Queue 模塊，提供了隊(duì)列數(shù)據(jù)結(jié)構(gòu)，它可以用于實(shí)現(xiàn)多線程之間的安全通信?？梢允褂藐?duì)列的 put() 方法往隊(duì)列中添加元素，使用 get() 方法從隊(duì)列中取出元素。

Queue模塊提供了以下幾個(gè)類：

• Queue：基本隊(duì)列，實(shí)現(xiàn)FIFO（先進(jìn)先出）的算法。
• LifoQueue：與Queue類似，但是實(shí)現(xiàn)了LIFO（后進(jìn)先出）的算法。
• PriorityQueue：隊(duì)列中的每個(gè)元素都有一個(gè)優(yōu)先級(jí)，每次彈出優(yōu)先級(jí)最高的元素。
• SimpleQueue：類似于Queue，但是沒有任務(wù)協(xié)作的功能，也就是說，不能在進(jìn)程之間使用。

Queue類中最常用的方法包括：

• put(item[, block[, timeout]])：將一個(gè)item放入隊(duì)列，如果隊(duì)列已滿，block為True則阻塞，直到隊(duì)列未滿或超時(shí)；block為False時(shí)，則拋出queue.Full異常。
• get([block[, timeout]])：從隊(duì)列中取出并返回一個(gè)item，如果隊(duì)列為空，block為True則阻塞，直到隊(duì)列不為空或超時(shí)；block為False時(shí)，則拋出queue.Empty異常。
• task_done()：通知隊(duì)列，一個(gè)先前放入隊(duì)列的任務(wù)已經(jīng)完成。
• join()：阻塞主線程，直到隊(duì)列中所有的任務(wù)都被處理完。

下面舉一個(gè)簡單的例子：

import threading
import queue
import time

# 生產(chǎn)者線程，負(fù)責(zé)向隊(duì)列中添加數(shù)據(jù)
class ProducerThread(threading.Thread):
    def __init__(self, queue, name):
        threading.Thread.__init__(self)
        self.queue = queue
        self.name = name

    def run(self):
        for i in range(5):
            item = "item-" + str(i)
            self.queue.put(item)
            print(self.name, "produced", item)
            time.sleep(1)

# 消費(fèi)者線程，負(fù)責(zé)從隊(duì)列中取出數(shù)據(jù)
class ConsumerThread(threading.Thread):
    def __init__(self, queue, name):
        threading.Thread.__init__(self)
        self.queue = queue
        self.name = name

    def run(self):
        while True:
            item = self.queue.get()
            if item is None:
                break
            print(self.name, "consumed", item)
            time.sleep(0.5)

# 創(chuàng)建一個(gè)隊(duì)列對(duì)象
q = queue.Queue()

# 創(chuàng)建兩個(gè)線程對(duì)象，分別作為生產(chǎn)者和消費(fèi)者
producer = ProducerThread(q, "Producer")
consumer = ConsumerThread(q, "Consumer")

# 啟動(dòng)線程
producer.start()
consumer.start()

# 等待生產(chǎn)者線程完成生產(chǎn)
producer.join()

# 停止消費(fèi)者線程
q.put(None)
consumer.join()

在上面的例子中，我們創(chuàng)建了一個(gè)生產(chǎn)者線程和一個(gè)消費(fèi)者線程。生產(chǎn)者線程負(fù)責(zé)向隊(duì)列中添加數(shù)據(jù)，消費(fèi)者線程負(fù)責(zé)從隊(duì)列中取出數(shù)據(jù)。生產(chǎn)者線程每隔一秒鐘向隊(duì)列中添加一個(gè)字符串，消費(fèi)者線程每隔半秒鐘從隊(duì)列中取出一個(gè)字符串。為了避免消費(fèi)者線程在隊(duì)列為空時(shí)陷入死循環(huán)，我們?cè)陉?duì)列的末尾放置了一個(gè) None 值，當(dāng)消費(fèi)者線程取出該值時(shí)，就會(huì)退出循環(huán)。

事件（Event）

事件是一種同步對(duì)象，可以用于多線程之間的通信，常用于控制線程的執(zhí)行順序?？梢允褂檬录?set() 方法設(shè)置事件，使用 wait() 方法等待事件被設(shè)置，使用 clear() 方法清除事件。

以下是一個(gè)使用事件實(shí)現(xiàn)多線程間通信的示例代碼：

import threading

def worker1(event):
    print('Worker 1 is waiting...')
    event.wait()
    print('Worker 1 is running...')

def worker2(event):
    print('Worker 2 is waiting...')
    event.wait()
    print('Worker 2 is running...')

event = threading.Event()

t1 = threading.Thread(target=worker1, args=(event,))
t2 = threading.Thread(target=worker2, args=(event,))

t1.start()
t2.start()

print('Main thread is sleeping...')
time.sleep(3)
event.set()

t1.join()
t2.join()

該代碼創(chuàng)建了兩個(gè)線程，它們都等待事件被設(shè)置，當(dāng)事件被設(shè)置后，它們才開始執(zhí)行。在主線程中，先休眠了 3 秒鐘，然后設(shè)置了事件，從而喚醒了兩個(gè)線程。在實(shí)際應(yīng)用中，事件可以用于控制線程的執(zhí)行順序，或者實(shí)現(xiàn)線程之間的協(xié)作。

鎖（Lock）

使用鎖可以實(shí)現(xiàn)多線程間的通信，可以通過共享變量和鎖的機(jī)制來實(shí)現(xiàn)線程間的同步和互斥。具體來說，一個(gè)線程需要訪問共享變量時(shí)，首先需要獲得鎖，然后讀取或修改共享變量的值，完成操作后再釋放鎖，以便其他線程訪問共享變量。

• 下面是一個(gè)簡單的示例代碼，其中兩個(gè)線程共享一個(gè)變量 counter，通過鎖的機(jī)制來實(shí)現(xiàn)對(duì)該變量的互斥訪問。

import threading

class CounterThread(threading.Thread):
    def __init__(self, lock):
        super().__init__()
        self.lock = lock

    def run(self):
        global counter
        for i in range(100000):
            with self.lock:
                counter += 1

if __name__ == '__main__':
    lock = threading.Lock()
    counter = 0
    threads = [CounterThread(lock) for _ in range(10)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    print(f'counter = {counter}')

在這個(gè)例子中，CounterThread 是一個(gè)繼承自 threading.Thread 的線程類，它有一個(gè)成員變量 lock，用于控制對(duì)共享變量 counter 的訪問。在 run 方法中，線程會(huì)循環(huán)執(zhí)行一定次數(shù)的加操作，每次操作前會(huì)獲取鎖并對(duì) counter 做加一操作，完成后再釋放鎖。在主線程中創(chuàng)建了 10 個(gè) CounterThread 線程，并啟動(dòng)它們進(jìn)行計(jì)數(shù)操作。在所有線程都執(zhí)行完畢后，打印出 counter 的最終值。

使用鎖可以確保多個(gè)線程對(duì)共享變量的訪問是互斥的，從而避免競(jìng)態(tài)條件和數(shù)據(jù)損壞等問題。但是，使用鎖也可能會(huì)導(dǎo)致性能問題和死鎖等問題，因此需要謹(jǐn)慎使用，并根據(jù)實(shí)際情況選擇合適的同步機(jī)制。

• 或者

import threading

class Counter:
    def __init__(self):
        self.value = 0
        self.lock = threading.Lock()

    def increment(self):
        with self.lock:
            self.value += 1

def worker(counter, n):
    for i in range(n):
        counter.increment()

counter = Counter()
threads = []
for i in range(10):
    t = threading.Thread(target=worker, args=(counter, 10000))
    threads.append(t)

for t in threads:
    t.start()

for t in threads:
    t.join()

print(counter.value)

在這個(gè)示例中，我們創(chuàng)建了一個(gè) Counter 類，其中包含一個(gè)整數(shù) value 和一個(gè)鎖對(duì)象 lock。 increment 方法使用 with 語句獲取鎖并增加 value 的值。

我們還創(chuàng)建了 10 個(gè)線程，每個(gè)線程都會(huì)調(diào)用 worker 函數(shù)。這個(gè)函數(shù)會(huì)循環(huán) 10000 次調(diào)用 increment 方法來增加 value 的值。

由于每個(gè)線程都會(huì)獲取鎖來訪問共享資源，因此只有一個(gè)線程可以訪問 increment 方法，避免了多個(gè)線程同時(shí)修改 value 的值，從而確保了線程安全。最終的輸出結(jié)果應(yīng)該是 100000，即 10 個(gè)線程分別增加了 10000 次。

條件變量（Condition）實(shí)現(xiàn)多線程間的通信

條件變量（Condition）是Python多線程編程中常用的線程間通信機(jī)制之一，它可以用于線程間的通信和同步，提供了一個(gè)線程等待另一個(gè)線程通知其發(fā)生了某個(gè)事件的方法。

下面是一個(gè)使用條件變量實(shí)現(xiàn)多線程間通信的示例代碼：

import threading
import time

class Producer(threading.Thread):
    def __init__(self, name, cond):
        super().__init__(name=name)
        self.cond = cond

    def run(self):
        for i in range(10):
            with self.cond:
                print(f"{self.name} producing item {i}")
                self.cond.notify() # 通知消費(fèi)者線程
                self.cond.wait() # 等待消費(fèi)者線程通知

class Consumer(threading.Thread):
    def __init__(self, name, cond):
        super().__init__(name=name)
        self.cond = cond

    def run(self):
        for i in range(10):
            with self.cond:
                self.cond.wait() # 等待生產(chǎn)者線程通知
                print(f"{self.name} consuming item {i}")
                self.cond.notify() # 通知生產(chǎn)者線程

cond = threading.Condition()
producer = Producer("Producer", cond)
consumer = Consumer("Consumer", cond)
producer.start()
consumer.start()
producer.join()
consumer.join()

在這個(gè)示例代碼中，生產(chǎn)者線程通過cond.notify()方法通知消費(fèi)者線程，消費(fèi)者線程通過cond.wait()方法等待生產(chǎn)者線程的通知。條件變量cond用于實(shí)現(xiàn)線程之間的同步和通信，生產(chǎn)者線程和消費(fèi)者線程在共享同一把鎖的情況下，通過with self.cond:語句獲取條件變量的鎖并進(jìn)入臨界區(qū)，確保線程安全。

信號(hào)量（Semaphore）實(shí)現(xiàn)多線程間的通信

信號(hào)量（Semaphore）是一種用于控制并發(fā)訪問共享資源的同步原語。它是一種計(jì)數(shù)器，用于控制多個(gè)線程對(duì)共享資源的訪問。信號(hào)量維護(hù)一個(gè)計(jì)數(shù)器，初始值為一個(gè)非負(fù)整數(shù)，每當(dāng)一個(gè)線程訪問共享資源時(shí)，計(jì)數(shù)器減1；當(dāng)計(jì)數(shù)器為0時(shí)，所有試圖訪問共享資源的線程都會(huì)被阻塞，直到某個(gè)線程釋放了共享資源，此時(shí)計(jì)數(shù)器加1，被阻塞的線程才有機(jī)會(huì)繼續(xù)執(zhí)行。

在 Python 中，我們可以使用 threading.Semaphore 類來創(chuàng)建信號(hào)量對(duì)象。該類的構(gòu)造函數(shù)接受一個(gè)整數(shù)作為參數(shù)，表示初始計(jì)數(shù)器的值。Semaphore 類有兩個(gè)方法，acquire() 和 release()，分別用于獲取和釋放信號(hào)量。

以下是使用信號(hào)量實(shí)現(xiàn)的簡單示例代碼：

import threading

class Producer(threading.Thread):
    def __init__(self, name, buf, sem):
        super().__init__(name=name)
        self.buf = buf
        self.sem = sem

    def run(self):
        for i in range(5):
            self.sem.acquire()
            self.buf.append(i)
            print(f"{self.name} produced {i}")
            self.sem.release()

class Consumer(threading.Thread):
    def __init__(self, name, buf, sem):
        super().__init__(name=name)
        self.buf = buf
        self.sem = sem

    def run(self):
        while True:
            self.sem.acquire()
            if not self.buf:
                self.sem.release()
                break
            item = self.buf.pop(0)
            print(f"{self.name} consumed {item}")
            self.sem.release()

if __name__ == '__main__':
    buf = []
    sem = threading.Semaphore(1)
    producer = Producer("Producer", buf, sem)
    consumer1 = Consumer("Consumer1", buf, sem)
    consumer2 = Consumer("Consumer2", buf, sem)
    producer.start()
    consumer1.start()
    consumer2.start()
    producer.join()
    consumer1.join()
    consumer2.join()

在這個(gè)示例代碼中，有一個(gè)生產(chǎn)者線程和兩個(gè)消費(fèi)者線程。生產(chǎn)者線程向共享緩沖區(qū)中添加數(shù)據(jù)，而消費(fèi)者線程從緩沖區(qū)中獲取數(shù)據(jù)。為了避免競(jìng)爭條件，我們使用了信號(hào)量。

在生產(chǎn)者線程中，當(dāng)信號(hào)量可用時(shí)，它會(huì)獲取信號(hào)量并添加數(shù)據(jù)到緩沖區(qū)中，然后釋放信號(hào)量。在消費(fèi)者線程中，當(dāng)信號(hào)量可用時(shí)，它會(huì)獲取信號(hào)量并從緩沖區(qū)中獲取數(shù)據(jù)，然后釋放信號(hào)量。

通過使用信號(hào)量，我們可以確保生產(chǎn)者和消費(fèi)者線程之間的同步，從而避免了競(jìng)爭條件和死鎖問題。

管道（Pipe）

在 Python3 中，可以使用 multiprocessing 模塊中的 Pipe 類來實(shí)現(xiàn)多進(jìn)程間的通信，也可以用 multiprocessing.connection 模塊來創(chuàng)建多進(jìn)程間的通信通道。下面的例子是用 Pipe 類來實(shí)現(xiàn)多線程間的通信：

import threading
from multiprocessing import Pipe

def producer(pipe):
    for i in range(5):
        pipe.send(i)
        print(f"{threading.current_thread().name} produced {i}")
    pipe.close()

def consumer(pipe):
    while True:
        try:
            item = pipe.recv()
            print(f"{threading.current_thread().name} consumed {item}")
        except EOFError:
            break

if __name__ == '__main__':
    producer_conn, consumer_conn = Pipe()
    producer_thread = threading.Thread(target=producer, args=(producer_conn,))
    consumer_thread = threading.Thread(target=consumer, args=(consumer_conn,))
    producer_thread.start()
    consumer_thread.start()
    producer_thread.join()
    consumer_thread.join()

在這個(gè)例子中，我們創(chuàng)建了兩個(gè)線程，一個(gè)生產(chǎn)者線程和一個(gè)消費(fèi)者線程。它們之間共享一個(gè)管道（Pipe），其中生產(chǎn)者將數(shù)據(jù)寫入管道，而消費(fèi)者從管道中讀取數(shù)據(jù)。當(dāng)生產(chǎn)者完成其工作后，它會(huì)關(guān)閉管道以通知消費(fèi)者停止運(yùn)行。

需要注意的是，在 Pipe 中，發(fā)送和接收操作是阻塞的。因此，在發(fā)送或接收數(shù)據(jù)時(shí)，如果沒有可用的空間或數(shù)據(jù)，線程將被阻塞，直到有數(shù)據(jù)可用或空間可用。

定時(shí)器Timer

定時(shí)器（Timer）是Python中的一個(gè)線程類，它可以在一定時(shí)間之后調(diào)用指定的函數(shù)或方法。Timer是繼承自Thread類的，因此可以像Thread一樣啟動(dòng)、停止和等待它。

定時(shí)器的構(gòu)造函數(shù)如下：

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

這個(gè)程序中，我們創(chuàng)建了一個(gè)定時(shí)器t，它會(huì)在5秒后調(diào)用hello函數(shù)，并啟動(dòng)定時(shí)器。程序?qū)⒃趩?dòng)定時(shí)器后立即繼續(xù)執(zhí)行，而定時(shí)器則在后臺(tái)等待5秒，然后調(diào)用hello函數(shù)。

如果我們想要停止定時(shí)器，可以使用cancel()方法：

t.cancel()  # 停止定時(shí)器

需要注意的是，如果定時(shí)器已經(jīng)超時(shí)并且在調(diào)用函數(shù)之前被取消，那么函數(shù)將不會(huì)被調(diào)用。因此，需要在調(diào)用cancel()方法之前等待定時(shí)器超時(shí)。

python3線程池

concurrent.futures實(shí)現(xiàn)多線程

Python 3中的線程池是一種常見的多線程編程模型，可以提高多線程程序的性能和可維護(hù)性。在Python 3中，線程池可以通過標(biāo)準(zhǔn)庫中的concurrent.futures模塊來實(shí)現(xiàn)。

concurrent.futures模塊定義了兩個(gè)類：ThreadPoolExecutor和ProcessPoolExecutor。這兩個(gè)類都實(shí)現(xiàn)了Python 3中的執(zhí)行器(Executor)接口，提供了一種方便的方式來異步執(zhí)行函數(shù)或方法，并返回其結(jié)果。

Exectuor 提供了如下常用方法：

• submit(fn, *args, **kwargs)：將 fn 函數(shù)提交給線程池。*args 代表傳給 fn 函數(shù)的參數(shù)，*kwargs 代表以關(guān)鍵字參數(shù)的形式為 fn 函數(shù)傳入?yún)?shù)。
• map(func, *iterables, timeout=None, chunksize=1)：該函數(shù)類似于全局函數(shù) map(func, *iterables)，只是該函數(shù)將會(huì)啟動(dòng)多個(gè)線程，以異步方式立即對(duì) iterables 執(zhí)行 map 處理。超時(shí)拋出TimeoutError錯(cuò)誤。返回每個(gè)函數(shù)的結(jié)果，注意不是返回future。
• shutdown(wait=True)：關(guān)閉線程池。關(guān)閉之后線程池不再接受新任務(wù)，但會(huì)將之前提交的任務(wù)完成。

程序?qū)ask函數(shù)submit給線程池后，會(huì)返回一個(gè)Future對(duì)象，F(xiàn)uture主要用來獲取task的返回值。

Future 提供了如下方法：

• cancel()：取消該 Future 代表的線程任務(wù)。如果該任務(wù)正在執(zhí)行，不可取消，則該方法返回 False；否則，程序會(huì)取消該任務(wù)，并返回 True。
• cancelled()：返回 Future 代表的線程任務(wù)是否被成功取消。
• running()：如果該 Future 代表的線程任務(wù)正在執(zhí)行、不可被取消，該方法返回 True。
• done()：如果該 Funture 代表的線程任務(wù)被成功取消或執(zhí)行完成，則該方法返回 True。
• result(timeout=None)：獲取該 Future 代表的線程任務(wù)最后返回的結(jié)果。如果 Future 代表的線程任務(wù)還未完成，該方法將會(huì)阻塞當(dāng)前線程，其中 timeout 參數(shù)指定最多阻塞多少秒。超時(shí)拋出TimeoutError，取消拋出CancelledError。
• exception(timeout=None)：獲取該 Future 代表的線程任務(wù)所引發(fā)的異常。如果該任務(wù)成功完成，沒有異常，則該方法返回 None。
• add_done_callback(fn)：為該 Future 代表的線程任務(wù)注冊(cè)一個(gè)“回調(diào)函數(shù)”，當(dāng)該任務(wù)成功完成時(shí)，程序會(huì)自動(dòng)觸發(fā)該 fn 函數(shù)，參數(shù)是future。

使用線程池來執(zhí)行線程任務(wù)的步驟如下：

• 調(diào)用 ThreadPoolExecutor 類的構(gòu)造器創(chuàng)建一個(gè)線程池。
• 定義一個(gè)普通函數(shù)作為線程任務(wù)。
• 調(diào)用 ThreadPoolExecutor 對(duì)象的 submit() 方法來提交線程任務(wù)。
• 當(dāng)不想提交任何任務(wù)時(shí)，調(diào)用 ThreadPoolExecutor 對(duì)象的 shutdown() 方法來關(guān)閉線程池。

ThreadPoolExecutor是一個(gè)線程池執(zhí)行器，可以用來執(zhí)行異步任務(wù)，它管理著一個(gè)線程池，其中包含若干個(gè)線程。當(dāng)一個(gè)任務(wù)被提交給執(zhí)行器時(shí)，執(zhí)行器會(huì)將其分配給一個(gè)線程來執(zhí)行。當(dāng)線程池中的所有線程都在執(zhí)行任務(wù)時(shí)，新提交的任務(wù)會(huì)被放入隊(duì)列中，直到有可用的線程為止。

以下是一個(gè)使用ThreadPoolExecutor的簡單示例：

from concurrent.futures import ThreadPoolExecutor
import time

def worker(num):
    print(f"Worker {num} starting")
    time.sleep(1)
    print(f"Worker {num} finished")

if __name__ == '__main__':
    with ThreadPoolExecutor(max_workers=3) as executor:
        futures = [executor.submit(worker, i) for i in range(5)]
    for future in concurrent.futures.as_completed(futures):
        try:
            result = future.result()
        except Exception as e:
            print(f"Task raised an exception: {e}")
        else: 
            print(f"Task returned: {result}")

在這個(gè)例子中，我們創(chuàng)建了一個(gè)線程池執(zhí)行器，并指定了最大線程數(shù)為3。然后我們循環(huán)提交5個(gè)任務(wù)給執(zhí)行器，每個(gè)任務(wù)都是一個(gè)worker函數(shù)，并傳入不同的參數(shù)。由于我們?cè)O(shè)置了最大線程數(shù)為3，所以只會(huì)有3個(gè)任務(wù)同時(shí)被執(zhí)行，另外2個(gè)任務(wù)會(huì)在之后的某個(gè)時(shí)間點(diǎn)被執(zhí)行。

執(zhí)行結(jié)果會(huì)存儲(chǔ)在Future對(duì)象中，我們可以通過as_completed()方法獲取任務(wù)的執(zhí)行結(jié)果。如果任務(wù)執(zhí)行過程中發(fā)生了異常，result()方法會(huì)拋出相應(yīng)的異常。否則，它會(huì)返回任務(wù)的執(zhí)行結(jié)果。

ThreadPoolExecutor還有其他一些有用的方法，例如shutdown()方法可以等待所有已提交的任務(wù)執(zhí)行完畢并關(guān)閉線程池。

總之，Python 3中的線程池提供了一種方便的方式來執(zhí)行異步任務(wù)，可以大大提高多線程程序的性能和可維護(hù)性。

使用線程池的好處和場(chǎng)景

使用線程池的優(yōu)點(diǎn)是可以避免線程的頻繁創(chuàng)建和銷毀，從而提高線程的利用率，減少系統(tǒng)的開銷。因此，當(dāng)需要頻繁執(zhí)行短時(shí)間的任務(wù)時(shí)，可以考慮使用線程池。例如：

• 網(wǎng)絡(luò)服務(wù)器：在服務(wù)器端接收客戶端請(qǐng)求后，可以使用線程池來處理客戶端請(qǐng)求，以提高服務(wù)器的并發(fā)性能。
• 圖像處理：在圖像處理過程中，需要頻繁啟動(dòng)和停止線程來處理每個(gè)像素點(diǎn)的計(jì)算，使用線程池可以減少線程的創(chuàng)建和銷毀，提高處理效率。
• 數(shù)據(jù)庫連接池：在數(shù)據(jù)庫操作中，需要頻繁創(chuàng)建和銷毀數(shù)據(jù)庫連接，使用線程池可以減少這種開銷，提高數(shù)據(jù)庫操作的效率。

總之，當(dāng)需要頻繁執(zhí)行短時(shí)間的任務(wù)時(shí)，可以考慮使用線程池來優(yōu)化程序性能。

以上就是Python3多線程詳解的詳細(xì)內(nèi)容，更多關(guān)于Python3多線程的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

最新評(píng)論