Python案例如何开启多线程？

wen python案例 2026-06-10 10

Python案例如何开启多线程？从入门到实战，一文掌握核心技巧

目录导读

多线程基础：为什么需要多线程？
Python中的线程模块：threading vs _thread
实战案例一：用多线程加速文件下载
实战案例二：多线程爬虫中的生产者-消费者模式
常见陷阱与最佳实践
线程安全与锁机制详解
Q&A：高频面试与误区解答

多线程基础：为什么需要多线程？

在编写Python程序时，我们常遇到需要同时执行多个任务的场景，同时下载多个文件、并发处理用户请求、实时监控多个传感器数据。多线程允许程序在同一进程内创建多个独立执行流，实现任务并行处理，从而大幅提升I/O密集型任务的效率。

Python案例如何开启多线程？

关键概念：

线程：操作系统调度的最小单位,共享进程内存空间。
并行 vs 并发：Python的GIL（全局解释器锁）限制了CPU密集型任务的并行，但对I/O密集型任务,多线程仍能通过异步阻塞实现高效并发。

Python中的线程模块：`threading` vs `_thread`

Python官方推荐使用threading模块，而非底层的_thread，原因在于threading提供了更高级的API、线程管理功能以及内置的锁机制。

# 基础示例：创建并启动线程
import threading
import time
def print_numbers():
    for i in range(5):
        time.sleep(1)
        print(f"线程: {threading.current_thread().name} 打印 {i}")
# 创建线程
t1 = threading.Thread(target=print_numbers, name="Thread-1")
t2 = threading.Thread(target=print_numbers, name="Thread-2")
# 启动线程
t1.start()
t2.start()
# 主线程等待子线程结束
t1.join()
t2.join()
print("主线程结束")

输出说明： 两个线程交替打印数字（观察结果会发现输出顺序并不固定），主线程会等待所有子线程完成后才继续执行join后的代码。

实战案例一：用多线程加速文件下载

假设你要从服务器批量下载100个文件，单线程逐个下载需要100秒，使用多线程后，10个线程并发下载，总时间可缩短至约10秒（理想情况）。

import threading
import requests
import time
# 模拟下载函数
def download_file(url):
    print(f"开始下载: {url}")
    # 模拟网络延迟
    time.sleep(2)
    print(f"下载完成: {url}")
# 准备下载任务
urls = [f"http://example.com/file_{i}" for i in range(10)]
# 多线程方式
threads = []
start_time = time.time()
for url in urls:
    t = threading.Thread(target=download_file, args=(url,))
    t.start()
    threads.append(t)
# 等待所有线程完成
for t in threads:
    t.join()
print(f"多线程总耗时: {time.time() - start_time:.2f}秒")

对比测试： 如果使用单线程循环，耗时将是20秒（10个任务×2秒），多线程将时间压缩到约2秒（理论上，取决于网络I/O的并行能力）。

实战案例二：多线程爬虫中的生产者-消费者模式

当爬取数据需要同时处理请求和解析时，可使用队列（queue.Queue） 来协调生产者和消费者线程。

import threading
import queue
import time
import random
# 模拟数据生成（生产者）
def producer(q, count=10):
    for i in range(count):
        data = f"item_{i}"
        q.put(data)
        print(f"[生产者] 放入: {data}")
        time.sleep(random.uniform(0.5, 1.5))
    q.put(None)  # 结束信号
# 模拟数据处理（消费者）
def consumer(q, thread_id):
    while True:
        data = q.get()
        if data is None:
            q.put(None)  # 传递结束信号给其他消费者
            break
        print(f"[消费者-{thread_id}] 处理: {data}")
        time.sleep(random.uniform(1, 2))
    print(f"[消费者-{thread_id}] 退出")
# 队列和线程创建
q = queue.Queue()
producer_thread = threading.Thread(target=producer, args=(q, 10))
consumer_threads = [threading.Thread(target=consumer, args=(q, i)) for i in range(3)]
# 启动所有线程
producer_thread.start()
for t in consumer_threads:
    t.start()
# 等待生产者结束
producer_thread.join()
for t in consumer_threads:
    t.join()
print("所有线程执行完毕")

核心要点：生产者和消费者通过Queue自动同步，None作为终止信号确保所有消费者正确退出，此模式广泛应用于爬虫、日志处理和数据管道场景。

常见陷阱与最佳实践

陷阱	解决方案
线程数过多导致上下文切换开销	使用线程池 (`concurrent.futures.ThreadPoolExecutor`)
共享资源访问冲突（数据竞争）	使用`threading.Lock`、`RLock`或`Semaphore`
死锁	避免嵌套锁，使用`with`语句自动释放
GIL对CPU密集型的限制	CPU密集型任务改用`multiprocessing`

推荐工具： Python 3.2+ 引入的 concurrent.futures 模块提供更简洁的线程池接口：

from concurrent.futures import ThreadPoolExecutor, as_completed
def task(n):
    time.sleep(1)
    return f"Task {n} done"
with ThreadPoolExecutor(max_workers=5) as executor:
    futures = [executor.submit(task, i) for i in range(10)]
    for future in as_completed(futures):
        print(future.result())

线程安全与锁机制详解

当多个线程同时修改同一变量时,可能产生不可预测的结果。

counter = 0
def increment():
    global counter
    for _ in range(100000):
        counter += 1  # 非原子操作
threads = [threading.Thread(target=increment) for _ in range(10)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(f"期望值: 1000000, 实际值: {counter}")  # 往往小于期望值

使用锁修复：

lock = threading.Lock()
counter = 0
def safe_increment():
    global counter
    for _ in range(100000):
        with lock:  # 自动获取和释放锁
            counter += 1
# 其余代码相同
print(f"期望值: 1000000, 实际值: {counter}")  # 正确

锁的类型：