Python 多线程：完整指南与实战示例

import threading
import time
 
def cpu_bound(n):
    """CPU 密集型：GIL 阻止并行执行"""
    total = 0
    for i in range(n):
        total += i * i
    return total
 
def io_bound(url):
    """I/O 密集型：网络等待期间 GIL 会释放"""
    import urllib.request
    return urllib.request.urlopen(url).read()
 
# CPU 密集型：4 个线程逐个运行（无加速效果）
start = time.time()
threads = [threading.Thread(target=cpu_bound, args=(10_000_000,)) for _ in range(4)]
for t in threads: t.start()
for t in threads: t.join()
print(f"CPU 密集型使用线程：{time.time() - start:.2f}s")
 
# I/O 密集型：4 个线程重叠等待时间（大幅加速）

import threading
import time
 
def download_file(filename):
    print(f"[{threading.current_thread().name}] 正在下载 {filename}...")
    time.sleep(2)  # 模拟下载
    print(f"[{threading.current_thread().name}] 完成 {filename}")
 
# 创建线程
t1 = threading.Thread(target=download_file, args=("data.csv",))
t2 = threading.Thread(target=download_file, args=("report.pdf",))
 
# 启动线程
t1.start()
t2.start()
 
# 等待两个线程都完成
t1.join()
t2.join()
 
print("所有下载已完成")

import threading
import time
 
def slow_task():
    time.sleep(10)
 
t = threading.Thread(target=slow_task)
t.start()
 
# 最多等待 3 秒
t.join(timeout=3)
 
if t.is_alive():
    print("3 秒后线程仍在运行")
else:
    print("线程已完成")

import threading
 
def worker():
    name = threading.current_thread().name
    print(f"运行在线程中：{name}")
 
t = threading.Thread(target=worker, name="DataProcessor")
t.start()
t.join()

import threading
import time
 
def background_monitor():
    while True:
        print("监控系统健康状态...")
        time.sleep(5)
 
# 守护线程：主程序退出时自动终止
monitor = threading.Thread(target=background_monitor, daemon=True)
monitor.start()
 
# 主程序执行工作
time.sleep(12)
print("主程序退出")
# monitor 线程被自动终止

import threading
import time
 
class FileProcessor(threading.Thread):
    def __init__(self, filepath):
        super().__init__()
        self.filepath = filepath
        self.result = None
 
    def run(self):
        """重写 run() 方法实现线程逻辑"""
        print(f"正在处理 {self.filepath}")
        time.sleep(1)  # 模拟工作
        self.result = f"已处理：{self.filepath}"
 
# 创建并运行
processor = FileProcessor("/data/report.csv")
processor.start()
processor.join()
print(processor.result)

import threading
 
def fetch_data(url, timeout, retries=3, verbose=False):
    print(f"正在获取 {url}（超时={timeout}秒，重试={retries}，详细={verbose}）")
 
# 位置参数作为元组
t1 = threading.Thread(target=fetch_data, args=("https://api.example.com", 30))
 
# 关键字参数作为字典
t2 = threading.Thread(
    target=fetch_data,
    args=("https://api.example.com",),
    kwargs={"timeout": 30, "retries": 5, "verbose": True}
)
 
t1.start()
t2.start()
t1.join()
t2.join()

import threading
 
results = {}
lock = threading.Lock()
 
def compute(task_id, value):
    result = value ** 2
    with lock:
        results[task_id] = result
 
threads = []
for i in range(5):
    t = threading.Thread(target=compute, args=(i, i * 10))
    threads.append(t)
    t.start()
 
for t in threads:
    t.join()
 
print(results)  # {0: 0, 1: 100, 2: 400, 3: 900, 4: 1600}

from concurrent.futures import ThreadPoolExecutor, as_completed
import time
 
def fetch_url(url):
    time.sleep(1)  # 模拟网络请求
    return f"来自 {url} 的内容"
 
urls = [
    "https://example.com/page1",
    "https://example.com/page2",
    "https://example.com/page3",
    "https://example.com/page4",
    "https://example.com/page5",
]
 
with ThreadPoolExecutor(max_workers=3) as executor:
    # 提交任务并获取 Future 对象
    future_to_url = {executor.submit(fetch_url, url): url for url in urls}
 
    # 按完成顺序处理结果
    for future in as_completed(future_to_url):
        url = future_to_url[future]
        try:
            data = future.result()
            print(f"{url}：{data}")
        except Exception as e:
            print(f"{url} 生成异常：{e}")

from concurrent.futures import ThreadPoolExecutor
 
def process_item(item):
    return item.upper()
 
items = ["apple", "banana", "cherry", "date"]
 
with ThreadPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(process_item, items))
 
print(results)  # ['APPLE', 'BANANA', 'CHERRY', 'DATE']

from concurrent.futures import ThreadPoolExecutor, as_completed
 
def risky_task(n):
    if n == 3:
        raise ValueError(f"错误输入：{n}")
    return n * 10
 
with ThreadPoolExecutor(max_workers=4) as executor:
    futures = {executor.submit(risky_task, i): i for i in range(5)}
 
    for future in as_completed(futures):
        task_id = futures[future]
        try:
            result = future.result(timeout=5)
            print(f"任务 {task_id}：{result}")
        except ValueError as e:
            print(f"任务 {task_id} 失败：{e}")
        except TimeoutError:
            print(f"任务 {task_id} 超时")

from concurrent.futures import ThreadPoolExecutor
import time
 
def long_task(n):
    time.sleep(5)
    return n
 
with ThreadPoolExecutor(max_workers=2) as executor:
    futures = [executor.submit(long_task, i) for i in range(10)]
 
    # 取消待处理任务（已在运行的任务无法取消）
    for f in futures[4:]:
        cancelled = f.cancel()
        print(f"已取消：{cancelled}")

import threading
 
class BankAccount:
    def __init__(self, balance):
        self.balance = balance
        self.lock = threading.Lock()
 
    def withdraw(self, amount):
        with self.lock:  # 同一时间只有一个线程
            if self.balance >= amount:
                self.balance -= amount
                return True
            return False
 
    def deposit(self, amount):
        with self.lock:
            self.balance += amount
 
account = BankAccount(1000)
 
def make_transactions():
    for _ in range(100):
        account.deposit(10)
        account.withdraw(10)
 
threads = [threading.Thread(target=make_transactions) for _ in range(10)]
for t in threads: t.start()
for t in threads: t.join()
 
print(f"最终余额：{account.balance}")  # 始终为 1000

import threading
 
class SafeCache:
    def __init__(self):
        self._data = {}
        self._lock = threading.RLock()
 
    def get(self, key):
        with self._lock:
            return self._data.get(key)
 
    def set(self, key, value):
        with self._lock:
            self._data[key] = value
 
    def get_or_set(self, key, default):
        with self._lock:
            # 这里调用 get()，它也会获取 _lock
            # RLock 允许这样做；普通 Lock 会导致死锁
            existing = self.get(key)
            if existing is None:
                self.set(key, default)
                return default
            return existing

import threading
import time
 
# 最多允许 3 个并发数据库连接
db_semaphore = threading.Semaphore(3)
 
def query_database(query_id):
    with db_semaphore:
        print(f"查询 {query_id}：已连接（活动连接数：{3 - db_semaphore._value}）")
        time.sleep(2)  # 模拟查询
        print(f"查询 {query_id}：完成")
 
threads = [threading.Thread(target=query_database, args=(i,)) for i in range(10)]
for t in threads: t.start()
for t in threads: t.join()

import threading
import time
 
data_ready = threading.Event()
shared_data = []
 
def producer():
    print("生产者：准备数据...")
    time.sleep(3)
    shared_data.extend([1, 2, 3, 4, 5])
    print("生产者：数据就绪，通知消费者")
    data_ready.set()
 
def consumer(name):
    print(f"消费者 {name}：等待数据...")
    data_ready.wait()  # 阻塞直到事件被设置
    print(f"消费者 {name}：获取数据 = {shared_data}")
 
threads = [
    threading.Thread(target=producer),
    threading.Thread(target=consumer, args=("A",)),
    threading.Thread(target=consumer, args=("B",)),
]
for t in threads: t.start()
for t in threads: t.join()

import threading
import time
import random
 
buffer = []
MAX_SIZE = 5
condition = threading.Condition()
 
def producer():
    for i in range(20):
        with condition:
            while len(buffer) >= MAX_SIZE:
                condition.wait()  # 等待直到有空间
            item = random.randint(1, 100)
            buffer.append(item)
            print(f"生产：{item}（缓冲区大小：{len(buffer)}）")
            condition.notify_all()
        time.sleep(0.1)
 
def consumer(name):
    for _ in range(10):
        with condition:
            while len(buffer) == 0:
                condition.wait()  # 等待直到有数据
            item = buffer.pop(0)
            print(f"消费者 {name} 消费：{item}（缓冲区大小：{len(buffer)}）")
            condition.notify_all()
        time.sleep(0.15)
 
t1 = threading.Thread(target=producer)
t2 = threading.Thread(target=consumer, args=("X",))
t3 = threading.Thread(target=consumer, args=("Y",))
t1.start(); t2.start(); t3.start()
t1.join(); t2.join(); t3.join()

import threading
import queue
import time
 
task_queue = queue.Queue()
results = queue.Queue()
 
def worker():
    while True:
        item = task_queue.get()  # 阻塞直到有数据
        if item is None:
            break
        result = item ** 2
        results.put(result)
        task_queue.task_done()
 
# 启动 4 个工作者
workers = []
for _ in range(4):
    t = threading.Thread(target=worker, daemon=True)
    t.start()
    workers.append(t)
 
# 提交任务
for i in range(20):
    task_queue.put(i)
 
# 等待所有任务完成
task_queue.join()
 
# 停止工作者
for _ in range(4):
    task_queue.put(None)
for w in workers:
    w.join()
 
# 收集结果
all_results = []
while not results.empty():
    all_results.append(results.get())
print(f"结果：{sorted(all_results)}")

from collections import deque
import threading
import time
 
buffer = deque(maxlen=1000)
 
def producer():
    for i in range(100):
        buffer.append(i)
        time.sleep(0.01)
 
def consumer():
    consumed = 0
    while consumed < 100:
        if buffer:
            item = buffer.popleft()
            consumed += 1
        else:
            time.sleep(0.01)
    print(f"消费了 {consumed} 个项目")
 
t1 = threading.Thread(target=producer)
t2 = threading.Thread(target=consumer)
t1.start(); t2.start()
t1.join(); t2.join()

import threading
import queue
import time
import random
 
def producer(q, name, num_items):
    for i in range(num_items):
        item = f"{name}-item-{i}"
        q.put(item)
        print(f"生产者 {name}：创建 {item}")
        time.sleep(random.uniform(0.05, 0.15))
    print(f"生产者 {name}：完成")
 
def consumer(q, name, stop_event):
    while not stop_event.is_set() or not q.empty():
        try:
            item = q.get(timeout=0.5)
            print(f"消费者 {name}：处理 {item}")
            time.sleep(random.uniform(0.1, 0.2))
            q.task_done()
        except queue.Empty:
            continue
    print(f"消费者 {name}：关闭")
 
task_queue = queue.Queue(maxsize=10)
stop_event = threading.Event()
 
producers = [
    threading.Thread(target=producer, args=(task_queue, "P1", 10)),
    threading.Thread(target=producer, args=(task_queue, "P2", 10)),
]
consumers = [
    threading.Thread(target=consumer, args=(task_queue, "C1", stop_event)),
    threading.Thread(target=consumer, args=(task_queue, "C2", stop_event)),
    threading.Thread(target=consumer, args=(task_queue, "C3", stop_event)),
]
 
for c in consumers: c.start()
for p in producers: p.start()
for p in producers: p.join()
 
task_queue.join()  # 等待所有项目被处理
stop_event.set()   # 通知消费者停止
for c in consumers: c.join()

import threading
import queue
 
class WorkerPool:
    def __init__(self, num_workers):
        self.task_queue = queue.Queue()
        self.result_queue = queue.Queue()
        self.workers = []
 
        for _ in range(num_workers):
            t = threading.Thread(target=self._worker, daemon=True)
            t.start()
            self.workers.append(t)
 
    def _worker(self):
        while True:
            func, args, kwargs, future_id = self.task_queue.get()
            if func is None:
                break
            try:
                result = func(*args, **kwargs)
                self.result_queue.put((future_id, result, None))
            except Exception as e:
                self.result_queue.put((future_id, None, e))
            finally:
                self.task_queue.task_done()
 
    def submit(self, func, *args, **kwargs):
        future_id = id(func)  # 简单 ID
        self.task_queue.put((func, args, kwargs, future_id))
        return future_id
 
    def shutdown(self):
        for _ in self.workers:
            self.task_queue.put((None, None, None, None))
        for w in self.workers:
            w.join()
 
# 使用
pool = WorkerPool(4)
for i in range(10):
    pool.submit(lambda x: x * x, i)
pool.task_queue.join()
pool.shutdown()

import threading
import time
from concurrent.futures import ThreadPoolExecutor
 
class RateLimiter:
    def __init__(self, max_per_second):
        self.interval = 1.0 / max_per_second
        self.lock = threading.Lock()
        self.last_call = 0
 
    def wait(self):
        with self.lock:
            elapsed = time.time() - self.last_call
            wait_time = self.interval - elapsed
            if wait_time > 0:
                time.sleep(wait_time)
            self.last_call = time.time()
 
limiter = RateLimiter(max_per_second=5)
 
def rate_limited_fetch(url):
    limiter.wait()
    print(f"正在获取 {url}，时间 {time.time():.2f}")
    time.sleep(0.5)  # 模拟请求
    return f"来自 {url} 的数据"
 
urls = [f"https://api.example.com/item/{i}" for i in range(20)]
 
with ThreadPoolExecutor(max_workers=10) as executor:
    results = list(executor.map(rate_limited_fetch, urls))

import threading
 
# 错误：竞态条件
counter = 0
 
def increment_unsafe():
    global counter
    for _ in range(100_000):
        counter += 1  # 读、增、写：不是原子操作
 
threads = [threading.Thread(target=increment_unsafe) for _ in range(5)]
for t in threads: t.start()
for t in threads: t.join()
print(f"预期：500000，实际：{counter}")  # 通常小于 500000
 
# 正确：使用锁保护
counter = 0
lock = threading.Lock()
 
def increment_safe():
    global counter
    for _ in range(100_000):
        with lock:
            counter += 1
 
threads = [threading.Thread(target=increment_safe) for _ in range(5)]
for t in threads: t.start()
for t in threads: t.join()
print(f"预期：500000，实际：{counter}")  # 始终为 500000

import threading
 
lock_a = threading.Lock()
lock_b = threading.Lock()
 
def thread_1():
    with lock_a:
        print("线程 1：获取 lock_a")
        with lock_b:  # 如果 thread_2 持有 lock_b 则永远等待
            print("线程 1：获取 lock_b")
 
def thread_2():
    with lock_b:
        print("线程 2：获取 lock_b")
        with lock_a:  # 如果 thread_1 持有 lock_a 则永远等待
            print("线程 2：获取 lock_a")
 
# 这将导致死锁
# t1 = threading.Thread(target=thread_1)
# t2 = threading.Thread(target=thread_2)
# t1.start(); t2.start()

def thread_1_fixed():
    with lock_a:    # 始终先 lock_a
        with lock_b:
            print("线程 1：获取两个锁")
 
def thread_2_fixed():
    with lock_a:    # 始终先 lock_a（相同顺序）
        with lock_b:
            print("线程 2：获取两个锁")

def safe_acquire():
    acquired_a = lock_a.acquire(timeout=2)
    if not acquired_a:
        print("无法获取 lock_a，回退")
        return
    try:
        acquired_b = lock_b.acquire(timeout=2)
        if not acquired_b:
            print("无法获取 lock_b，释放 lock_a")
            return
        try:
            print("安全获取两个锁")
        finally:
            lock_b.release()
    finally:
        lock_a.release()

from concurrent.futures import ThreadPoolExecutor, as_completed
import urllib.request
import time
 
def fetch_page(url):
    """获取网页并返回内容长度"""
    try:
        with urllib.request.urlopen(url, timeout=10) as response:
            content = response.read()
            return url, len(content), None
    except Exception as e:
        return url, 0, str(e)
 
urls = [
    "https://python.org",
    "https://docs.python.org",
    "https://pypi.org",
    "https://realpython.com",
    "https://github.com",
    "https://stackoverflow.com",
    "https://news.ycombinator.com",
    "https://httpbin.org",
]
 
# 顺序执行
start = time.time()
for url in urls:
    fetch_page(url)
sequential_time = time.time() - start
 
# 使用多线程并发
start = time.time()
with ThreadPoolExecutor(max_workers=8) as executor:
    futures = {executor.submit(fetch_page, url): url for url in urls}
    for future in as_completed(futures):
        url, size, error = future.result()
        if error:
            print(f"  失败 {url}：{error}")
        else:
            print(f"  成功 {url}：{size:,} 字节")
threaded_time = time.time() - start
 
print(f"\n顺序执行：{sequential_time:.2f}s")
print(f"多线程：  {threaded_time:.2f}s")
print(f"加速比：  {sequential_time / threaded_time:.1f}x")

from concurrent.futures import ThreadPoolExecutor
import os
import hashlib
 
def process_file(filepath):
    """读取文件并计算其 SHA-256 哈希值"""
    with open(filepath, 'rb') as f:
        content = f.read()
    file_hash = hashlib.sha256(content).hexdigest()
    size = os.path.getsize(filepath)
    return filepath, file_hash, size
 
def hash_all_files(directory, pattern="*.py"):
    """使用多线程哈希目录中所有匹配的文件"""
    import glob
    files = glob.glob(os.path.join(directory, "**", pattern), recursive=True)
 
    results = {}
    with ThreadPoolExecutor(max_workers=8) as executor:
        futures = {executor.submit(process_file, f): f for f in files}
        for future in futures:
            try:
                path, hash_val, size = future.result()
                results[path] = {"hash": hash_val, "size": size}
            except Exception as e:
                print(f"处理 {futures[future]} 时出错：{e}")
 
    return results
 
# 使用
# file_hashes = hash_all_files("/path/to/project")

from concurrent.futures import ThreadPoolExecutor, as_completed
import urllib.request
import json
import time
 
def fetch_api(endpoint, max_retries=3, backoff=1.0):
    """获取 API 端点，带指数退避重试"""
    for attempt in range(max_retries):
        try:
            url = f"https://jsonplaceholder.typicode.com{endpoint}"
            req = urllib.request.Request(url)
            with urllib.request.urlopen(req, timeout=10) as response:
                data = json.loads(response.read())
                return {"endpoint": endpoint, "data": data, "error": None}
        except Exception as e:
            if attempt < max_retries - 1:
                wait = backoff * (2 ** attempt)
                time.sleep(wait)
            else:
                return {"endpoint": endpoint, "data": None, "error": str(e)}
 
endpoints = [f"/posts/{i}" for i in range(1, 21)]
 
start = time.time()
with ThreadPoolExecutor(max_workers=10) as executor:
    futures = [executor.submit(fetch_api, ep) for ep in endpoints]
    results = [f.result() for f in futures]
 
elapsed = time.time() - start
success = sum(1 for r in results if r["error"] is None)
print(f"在 {elapsed:.2f} 秒内获取了 {success}/{len(endpoints)} 个端点")

import threading
import time
 
class PeriodicTask:
    """在后台线程中以固定间隔运行函数"""
    def __init__(self, interval, func, *args, **kwargs):
        self.interval = interval
        self.func = func
        self.args = args
        self.kwargs = kwargs
        self._stop_event = threading.Event()
        self._thread = None
 
    def start(self):
        self._thread = threading.Thread(target=self._run, daemon=True)
        self._thread.start()
 
    def _run(self):
        while not self._stop_event.is_set():
            self.func(*self.args, **self.kwargs)
            self._stop_event.wait(self.interval)
 
    def stop(self):
        self._stop_event.set()
        if self._thread:
            self._thread.join()
 
# 使用
def check_health():
    print(f"健康检查时间 {time.strftime('%H:%M:%S')}")
 
task = PeriodicTask(2.0, check_health)
task.start()
time.sleep(7)
task.stop()
print("已停止")

import time
import threading
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
 
def io_task():
    time.sleep(0.2)
 
def cpu_task(n=2_000_000):
    return sum(i * i for i in range(n))
 
# 对比 Threading 和 Multiprocessing 的基准测试
NUM_TASKS = 20
 
# Threading - I/O 密集型
start = time.time()
with ThreadPoolExecutor(max_workers=4) as pool:
    list(pool.map(lambda _: io_task(), range(NUM_TASKS)))
print(f"Threading (I/O)：{time.time() - start:.2f}s")
 
# Threading - CPU 密集型
start = time.time()
with ThreadPoolExecutor(max_workers=4) as pool:
    list(pool.map(lambda _: cpu_task(), range(NUM_TASKS)))
print(f"Threading (CPU)：{time.time() - start:.2f}s")