Python 多線(xiàn)程編程的十個(gè)關(guān)鍵點(diǎn)
多線(xiàn)程編程是 Python 中一個(gè)非常重要的主題，它可以幫助你在處理并發(fā)任務(wù)時(shí)提高程序的效率。本文將詳細(xì)介紹 Python 多線(xiàn)程編程的 10 個(gè)關(guān)鍵點(diǎn)，從基礎(chǔ)到高級(jí)，逐步引導(dǎo)你掌握這一強(qiáng)大工具。
1. 線(xiàn)程的基本概念
什么是線(xiàn)程？
線(xiàn)程是操作系統(tǒng)能夠進(jìn)行運(yùn)算調(diào)度的最小單位。它被包含在進(jìn)程之中，是進(jìn)程中的實(shí)際運(yùn)作單位。一個(gè)進(jìn)程可以包含多個(gè)線(xiàn)程，這些線(xiàn)程共享進(jìn)程的資源，如內(nèi)存和文件句柄。
Python 中的線(xiàn)程
在 Python 中，我們可以使用 threading 模塊來(lái)創(chuàng)建和管理線(xiàn)程。下面是一個(gè)簡(jiǎn)單的示例：
import threading
def print_numbers():
for i in range(10):
print(i)
# 創(chuàng)建線(xiàn)程
thread = threading.Thread(target=print_numbers)
# 啟動(dòng)線(xiàn)程
thread.start()
# 等待線(xiàn)程完成
thread.join()輸出結(jié)果：
0
1
2
3
4
5
6
7
8
92. 創(chuàng)建和啟動(dòng)線(xiàn)程
創(chuàng)建線(xiàn)程
可以通過(guò)繼承 Thread 類(lèi)來(lái)創(chuàng)建自定義線(xiàn)程類(lèi)，然后重寫(xiě) run 方法來(lái)定義線(xiàn)程的行為。
import threading
class MyThread(threading.Thread):
def run(self):
for i in range(10):
print(f"Thread {self.name}: {i}")
# 創(chuàng)建線(xiàn)程實(shí)例
thread1 = MyThread(name="Thread 1")
thread2 = MyThread(name="Thread 2")
# 啟動(dòng)線(xiàn)程
thread1.start()
thread2.start()
# 等待線(xiàn)程完成
thread1.join()
thread2.join()輸出結(jié)果：
Thread Thread 1: 0
Thread Thread 2: 0
Thread Thread 1: 1
Thread Thread 2: 1
...
Thread Thread 1: 9
Thread Thread 2: 93. 線(xiàn)程同步
線(xiàn)程同步問(wèn)題
多個(gè)線(xiàn)程同時(shí)訪(fǎng)問(wèn)共享資源時(shí)可能會(huì)導(dǎo)致數(shù)據(jù)不一致的問(wèn)題。為了防止這種情況，可以使用鎖（Lock）來(lái)同步線(xiàn)程。
import threading
# 共享資源
counter = 0
# 鎖
lock = threading.Lock()
def increment_counter():
global counter
for _ in range(100000):
lock.acquire() # 獲取鎖
counter += 1
lock.release() # 釋放鎖
# 創(chuàng)建線(xiàn)程
thread1 = threading.Thread(target=increment_counter)
thread2 = threading.Thread(target=increment_counter)
# 啟動(dòng)線(xiàn)程
thread1.start()
thread2.start()
# 等待線(xiàn)程完成
thread1.join()
thread2.join()
print(f"Final counter value: {counter}")輸出結(jié)果：
Final counter value: 2000004. 線(xiàn)程間通信
使用隊(duì)列進(jìn)行線(xiàn)程間通信
queue 模塊提供了線(xiàn)程安全的隊(duì)列，可以用于線(xiàn)程間的通信。
import threading
import queue
# 創(chuàng)建隊(duì)列
q = queue.Queue()
def producer(q):
for i in range(10):
q.put(i)
print(f"Produced: {i}")
def consumer(q):
while True:
item = q.get()
if item is None:
break
print(f"Consumed: {item}")
q.task_done()
# 創(chuàng)建線(xiàn)程
producer_thread = threading.Thread(target=producer, args=(q,))
consumer_thread = threading.Thread(target=consumer, args=(q,))
# 啟動(dòng)線(xiàn)程
producer_thread.start()
consumer_thread.start()
# 等待生產(chǎn)者完成
producer_thread.join()
# 停止消費(fèi)者
q.put(None)
consumer_thread.join()輸出結(jié)果：
Produced: 0
Consumed: 0
Produced: 1
Consumed: 1
...
Produced: 9
Consumed: 95. 線(xiàn)程池
使用 concurrent.futures 模塊
concurrent.futures 模塊提供了線(xiàn)程池的實(shí)現(xiàn)，可以簡(jiǎn)化多線(xiàn)程編程。
import concurrent.futures
def task(n):
return n * n
# 創(chuàng)建線(xiàn)程池
with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
results = list(executor.map(task, range(10)))
print(results)輸出結(jié)果：
[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]6. 線(xiàn)程局部存儲(chǔ)
線(xiàn)程局部存儲(chǔ)
線(xiàn)程局部存儲(chǔ)（Thread-local storage）允許每個(gè)線(xiàn)程擁有其自己的局部變量副本，這些變量不會(huì)被其他線(xiàn)程訪(fǎng)問(wèn)。
import threading
# 創(chuàng)建線(xiàn)程局部存儲(chǔ)
local_storage = threading.local()
def set_local_value(value):
local_storage.value = value
print(f"Thread {threading.current_thread().name} set value to {value}")
def get_local_value():
try:
print(f"Thread {threading.current_thread().name} got value: {local_storage.value}")
except AttributeError:
print(f"Thread {threading.current_thread().name} has no value")
# 創(chuàng)建線(xiàn)程
thread1 = threading.Thread(target=set_local_value, args=(10,), name="Thread 1")
thread2 = threading.Thread(target=get_local_value, name="Thread 2")
# 啟動(dòng)線(xiàn)程
thread1.start()
thread1.join()
thread2.start()
thread2.join()輸出結(jié)果：
Thread Thread 1 set value to 10
Thread Thread 2 got value: 107. 守護(hù)線(xiàn)程
守護(hù)線(xiàn)程
守護(hù)線(xiàn)程（Daemon thread）是一種在后臺(tái)運(yùn)行的線(xiàn)程，當(dāng)所有非守護(hù)線(xiàn)程結(jié)束后，守護(hù)線(xiàn)程會(huì)自動(dòng)結(jié)束。
import threading
import time
def daemon_thread():
while True:
print("Daemon thread running...")
time.sleep(1)
# 創(chuàng)建守護(hù)線(xiàn)程
daemon = threading.Thread(target=daemon_thread, daemon=True)
# 啟動(dòng)守護(hù)線(xiàn)程
daemon.start()
# 主線(xiàn)程運(yùn)行一段時(shí)間后結(jié)束
time.sleep(5)
print("Main thread finished")輸出結(jié)果：
Daemon thread running...
Daemon thread running...
Daemon thread running...
Daemon thread running...
Daemon thread running...
Main thread finished8. 線(xiàn)程優(yōu)先級(jí)
線(xiàn)程優(yōu)先級(jí)
Python 的 threading 模塊不直接支持設(shè)置線(xiàn)程優(yōu)先級(jí)，但可以通過(guò)一些間接方法來(lái)實(shí)現(xiàn)。例如，可以使用 queue.PriorityQueue 來(lái)管理任務(wù)的優(yōu)先級(jí)。
import threading
import queue
# 創(chuàng)建優(yōu)先級(jí)隊(duì)列
q = queue.PriorityQueue()
def worker():
while True:
priority, item = q.get()
if item is None:
break
print(f"Processing item with priority {priority}: {item}")
q.task_done()
# 創(chuàng)建線(xiàn)程
worker_thread = threading.Thread(target=worker, daemon=True)
# 啟動(dòng)線(xiàn)程
worker_thread.start()
# 添加任務(wù)
q.put((1, "High priority task"))
q.put((3, "Low priority task"))
q.put((2, "Medium priority task"))
# 等待所有任務(wù)完成
q.join()
# 停止線(xiàn)程
q.put((None, None))
worker_thread.join()輸出結(jié)果：
Processing item with priority 1: High priority task
Processing item with priority 2: Medium priority task
Processing item with priority 3: Low priority task9. 線(xiàn)程安全的數(shù)據(jù)結(jié)構(gòu)
線(xiàn)程安全的數(shù)據(jù)結(jié)構(gòu)
Python 的 queue 模塊提供了線(xiàn)程安全的隊(duì)列，此外，還可以使用 multiprocessing 模塊中的 Manager 類(lèi)來(lái)創(chuàng)建線(xiàn)程安全的數(shù)據(jù)結(jié)構(gòu)。
import threading
from multiprocessing import Manager
# 創(chuàng)建線(xiàn)程安全的列表
manager = Manager()
safe_list = manager.list()
def add_to_list(item):
safe_list.append(item)
print(f"Added {item} to the list")
# 創(chuàng)建線(xiàn)程
thread1 = threading.Thread(target=add_to_list, args=("A",))
thread2 = threading.Thread(target=add_to_list, args=("B",))
# 啟動(dòng)線(xiàn)程
thread1.start()
thread2.start()
# 等待線(xiàn)程完成
thread1.join()
thread2.join()
print(f"Final list: {list(safe_list)}")輸出結(jié)果：
Added A to the list
Added B to the list
Final list: ['A', 'B']10. 實(shí)戰(zhàn)案例：多線(xiàn)程下載圖片
案例背景
假設(shè)我們需要從多個(gè) URL 下載圖片并保存到本地。使用多線(xiàn)程可以顯著提高下載速度。
import os
import requests
import threading
# 圖片 URL 列表
image_urls = [
"https://example.com/image1.jpg",
"https://example.com/image2.jpg",
"https://example.com/image3.jpg"
]
# 保存圖片的目錄
save_dir = "images"
# 創(chuàng)建目錄
if not os.path.exists(save_dir):
os.makedirs(save_dir)
def download_image(url):
response = requests.get(url)
if response.status_code == 200:
filename = os.path.join(save_dir, url.split("/")[-1])
with open(filename, "wb") as f:
f.write(response.content)
print(f"Downloaded {url} to {filename}")
else:
print(f"Failed to download {url}")
# 創(chuàng)建線(xiàn)程
threads = []
for url in image_urls:
thread = threading.Thread(target=download_image, args=(url,))
threads.append(thread)
thread.start()
# 等待所有線(xiàn)程完成
for thread in threads:
thread.join()輸出結(jié)果：
Downloaded https://example.com/image1.jpg to images/image1.jpg
Downloaded https://example.com/image2.jpg to images/image2.jpg
Downloaded https://example.com/image3.jpg to images/image3.jpg總結(jié)
本文詳細(xì)介紹了 Python 多線(xiàn)程編程的 10 個(gè)關(guān)鍵點(diǎn)，包括線(xiàn)程的基本概念、創(chuàng)建和啟動(dòng)線(xiàn)程、線(xiàn)程同步、線(xiàn)程間通信、線(xiàn)程池、線(xiàn)程局部存儲(chǔ)、守護(hù)線(xiàn)程、線(xiàn)程優(yōu)先級(jí)、線(xiàn)程安全的數(shù)據(jù)結(jié)構(gòu)以及一個(gè)實(shí)戰(zhàn)案例。通過(guò)這些內(nèi)容，你應(yīng)該能夠更好地理解和應(yīng)用 Python 的多線(xiàn)程編程技術(shù)。
