多线程编程

Python的 threading 模块提供了丰富的多线程编程工具。本文将深入讲解线程管理、同步机制和线程池的使用。

创建线程

import threading

def worker(num):
    print(f"工作线程 {num} 执行中")

# 创建并启动线程
threads = []
for i in range(3):
    t = threading.Thread(target=worker, args=(i,), name=f"Thread-{i}")
    threads.append(t)
    t.start()

# 等待所有线程完成
for t in threads:
    t.join()

print("所有线程完成")

线程属性与方法

import threading
import time

def task():
    time.sleep(1)

t = threading.Thread(target=task)
print(f"线程名: {t.name}")      # Thread-1
print(f"线程ID: {t.ident}")     # 线程ID
print(f"是否存活: {t.is_alive()}")  # False
print(f"是否守护线程: {t.daemon}")  # False

# 设置守护线程（主线程结束时自动结束）
t.daemon = True

Lock 互斥锁

import threading

class BankAccount:
    def __init__(self):
        self.balance = 1000
        self.lock = threading.Lock()

    def withdraw(self, amount):
        with self.lock:  # 自动获取和释放锁
            if self.balance >= amount:
                self.balance -= amount
                return True
            return False

account = BankAccount()
threads = [
    threading.Thread(target=lambda: account.withdraw(500))
    for _ in range(3)
]

for t in threads:
    t.start()
for t in threads:
    t.join()

print(f"余额: {account.balance}")  # 余额正确

RLock 可重入锁

import threading

class SafeCounter:
    def __init__(self):
        self.lock = threading.RLock()
        self.count = 0

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

    def other_operation(self):
        with self.lock:  # 同一线程可以再次获取
            self.count += 1

counter = SafeCounter()
threads = [threading.Thread(target=counter.increment) for _ in range(10)]

for t in threads:
    t.start()
for t in threads:
    t.join()

print(counter.count)  # 20

Queue 线程安全队列

import threading
import queue
import time

def producer(q):
    for i in range(5):
        q.put(f"任务-{i}")
        print(f"生产: 任务-{i}")
        time.sleep(0.1)
    q.put(None)  # 发送结束信号

def consumer(q):
    while True:
        item = q.get()
        if item is None:
            break
        print(f"消费: {item}")
        q.task_done()

q = queue.Queue()

# 启动生产者和消费者
producer_thread = threading.Thread(target=producer, args=(q,))
consumer_thread = threading.Thread(target=consumer, args=(q,))

producer_thread.start()
consumer_thread.start()

producer_thread.join()
consumer_thread.join()

线程池 ThreadPoolExecutor

from concurrent.futures import ThreadPoolExecutor
import requests

def fetch_url(url):
    response = requests.get(url, timeout=5)
    return response.status_code

urls = [
    "https://httpbin.org/get",
    "https://httpbin.org/status/200",
    "https://httpbin.org/status/404",
]

with ThreadPoolExecutor(max_workers=3) as executor:
    results = list(executor.map(fetch_url, urls))
    print(results)  # [200, 200, 404]

线程池 with 回调

from concurrent.futures import ThreadPoolExecutor
import time

def slow_task(n):
    time.sleep(1)
    return n * n

def callback(future):
    print(f"结果: {future.result()}")

with ThreadPoolExecutor(max_workers=3) as executor:
    futures = [executor.submit(slow_task, i) for i in range(5)]
    for future in futures:
        future.add_done_callback(callback)

掌握多线程编程能让你写出高效的并发程序，特别是处理I/O密集型任务时。