小菜鸟

java菜鸟号正在起航

0%

多进程

发表于 更新于 分类于 阅读次数： Valine：
本文字数： 4.1k 阅读时长 ≈ 4 分钟

Python 多进程：8 核 CPU 只有 1 个在跑？那是 GIL 在”捣乱”

写了一个计算密集型的程序，跑起来 CPU 占用 100%，但仔细一看——8 个核心只有 1 个在满负荷运转，剩下 7 个几乎闲置。

这不是程序写得不对，是 Python 的 GIL（全局解释器锁） 在”从中作梗”。GIL 规定：同一时刻，只有一个线程能执行 Python 代码。所以多线程在计算任务上，形同虚设。

多进程是绕过 GIL 的方案——每个进程有自己独立的 Python 解释器，各自跑各自的，操作系统把它们调度到不同的 CPU 核心上，真正实现了并行。

什么场景用多进程？ CPU 密集型——大量计算、加密解密、图像处理、数据压缩。这种任务”算”的时间多，”等”的时间少，多线程没用，多进程才是答案。

先看证据：多线程 vs 多进程，差距一目了然

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
import multiprocessing
import threading
import time
from concurrent.futures import ThreadPoolExecutor

def cpu_work(n):
    """纯计算，不干别的"""
    count = 0
    for i in range(2, n):
        is_prime = all(i % j != 0 for j in range(2, int(i**0.5) + 1))
        if is_prime:
            count += 1
    return count

# 单线程
start = time.time()
cpu_work(100000)
print(f"单线程: {time.time() - start:.2f}s")

# 多线程（4个）
start = time.time()
with ThreadPoolExecutor(4) as ex:
    [cpu_work(100000) for _ in range(4)]
print(f"多线程: {time.time() - start:.2f}s")   # 比单线程还慢！

# 多进程（4个）
start = time.time()
with multiprocessing.Pool(4) as pool:
    pool.map(cpu_work, [100000] * 4)
print(f"多进程: {time.time() - start:.2f}s")   # 约单线程的 1/4

典型结果：

1
2
3
单线程: 2.35s
多线程: 2.78s   ← 不仅没快，还慢了
多进程: 0.68s   ← 接近 4 倍提速

多线程在计算任务上不仅无效，线程切换还带来额外开销。多进程才是真正的并行。

多进程池：别反复创建进程，用池复用

每次创建进程都要分配内存、启动解释器，开销不小。实际开发用进程池——预先创建一批进程，任务来了分配，干完了回收。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
import multiprocessing

def worker(n):
    return n ** 2

if __name__ == "__main__":
    # 默认进程数 = CPU 核心数
    with multiprocessing.Pool() as pool:
        # map：批量提交，保持顺序
        results = pool.map(worker, range(100))
        
        # apply_async：异步提交单个任务
        async_result = pool.apply_async(worker, (10,))
        result = async_result.get()  # 获取结果（会阻塞）

进程数设多少？

任务类型 建议进程数
CPU 密集型 CPU 核心数
I/O 密集型（但不想用异步） CPU 核心数 × 2
内存密集型 CPU 核心数 ÷ 2

设多了没用，操作系统频繁切换进程反而拖慢速度。

进程间通信：进程不共享内存，怎么传数据？

多进程的代价是：进程之间不共享内存。传递数据需要拷贝，有开销。

Queue：最常用，生产者-消费者模式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
import multiprocessing

def producer(q, items):
    for item in items:
        q.put(item)
    q.put(None)   # 结束信号

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

if __name__ == "__main__":
    q = multiprocessing.Queue()
    p1 = multiprocessing.Process(target=producer, args=(q, [1,2,3,4,5]))
    p2 = multiprocessing.Process(target=consumer, args=(q,))
    p1.start()
    p2.start()
    p1.join()
    p2.join()

Pipe：两个进程之间点对点通信

1
2
parent_conn, child_conn = multiprocessing.Pipe()
# 一个用 parent_conn 发，另一个用 child_conn 收

共享内存：高频读写，但得加锁

1
2
3
4
5
6
7
counter = multiprocessing.Value('i', 0)   # 'i' 表示整数
lock = multiprocessing.Lock()

def increment():
    for _ in range(1000):
        with lock:
            counter.value += 1

不加锁的话，多个进程同时改同一个值会乱。with lock 保证同一时刻只有一个进程在改。

通信方式怎么选？

场景 推荐
多对多，任务分发 Queue
两个进程双向通信 Pipe
高频读写简单数值 Value/Array + Lock
复杂数据结构（省事但慢） Manager

Manager 是最方便的，不用操心锁的问题，但性能最差。能用 Queue 和 Value 就不用 Manager。

Windows 上必写的保护：if __name__ == "__main__":

Windows 创建新进程时会重新导入主模块，如果不加保护，程序会无限递归创建进程直到崩溃。

1
2
3
4
5
6
7
8
#  Windows 上会出问题
p = multiprocessing.Process(target=worker)
p.start()

# 必须加
if __name__ == "__main__":
    p = multiprocessing.Process(target=worker)
    p.start()

Linux 和 Mac 上没有这个限制，但为了跨平台兼容，永远加上。

三个容易踩的坑

坑1：传递大对象，序列化开销巨大

进程间传递数据需要 pickle 序列化。一个 500MB 的 DataFrame，传一次可能要好几秒。

解决方案：

  • 用共享内存（multiprocessing.shared_memory）
  • 或者把数据切分好再传给每个进程，让每个进程自己读数据（比如从磁盘读）

坑2：日志混乱

多个进程同时写一个日志文件，内容会穿插，格式乱掉。

解决方案： 所有进程把日志发到一个队列里，用一个专门的进程负责写文件。

1
2
3
4
5
import logging.handlers

log_queue = multiprocessing.Queue()
handler = logging.handlers.QueueHandler(log_queue)
# 另一个进程从队列取日志，写入文件

坑3：死锁

生产者往队列放数据，队列满了阻塞；主进程又在等生产者结束，形成死锁。

解决方案： 用不限大小的队列，或者确保消费者及时取数据。

实战：批量处理 1 万张图片

一个真实场景——处理 1 万张图片，每张缩放、加水印、调色。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
import multiprocessing
from PIL import Image
import os

def process_one(args):
    input_path, output_path, size = args
    try:
        with Image.open(input_path) as img:
            img.thumbnail(size)
            img.save(output_path, optimize=True, quality=85)
        return True, input_path
    except Exception as e:
        return False, f"{input_path}: {e}"

def batch_process(input_dir, output_dir):
    tasks = []
    for f in os.listdir(input_dir):
        if f.lower().endswith(('.png', '.jpg', '.jpeg')):
            tasks.append((
                os.path.join(input_dir, f),
                os.path.join(output_dir, f),
                (800, 800)
            ))
    
    # 8 个进程并行处理
    with multiprocessing.Pool(8) as pool:
        results = pool.map(process_one, tasks)
    
    success = sum(1 for ok, _ in results if ok)
    print(f"成功: {success}/{len(tasks)}")

if __name__ == "__main__":
    batch_process("./raw", "./processed")

设计思路：

  • 每张图片独立处理 → 天然可并行
  • 进程数设为 8（假设机器 8 核）
  • 单张失败不影响整体
  • 若需要进度条，可用 tqdm 配合回调
相关文章

欢迎关注我的其它发布渠道