Python多线程编程实战指南

一、理论基础

1.1 线程与进程的区别

• 进程：拥有独立的内存空间、数据栈等，系统开销大，进程间通信复杂。

• 线程：依附于进程，共享进程的内存空间，创建和切换开销小，但需要自己处理共享数据的同步问题。

1.2 Python 的 GIL（全局解释器锁）

• CPython 解释器中有一个 GIL，同一时刻只允许一个线程执行 Python 字节码。

• 因此，Python 的多线程并不能利用多核 CPU 并行计算（计算密集型任务反而可能变慢）。

• 适用场景：I/O 密集型任务（网络请求、文件读写、用户输入等），线程在等待 I/O 时会释放 GIL，从而真正并发。

1.3 两个线程模块

• thread（Python 2 中的名字，Python 3 中为 _thread）：底层、原始，不推荐直接使用。

• threading：基于 _thread 封装，提供了更强大的 Thread 类和各种同步工具，推荐使用。

二、实操环境准备

• Python 3.6+（任何较新版本都可以）

• 任意代码编辑器（VS Code、PyCharm、甚至记事本）

• 命令行 / 终端

验证 Python 版本：

python –version # 应显示 3.x

三、实操步骤 1：创建并启动线程（两种方式）

3.1 方式一：函数式（使用 threading.Thread(target=…)）

新建文件 thread_func.py，写入以下代码：

import threading
import time

def print_time(thread_name, delay, counter):
"""线程函数：每隔 delay 秒打印一次当前时间，重复 counter 次"""
while counter > 0:
time.sleep(delay)
print(f"{thread_name}: {time.ctime()}")
counter -= 1

# 创建两个线程
t1 = threading.Thread(target=print_time, args=("Thread-1", 2, 5))
t2 = threading.Thread(target=print_time, args=("Thread-2", 4, 5))

# 启动线程
t1.start()
t2.start()

# 等待两个线程都结束（否则主线程可能提前退出）
t1.join()
t2.join()

print("所有线程执行完毕，主线程退出。")

运行：

python thread_func.py

预期输出（时间会变化，但顺序类似）：

Thread-1: Thu Jun 11 15:42:17 2026
Thread-1: Thu Jun 11 15:42:19 2026
Thread-2: Thu Jun 11 15:42:19 2026
Thread-1: Thu Jun 11 15:42:21 2026
Thread-2: Thu Jun 11 15:42:23 2026
…

说明：

• threading.Thread(target=函数名, args=元组) 创建线程对象。

• start() 使线程开始运行（调用 run()，默认 run() 会执行 target 指定的函数）。

• join() 阻塞主线程，直到被调用的线程结束，避免主线程提前退出。

3.2 方式二：继承 threading.Thread 类

新建文件 thread_class.py：

import threading
import time

class MyThread(threading.Thread):
def __init__(self, thread_id, name, delay, counter):
super().__init__() # 调用父类初始化
self.thread_id = thread_id
self.name = name
self.delay = delay
self.counter = counter

def run(self):
"""线程启动后自动执行的方法"""
print(f"开始线程：{self.name}")
while self.counter > 0:
time.sleep(self.delay)
print(f"{self.name}: {time.ctime()}")
self.counter -= 1
print(f"退出线程：{self.name}")

# 创建线程实例
thread1 = MyThread(1, "Thread-1", 1, 5)
thread2 = MyThread(2, "Thread-2", 2, 5)

thread1.start()
thread2.start()

thread1.join()
thread2.join()

print("主线程结束")

运行：

python thread_class.py

效果与函数式类似，但更符合面向对象风格，适合需要保存更多状态或复用的场景。

四、实操步骤 2：线程同步（使用锁）

当多个线程修改同一份数据时，会发生“竞态条件”。下面用两个线程同时对一个全局变量加 1000000 次来演示问题，然后用锁解决。

4.1 无锁的错误示例

新建 race_condition.py：

import threading

# 共享资源
counter = 0

def increment(thread_name, times):
global counter
for _ in range(times):
counter += 1 # 这不是原子操作！可能被线程切换中断

# 创建两个线程，每个加 1000000 次
t1 = threading.Thread(target=increment, args=("A", 1000000))
t2 = threading.Thread(target=increment, args=("B", 1000000))

t1.start()
t2.start()
t1.join()
t2.join()

print(f"期望结果: 2000000")
print(f"实际结果: {counter}")

运行几次，你会发现结果往往小于 2000000（如 1845214 等），这就是因为线程交替执行时丢失了更新。

4.2 使用锁（threading.Lock）修复

新建 with_lock.py：

import threading

counter = 0
lock = threading.Lock() # 创建一个锁对象

def increment_safe(thread_name, times):
global counter
for _ in range(times):
lock.acquire() # 获取锁（如果已被其他线程持有，则阻塞）
# 以下临界区代码同一时刻只有一个线程能执行
counter += 1
lock.release() # 释放锁

# 也可以使用 with 语句（更推荐）：
def increment_safe_with(thread_name, times):
global counter
for _ in range(times):
with lock: # 自动 acquire 和 release
counter += 1

t1 = threading.Thread(target=increment_safe, args=("A", 1000000))
t2 = threading.Thread(target=increment_safe, args=("B", 1000000))

t1.start()
t2.start()
t1.join()
t2.join()

print(f"加锁后的结果: {counter}") # 稳定输出 2000000

说明：

• lock.acquire() 和 lock.release() 之间的代码称为“临界区”，一次只允许一个线程进入。

• 使用 with lock: 更加简洁安全，避免忘记释放锁。

五、实操步骤 3：线程间通信 – 队列（Queue）

queue.Queue 是线程安全的 FIFO 队列，非常适合生产者-消费者模式。

示例：三个工人线程处理一个任务队列

新建 queue_demo.py：

import threading
import queue
import time
import random

# 任务队列，最大长度 5
task_queue = queue.Queue(maxsize=5)

def producer(name, total_tasks):
"""生产者：往队列中放任务"""
for i in range(total_tasks):
task = f"任务-{i}"
task_queue.put(task) # 如果队列满，会阻塞直到有空间
print(f"{name} 生产了 {task}")
time.sleep(random.uniform(0.2, 0.5))
# 生产结束后发送“毒丸”信号，通知消费者退出（数量等于消费者个数）
for _ in range(3):
task_queue.put(None)

def consumer(name):
"""消费者：从队列中取任务并处理"""
while True:
task = task_queue.get() # 如果队列空，会阻塞直到有新任务
if task is None: # 收到毒丸，退出
break
print(f"{name} 开始处理 {task}")
time.sleep(random.uniform(0.3, 0.7)) # 模拟处理耗时
print(f"{name} 完成 {task}")
task_queue.task_done() # 告诉队列该任务已完成

# 创建线程
prod = threading.Thread(target=producer, args=("生产者", 10))
cons1 = threading.Thread(target=consumer, args=("消费者-1",))
cons2 = threading.Thread(target=consumer, args=("消费者-2",))
cons3 = threading.Thread(target=consumer, args=("消费者-3",))

# 启动消费者（先启动，会阻塞等待任务）
cons1.start()
cons2.start()
cons3.start()

# 启动生产者
prod.start()

# 等待生产者结束
prod.join()

# 等待所有任务被处理（队列为空且 task_done 调用次数匹配）
task_queue.join()

print("所有任务处理完毕")

运行：

bash

python queue_demo.py

你会看到三个消费者并发地从队列中取出任务处理，整个过程自动同步，无需手动加锁。

关键方法：

• put(item)：放入元素（队列满时阻塞）

• get()：取出元素（队列空时阻塞）

• task_done()：表示之前取出的任务已完成

• join()：阻塞直到队列中所有元素都被 task_done() 确认

六、实操步骤 4：使用 threading 的其他常用功能

6.1 获取当前线程信息

import threading

def show_info():
t = threading.current_thread()
print(f"当前线程: {t.name}, ID: {t.ident}, 是否存活: {t.is_alive()}")

t = threading.Thread(target=show_info, name="我的线程")
t.start()
t.join()

运行结果为：

6.2 线程枚举

print(f"活动线程数: {threading.active_count()}")
for t in threading.enumerate():
print(t.name)

6.3 守护线程（Daemon Thread）

主线程结束时，守护线程会自动被终止。适合后台监控任务。

d = threading.Thread(target=some_task, daemon=True)
d.start()
# 主线程结束后，d 会立即被强制结束

七、注意事项与最佳实践

八、扩展学习：线程池（ThreadPoolExecutor）

对于大量短期任务，频繁创建销毁线程开销大，线程池可以复用线程。

from concurrent.futures import ThreadPoolExecutor
import time

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

with ThreadPoolExecutor(max_workers=3) as executor:
results = executor.map(task, [1, 2, 3, 4, 5])

print(list(results)) # [1, 4, 9, 16, 25]

map 方法自动将迭代器的元素分配给线程池中的线程执行，非常方便