Python 并发编程完全指南：threading / multiprocessing / concurrent.futures 深度解析与实战

Python 并发编程是提升程序性能的关键技能。由于 GIL 的存在，Python 的并发编程与其他语言有显著差异。理解 threading、multiprocessing 和 concurrent.futures 的适用场景，是写出高效 Python 并发程序的基础。

本文全面讲解 Python 并发编程：

GIL 原理与影响
threading 多线程编程
multiprocessing 多进程编程
concurrent.futures 线程池与进程池
同步原语（Lock / RLock / Semaphore / Event / Condition）
Queue 线程安全队列
进程间通信（Pipe / Queue / SharedMemory）
CPU 密集型 vs IO 密集型策略选择
实战场景与最佳实践

一、GIL 原理与影响#

1.1 什么是 GIL#

1
# GIL（Global Interpreter Lock）是 CPython 中的一个互斥锁
2
# 确保同一时刻只有一个线程执行 Python 字节码
3

4
# 验证 GIL 的存在
5
import threading
6
import time
7

8
def count_down(n):
9
    while n > 0:
10
        n -= 1
11

12
# 单线程
13
start = time.time()
14
count_down(100_000_000)
15
single_time = time.time() - start
16
print(f"单线程: {single_time:.2f}s")
17

18
# 多线程（受 GIL 限制，不会加速）
19
t1 = threading.Thread(target=count_down, args=(50_000_000,))
20
t2 = threading.Thread(target=count_down, args=(50_000_000,))
21

22
start = time.time()
23
t1.start()
24
t2.start()
25
t1.join()
26
t2.join()
27
multi_time = time.time() - start
28
print(f"多线程: {multi_time:.2f}s")
29

30
# 结果：多线程可能更慢（线程切换开销）
31
# 单线程: 3.21s
32
# 多线程: 3.45s

1.2 GIL 的影响范围#

1
# GIL 在以下场景中不影响性能：
2

3
# 1. IO 操作（自动释放 GIL）
4
import urllib.request
5

6
def fetch_url(url):
7
    response = urllib.request.urlopen(url)
8
    return response.read()
9

10
# 2. time.sleep（释放 GIL）
11
import time
12
time.sleep(1)  # 其他线程可以执行
13

14
# 3. C 扩展（可手动释放 GIL）
15
import numpy as np
16
# NumPy 底层在计算时释放 GIL
17

18
# GIL 在以下场景中严重影响性能：
19
# 1. CPU 密集型 Python 代码
20
# 2. 纯 Python 循环计算
21
# 3. 大量 Python 对象操作

1.3 应对策略#

1
# 策略选择决策树
2
"""
3
任务类型？
4
├── IO 密集型（网络/文件/数据库）
5
│   ├── 低并发 → threading
6
│   ├── 高并发 → asyncio
7
│   └── 简单并行 → ThreadPoolExecutor
8
├── CPU 密集型（计算/加密/图像处理）
9
│   ├── 简单并行 → ProcessPoolExecutor
10
│   ├── 需要精细控制 → multiprocessing
11
│   └── 大规模数据 → multiprocessing + SharedMemory
12
└── 混合型
13
    └── ThreadPoolExecutor + ProcessPoolExecutor 组合
14
"""

二、threading 多线程编程#

2.1 创建线程#

1
import threading
2
import time
3

4
# 方法1：函数方式
5
def worker(name, delay):
6
    print(f"线程 {name} 开始")
7
    time.sleep(delay)
8
    print(f"线程 {name} 完成")
9

10
t1 = threading.Thread(target=worker, args=("A", 2))
11
t2 = threading.Thread(target=worker, args=("B", 3))
12

13
t1.start()  # 启动线程
14
t2.start()
15

16
t1.join()   # 等待线程完成
17
t2.join()
18
print("所有线程完成")
19

20
# 方法2：继承 Thread 类
21
class MyThread(threading.Thread):
22
    def __init__(self, name, delay):
23
        super().__init__()
24
        self.name = name
25
        self.delay = delay
26

27
    def run(self):
28
        print(f"线程 {self.name} 开始")
29
        time.sleep(self.delay)
30
        print(f"线程 {self.name} 完成")
31

32
t = MyThread("C", 2)
33
t.start()
34
t.join()

2.2 线程状态与生命周期#

1
import threading
2

3
def task():
4
    import time
5
    time.sleep(2)
6

7
t = threading.Thread(target=task)
8

9
# 线程状态
10
print(t.is_alive())  # False（未启动）
11
print(t.daemon)      # False（非守护线程）
12

13
t.start()
14
print(t.is_alive())  # True（运行中）
15
print(t.name)        # Thread-1
16
print(t.ident)       # 线程ID
17

18
t.join()  # 等待完成
19
print(t.is_alive())  # False
20

21
# 守护线程（主线程退出时自动结束）
22
daemon_t = threading.Thread(target=task, daemon=True)
23
# 或
24
daemon_t = threading.Thread(target=task)
25
daemon_t.daemon = True
26
daemon_t.start()
27
# 主线程退出时，daemon_t 会被强制终止

2.3 获取线程返回值#

1
import threading
2

3
# 方法1：共享变量
4
def worker(result, key):
5
    result[key] = "computed value"
6

7
result = {}
8
t = threading.Thread(target=worker, args=(result, "data"))
9
t.start()
10
t.join()
11
print(result["data"])  # "computed value"
12

13
# 方法2：继承 Thread，重写 run
14
class ResultThread(threading.Thread):
15
    def __init__(self, *args, **kwargs):
16
        super().__init__(*args, **kwargs)
17
        self.result = None
18

19
    def run(self):
20
        if self._target:
21
            self.result = self._target(*self._args, **self._kwargs)
22

23
def compute(x, y):
24
    return x + y
25

26
t = ResultThread(target=compute, args=(10, 20))
27
t.start()
28
t.join()
29
print(t.result)  # 30

三、同步原语#

3.1 Lock（互斥锁）#

1
import threading
2

3
# 问题：线程不安全
4
counter = 0
5

6
def unsafe_increment():
7
    global counter
8
    for _ in range(1_000_000):
9
        counter += 1  # 非原子操作
10

11
t1 = threading.Thread(target=unsafe_increment)
12
t2 = threading.Thread(target=unsafe_increment)
13
t1.start()
14
t2.start()
15
t1.join()
16
t2.join()
17
print(f"预期: 2000000, 实际: {counter}")  # 实际 < 2000000
18

19
# 解决：使用 Lock
20
counter = 0
21
lock = threading.Lock()
22

23
def safe_increment():
24
    global counter
25
    for _ in range(1_000_000):
26
        with lock:  # 自动获取和释放锁
27
            counter += 1
28

29
t1 = threading.Thread(target=safe_increment)
30
t2 = threading.Thread(target=safe_increment)
31
t1.start()
32
t2.start()
33
t1.join()
34
t2.join()
35
print(f"预期: 2000000, 实际: {counter}")  # 2000000 ✅

3.2 RLock（可重入锁）#

1
import threading
2

3
# Lock 的问题：同一线程不能多次获取
4
lock = threading.Lock()
5
lock.acquire()
6
# lock.acquire()  # 死锁！
7

8
# RLock 解决：同一线程可多次获取
9
rlock = threading.RLock()
10
rlock.acquire()
11
rlock.acquire()  # ✅ 同一线程可以多次获取
12
rlock.release()
13
rlock.release()
14

15
# 实际应用：递归调用
16
class BankAccount:
17
    def __init__(self, balance=0):
18
        self.balance = balance
19
        self.lock = threading.RLock()
20

21
    def deposit(self, amount):
22
        with self.lock:
23
            self.balance += amount
24

25
    def withdraw(self, amount):
26
        with self.lock:
27
            if self.balance >= amount:
28
                self.balance -= amount
29
                return True
30
            return False
31

32
    def transfer(self, other, amount):
33
        with self.lock:  # 外层加锁
34
            if self.withdraw(amount):  # 内层也需要锁 → RLock
35
                other.deposit(amount)
36
                return True
37
            return False

3.3 Semaphore（信号量）#

1
import threading
2
import time
3
import random
4

5
# 限制同时访问的线程数（如数据库连接池）
6
semaphore = threading.Semaphore(3)  # 最多 3 个线程同时执行
7

8
def worker(worker_id):
9
    print(f"Worker {worker_id} 等待获取信号量")
10
    with semaphore:
11
        print(f"Worker {worker_id} 开始工作")
12
        time.sleep(random.uniform(0.5, 2.0))
13
        print(f"Worker {worker_id} 完成")
14

15
threads = [threading.Thread(target=worker, args=(i,)) for i in range(10)]
16
for t in threads:
17
    t.start()
18
for t in threads:
19
    t.join()
20

21
# 同时只有 3 个 Worker 在工作

3.4 Event（事件）#

1
import threading
2
import time
3

4
# Event：线程间简单通信
5
event = threading.Event()
6

7
def waiter(name):
8
    print(f"{name} 等待事件...")
9
    event.wait()  # 阻塞直到 event.set()
10
    print(f"{name} 收到事件，继续执行")
11

12
def setter():
13
    time.sleep(2)
14
    print("设置事件")
15
    event.set()
16

17
# 启动等待者
18
for name in ["A", "B", "C"]:
19
    threading.Thread(target=waiter, args=(name,)).start()
20

21
# 启动设置者
22
threading.Thread(target=setter).start()
23

24
# 2 秒后所有等待者同时继续执行

3.5 Condition（条件变量）#

1
import threading
2
import time
3
import random
4

5
# 生产者-消费者模式
6
class BoundedBuffer:
7
    def __init__(self, capacity):
8
        self.buffer = []
9
        self.capacity = capacity
10
        self.condition = threading.Condition()
11

12
    def put(self, item):
13
        with self.condition:
14
            while len(self.buffer) >= self.capacity:
15
                self.condition.wait()  # 缓冲区满，等待
16
            self.buffer.append(item)
17
            self.condition.notify()  # 通知消费者
18

19
    def get(self):
20
        with self.condition:
21
            while not self.buffer:
22
                self.condition.wait()  # 缓冲区空，等待
23
            item = self.buffer.pop(0)
24
            self.condition.notify()  # 通知生产者
25
            return item
26

27
buffer = BoundedBuffer(5)
28

29
def producer(id):
30
    for i in range(10):
31
        item = f"产品-{id}-{i}"
32
        buffer.put(item)
33
        print(f"生产: {item}")
34
        time.sleep(random.uniform(0.1, 0.5))
35

36
def consumer(id):
37
    for _ in range(10):
38
        item = buffer.get()
39
        print(f"  消费({id}): {item}")
40
        time.sleep(random.uniform(0.2, 0.8))
41

42
# 启动生产者和消费者
43
for i in range(2):
44
    threading.Thread(target=producer, args=(i,)).start()
45
for i in range(3):
46
    threading.Thread(target=consumer, args=(i,)).start()

四、Queue 线程安全队列#

4.1 基本队列#

1
import queue
2
import threading
3
import time
4

5
# Queue：线程安全的 FIFO 队列
6
q = queue.Queue(maxsize=10)
7

8
def producer():
9
    for i in range(20):
10
        q.put(f"任务-{i}")
11
        print(f"生产: 任务-{i}")
12
        time.sleep(0.1)
13
    q.put(None)  # 结束信号
14

15
def consumer(id):
16
    while True:
17
        item = q.get()  # 阻塞获取
18
        if item is None:
19
            q.put(None)  # 传递结束信号给其他消费者
20
            break
21
        print(f"  消费者-{id}: 处理 {item}")
22
        time.sleep(0.3)
23
        q.task_done()  # 标记任务完成
24

25
# 启动
26
threading.Thread(target=producer).start()
27
for i in range(3):
28
    threading.Thread(target=consumer, args=(i,)).start()

4.2 队列类型#

1
import queue
2

3
# FIFO 队列（先进先出）
4
fifo = queue.Queue()
5
fifo.put(1)
6
fifo.put(2)
7
print(fifo.get())  # 1
8

9
# LIFO 队列（后进先出，栈）
10
lifo = queue.LifoQueue()
11
lifo.put(1)
12
lifo.put(2)
13
print(lifo.get())  # 2
14

15
# 优先级队列
16
pq = queue.PriorityQueue()
17
pq.put((3, "低优先级"))
18
pq.put((1, "高优先级"))
19
pq.put((2, "中优先级"))
20
print(pq.get()[1])  # "高优先级"
21
print(pq.get()[1])  # "中优先级"
22
print(pq.get()[1])  # "低优先级"

4.3 线程池工作队列模式#

1
import queue
2
import threading
3
import time
4
import requests
5

6
class ThreadPool:
7
    """简易线程池"""
8

9
    def __init__(self, num_threads):
10
        self.tasks = queue.Queue()
11
        self.threads = []
12
        self.running = True
13

14
        for _ in range(num_threads):
15
            t = threading.Thread(target=self._worker)
16
            t.daemon = True
17
            t.start()
18
            self.threads.append(t)
19

20
    def _worker(self):
21
        while self.running:
22
            try:
23
                func, args, callback = self.tasks.get(timeout=1)
24
                result = func(*args)
25
                if callback:
26
                    callback(result)
27
                self.tasks.task_done()
28
            except queue.Empty:
29
                continue
30

31
    def submit(self, func, *args, callback=None):
32
        self.tasks.put((func, args, callback))
33

34
    def shutdown(self):
35
        self.running = False
36
        for t in self.threads:
37
            t.join()
38

39
# 使用
40
pool = ThreadPool(4)
41

42
def fetch(url):
43
    response = requests.get(url)
44
    return len(response.text)
45

46
def on_complete(result):
47
    print(f"获取到 {result} 字节")
48

49
urls = ["https://httpbin.org/get"] * 5
50
for url in urls:
51
    pool.submit(fetch, url, callback=on_complete)
52

53
time.sleep(5)
54
pool.shutdown()

五、multiprocessing 多进程#

5.1 创建进程#

1
import multiprocessing
2
import time
3

4
def worker(name, delay):
5
    print(f"进程 {name} 开始")
6
    time.sleep(delay)
7
    print(f"进程 {name} 完成")
8

9
if __name__ == '__main__':
10
    p1 = multiprocessing.Process(target=worker, args=("A", 2))
11
    p2 = multiprocessing.Process(target=worker, args=("B", 3))
12

13
    p1.start()
14
    p2.start()
15

16
    p1.join()
17
    p2.join()
18
    print("所有进程完成")

5.2 CPU 密集型任务对比#

1
import threading
2
import multiprocessing
3
import time
4
import math
5

6
def cpu_intensive(n):
7
    """CPU 密集型任务：计算质数"""
8
    count = 0
9
    for i in range(2, n):
10
        is_prime = True
11
        for j in range(2, int(math.sqrt(i)) + 1):
12
            if i % j == 0:
13
                is_prime = False
14
                break
15
        if is_prime:
16
            count += 1
17
    return count
18

19
# 单线程
20
start = time.time()
21
result1 = cpu_intensive(500_000)
22
result2 = cpu_intensive(500_000)
23
single_time = time.time() - start
24
print(f"单线程: {single_time:.2f}s")
25

26
# 多线程（受 GIL 限制，无加速）
27
start = time.time()
28
t1 = threading.Thread(target=cpu_intensive, args=(500_000,))
29
t2 = threading.Thread(target=cpu_intensive, args=(500_000,))
30
t1.start()
31
t2.start()
32
t1.join()
33
t2.join()
34
thread_time = time.time() - start
35
print(f"多线程: {thread_time:.2f}s")
36

37
# 多进程（真正并行，接近 2 倍加速）
38
if __name__ == '__main__':
39
    start = time.time()
40
    p1 = multiprocessing.Process(target=cpu_intensive, args=(500_000,))
41
    p2 = multiprocessing.Process(target=cpu_intensive, args=(500_000,))
42
    p1.start()
43
    p2.start()
44
    p1.join()
45
    p2.join()
46
    process_time = time.time() - start
47
    print(f"多进程: {process_time:.2f}s")
48

49
    # 结果示例：
50
    # 单线程: 8.52s
51
    # 多线程: 8.71s（无加速，甚至更慢）
52
    # 多进程: 4.38s（接近 2 倍加速）

5.3 获取进程返回值#

1
import multiprocessing
2

3
def compute(x, y):
4
    return x * y
5

6
if __name__ == '__main__':
7
    # 方法1：使用 Queue
8
    result_queue = multiprocessing.Queue()
9

10
    def worker_with_queue(x, y, q):
11
        q.put(compute(x, y))
12

13
    p = multiprocessing.Process(
14
        target=worker_with_queue,
15
        args=(10, 20, result_queue)
16
    )
17
    p.start()
18
    p.join()
19
    print(result_queue.get())  # 200
20

21
    # 方法2：使用 Pool（推荐）
22
    with multiprocessing.Pool() as pool:
23
        result = pool.apply(compute, (10, 20))
24
        print(result)  # 200

六、进程间通信#

6.1 Pipe（管道）#

1
import multiprocessing
2

3
def sender(conn):
4
    conn.send("Hello from sender")
5
    data = conn.recv()
6
    print(f"Sender received: {data}")
7
    conn.close()
8

9
if __name__ == '__main__':
10
    parent_conn, child_conn = multiprocessing.Pipe()
11

12
    p = multiprocessing.Process(target=sender, args=(child_conn,))
13
    p.start()
14

15
    print(parent_conn.recv())  # "Hello from sender"
16
    parent_conn.send("Hello from parent")
17

18
    p.join()

6.2 Queue（进程队列）#

1
import multiprocessing
2
import time
3

4
def producer(q):
5
    for i in range(5):
6
        q.put(f"产品-{i}")
7
        print(f"生产: 产品-{i}")
8
        time.sleep(0.5)
9
    q.put(None)  # 结束信号
10

11
def consumer(q):
12
    while True:
13
        item = q.get()
14
        if item is None:
15
            break
16
        print(f"消费: {item}")
17
        time.sleep(1)
18

19
if __name__ == '__main__':
20
    q = multiprocessing.Queue(maxsize=10)
21

22
    p1 = multiprocessing.Process(target=producer, args=(q,))
23
    p2 = multiprocessing.Process(target=consumer, args=(q,))
24

25
    p1.start()
26
    p2.start()
27

28
    p1.join()
29
    p2.join()

6.3 Value 和 Array（共享内存）#

1
import multiprocessing
2

3
def increment_counter(counter, lock):
4
    for _ in range(100_000):
5
        with lock:
6
            counter.value += 1
7

8
if __name__ == '__main__':
9
    # 共享整数
10
    counter = multiprocessing.Value('i', 0)
11
    lock = multiprocessing.Lock()
12

13
    processes = [
14
        multiprocessing.Process(target=increment_counter, args=(counter, lock))
15
        for _ in range(4)
16
    ]
17

18
    for p in processes:
19
        p.start()
20
    for p in processes:
21
        p.join()
22

23
    print(f"最终值: {counter.value}")  # 400000
24

25
# 共享数组
26
def modify_array(arr, lock):
27
    for i in range(len(arr)):
28
        with lock:
29
            arr[i] += 1
30

31
if __name__ == '__main__':
32
    arr = multiprocessing.Array('i', [0, 0, 0, 0, 0])
33
    lock = multiprocessing.Lock()
34

35
    processes = [
36
        multiprocessing.Process(target=modify_array, args=(arr, lock))
37
        for _ in range(3)
38
    ]
39

40
    for p in processes:
41
        p.start()
42
    for p in processes:
43
        p.join()
44

45
    print(f"数组: {list(arr)}")  # [3, 3, 3, 3, 3]

6.4 Manager（共享对象）#

1
import multiprocessing
2

3
def worker(d, key, value):
4
    d[key] = value
5

6
if __name__ == '__main__':
7
    with multiprocessing.Manager() as manager:
8
        # 共享字典
9
        shared_dict = manager.dict()
10

11
        processes = []
12
        for i in range(5):
13
            p = multiprocessing.Process(
14
                target=worker,
15
                args=(shared_dict, f"key_{i}", f"value_{i}")
16
            )
17
            processes.append(p)
18
            p.start()
19

20
        for p in processes:
21
            p.join()
22

23
        print(dict(shared_dict))
24
        # {'key_0': 'value_0', 'key_1': 'value_1', ...}
25

26
        # Manager 还支持 list, dict, Namespace, Lock 等
27
        shared_list = manager.list()
28
        shared_list.append(1)

七、concurrent.futures（推荐）#

7.1 ThreadPoolExecutor#

1
from concurrent.futures import ThreadPoolExecutor, as_completed
2
import requests
3

4
# 线程池：IO 密集型任务
5
def fetch_url(url):
6
    response = requests.get(url, timeout=10)
7
    return url, response.status_code, len(response.text)
8

9
urls = [
10
    "https://httpbin.org/get",
11
    "https://httpbin.org/delay/1",
12
    "https://httpbin.org/delay/2",
13
    "https://httpbin.org/ip",
14
    "https://httpbin.org/uuid",
15
]
16

17
# 方法1：submit + as_completed（灵活）
18
with ThreadPoolExecutor(max_workers=3) as executor:
19
    futures = {executor.submit(fetch_url, url): url for url in urls}
20

21
    for future in as_completed(futures):
22
        url = futures[future]
23
        try:
24
            result = future.result()
25
            print(f"{result[0]}: {result[1]} ({result[2]} bytes)")
26
        except Exception as e:
27
            print(f"{url} 错误: {e}")
28

29
# 方法2：map（简单，按顺序返回）
30
with ThreadPoolExecutor(max_workers=3) as executor:
31
    results = executor.map(fetch_url, urls)
32
    for result in results:
33
        print(f"{result[0]}: {result[1]} ({result[2]} bytes)")

7.2 ProcessPoolExecutor#

1
from concurrent.futures import ProcessPoolExecutor, as_completed
2
import math
3

4
# 进程池：CPU 密集型任务
5
def is_prime(n):
6
    if n < 2:
7
        return False
8
    for i in range(2, int(math.sqrt(n)) + 1):
9
        if n % i == 0:
10
            return False
11
    return True
12

13
def count_primes(start, end):
14
    count = 0
15
    for i in range(start, end):
16
        if is_prime(i):
17
            count += 1
18
    return count
19

20
if __name__ == '__main__':
21
    # 将任务分成 4 块
22
    ranges = [
23
        (2, 250_000),
24
        (250_000, 500_000),
25
        (500_000, 750_000),
26
        (750_000, 1_000_000),
27
    ]
28

29
    with ProcessPoolExecutor(max_workers=4) as executor:
30
        futures = {
31
            executor.submit(count_primes, start, end): (start, end)
32
            for start, end in ranges
33
        }
34

35
        total = 0
36
        for future in as_completed(futures):
37
            count = future.result()
38
            total += count
39
            print(f"区块 {futures[future]}: {count} 个质数")
40

41
        print(f"总计: {total} 个质数")

7.3 回调函数#

1
from concurrent.futures import ThreadPoolExecutor
2
import requests
3

4
def fetch(url):
5
    response = requests.get(url)
6
    return response.json()
7

8
def on_success(future):
9
    """任务完成回调"""
10
    try:
11
        data = future.result()
12
        print(f"成功: {data.get('url', 'unknown')}")
13
    except Exception as e:
14
        print(f"失败: {e}")
15

16
with ThreadPoolExecutor(max_workers=3) as executor:
17
    future = executor.submit(fetch, "https://httpbin.org/get")
18
    future.add_done_callback(on_success)
19

20
    # 主线程可以继续做其他事
21
    print("任务已提交")

7.4 统一接口切换#

1
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
2

3
def task(n):
4
    import time
5
    time.sleep(1)
6
    return n * n
7

8
# 只需改一行即可切换线程池/进程池
9
def run(executor_class, tasks):
10
    with executor_class(max_workers=4) as executor:
11
        results = list(executor.map(task, tasks))
12
        return results
13

14
tasks = list(range(10))
15

16
# IO 密集型 → 用线程池
17
results = run(ThreadPoolExecutor, tasks)
18
print(results)
19

20
# CPU 密集型 → 用进程池
21
if __name__ == '__main__':
22
    results = run(ProcessPoolExecutor, tasks)
23
    print(results)

八、实战场景#

8.1 并发网页爬虫#

1
from concurrent.futures import ThreadPoolExecutor, as_completed
2
import requests
3
from bs4 import BeautifulSoup
4
import time
5

6
def crawl_page(url):
7
    try:
8
        response = requests.get(url, timeout=10)
9
        soup = BeautifulSoup(response.text, 'html.parser')
10
        title = soup.find('title')
11
        return {
12
            'url': url,
13
            'status': response.status_code,
14
            'title': title.text if title else 'N/A',
15
            'length': len(response.text),
16
        }
17
    except Exception as e:
18
        return {'url': url, 'error': str(e)}
19

20
def batch_crawl(urls, max_workers=10):
21
    results = []
22

23
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
24
        futures = {executor.submit(crawl_page, url): url for url in urls}
25

26
        for future in as_completed(futures):
27
            result = future.result()
28
            results.append(result)
29
            if 'error' not in result:
30
                print(f"[{result['status']}] {result['title']} - {result['url']}")
31

32
    return results
33

34
# 使用
35
urls = [f"https://httpbin.org/delay/{i % 3}" for i in range(20)]
36

37
start = time.time()
38
results = batch_crawl(urls, max_workers=10)
39
print(f"\n爬取 {len(urls)} 个页面，耗时: {time.time() - start:.2f}s")

8.2 并行数据处理#

1
from concurrent.futures import ProcessPoolExecutor
2
import time
3

4
def process_chunk(data_chunk):
5
    """处理数据块（CPU 密集型）"""
6
    result = []
7
    for item in data_chunk:
8
        # 模拟计算密集型操作
9
        value = sum(i * i for i in range(item))
10
        result.append(value)
11
    return result
12

13
def parallel_process(data, num_workers=4):
14
    """并行处理大数据集"""
15
    # 分块
16
    chunk_size = len(data) // num_workers
17
    chunks = [
18
        data[i:i + chunk_size]
19
        for i in range(0, len(data), chunk_size)
20
    ]
21

22
    # 并行处理
23
    with ProcessPoolExecutor(max_workers=num_workers) as executor:
24
        results = list(executor.map(process_chunk, chunks))
25

26
    # 合并结果
27
    return [item for chunk in results for item in chunk]
28

29
if __name__ == '__main__':
30
    data = list(range(1000, 2000))
31

32
    start = time.time()
33
    result = parallel_process(data, num_workers=4)
34
    print(f"并行处理 {len(data)} 项，耗时: {time.time() - start:.2f}s")

8.3 速率限制并发#

1
import threading
2
import time
3
import requests
4
from concurrent.futures import ThreadPoolExecutor
5

6
class RateLimiter:
7
    """令牌桶速率限制器"""
8

9
    def __init__(self, rate, capacity=None):
10
        self.rate = rate  # 每秒生成的令牌数
11
        self.capacity = capacity or rate
12
        self.tokens = self.capacity
13
        self.last_update = time.time()
14
        self.lock = threading.Lock()
15

16
    def acquire(self):
17
        with self.lock:
18
            now = time.time()
19
            elapsed = now - self.last_update
20
            self.tokens = min(
21
                self.capacity,
22
                self.tokens + elapsed * self.rate
23
            )
24
            self.last_update = now
25

26
            if self.tokens >= 1:
27
                self.tokens -= 1
28
                return True
29
            return False
30

31
def rate_limited_fetch(url, rate_limiter):
32
    while not rate_limiter.acquire():
33
        time.sleep(0.01)
34

35
    response = requests.get(url)
36
    return url, response.status_code
37

38
# 每秒最多 5 个请求
39
rate_limiter = RateLimiter(rate=5)
40
urls = [f"https://httpbin.org/get?i={i}" for i in range(20)]
41

42
with ThreadPoolExecutor(max_workers=10) as executor:
43
    futures = [
44
        executor.submit(rate_limited_fetch, url, rate_limiter)
45
        for url in urls
46
    ]
47

48
    for future in futures:
49
        url, status = future.result()
50
        print(f"{status} - {url}")

九、常见陷阱与最佳实践#

❌ 陷阱 1：竞态条件#

1
import threading
2

3
# ❌ 线程不安全
4
balance = 0
5

6
def transfer(amount):
7
    global balance
8
    # 以下操作非原子性
9
    current = balance
10
    time.sleep(0.001)  # 模拟延迟
11
    balance = current + amount
12

13
# ✅ 使用锁保护共享资源
14
balance = 0
15
lock = threading.Lock()
16

17
def safe_transfer(amount):
18
    global balance
19
    with lock:
20
        current = balance
21
        balance = current + amount

❌ 陷阱 2：死锁#

1
import threading
2

3
# ❌ 死锁：锁的获取顺序不一致
4
lock1 = threading.Lock()
5
lock2 = threading.Lock()
6

7
def task_a():
8
    with lock1:
9
        time.sleep(0.1)
10
        with lock2:  # 等待 lock2
11
            print("A done")
12

13
def task_b():
14
    with lock2:
15
        time.sleep(0.1)
16
        with lock1:  # 等待 lock1 → 死锁
17
            print("B done")
18

19
# ✅ 修复：统一锁的获取顺序
20
def task_a_fixed():
21
    with lock1:
22
        with lock2:
23
            print("A done")
24

25
def task_b_fixed():
26
    with lock1:  # 先获取 lock1
27
        with lock2:  # 再获取 lock2
28
            print("B done")

❌ 陷阱 3：守护线程中的资源泄漏#

1
import threading
2

3
# ❌ 守护线程可能被强制终止，导致资源未释放
4
def write_file():
5
    with open("data.txt", "w") as f:
6
        while True:
7
            f.write("data\n")
8
            time.sleep(1)
9

10
# daemon=True 的线程在主线程退出时会被强制终止
11
# 可能导致文件未正确关闭
12

13
# ✅ 使用非守护线程 + 优雅退出
14
def write_file_safe(stop_event):
15
    with open("data.txt", "w") as f:
16
        while not stop_event.is_set():
17
            f.write("data\n")
18
            f.flush()
19
            time.sleep(1)
20

21
stop = threading.Event()
22
t = threading.Thread(target=write_file_safe, args=(stop,))
23
t.start()
24

25
# 优雅退出
26
stop.set()
27
t.join()

✅ 最佳实践#

原则	说明
IO 密集用线程	网络请求、文件读写
CPU 密集用进程	计算、加密、图像处理
优先用 concurrent.futures	统一 API，简化代码
避免共享状态	使用 Queue 传递数据
锁要最小化	只锁必要的代码段
统一锁顺序	避免死锁
设置超时	避免永久阻塞
使用上下文管理器	自动释放资源
进程池注意序列化	参数必须可 pickle

十、总结#

并发方案对比#

方案	适用场景	优势	劣势
threading	IO 密集型	轻量、共享内存	GIL 限制
multiprocessing	CPU 密集型	真正并行	内存开销大
concurrent.futures	通用	统一 API、简化代码	灵活性略低
asyncio	高并发 IO	单线程高并发	需要异步库支持

选型决策#

1
需要并发？
2
├── IO 密集型
3
│   ├── 简单场景 → ThreadPoolExecutor
4
│   ├── 高并发 → asyncio
5
│   └── 需要同步代码 → threading + Queue
6
├── CPU 密集型
7
│   ├── 简单并行 → ProcessPoolExecutor
8
│   ├── 大数据 → multiprocessing + SharedMemory
9
│   └── 数值计算 → NumPy（释放 GIL）
10
└── 混合型
11
    └── ThreadPoolExecutor + ProcessPoolExecutor

Python 并发编程的核心是理解 GIL 的影响并选择正确的工具。IO 密集型用线程，CPU 密集型用进程，简单场景用 concurrent.futures。掌握同步原语和进程间通信，才能写出安全高效的并发程序。2026 年，concurrent.futures 已成为 Python 并发编程的首选方案，推荐优先使用。