Python多进程(Multiprocessing)编程实战：从进程池到分布式任务调度

Python虽然以简洁易用著称，但碍于全局解释器锁（GIL）的存在，多线程在CPU密集型任务中无法真正利用多核优势。当我们需要并行计算、批量数据处理或同时运行多个独立任务时，multiprocessing模块就是Python标准库中最强大的武器。

本文将从进程的创建与通信开始，逐步深入到进程池优化、共享内存、锁机制，最后介绍如何结合分布式任务队列构建真正的生产级并行系统。全程包含大量可直接运行的代码示例和最佳实践。

一、进程与线程的核心区别

在深入multiprocessing之前，首先要理解Python中进程和线程的根本差异：

简单总结：计算密集用多进程，IO密集用多线程或asyncio。

二、Process类的创建与管理

2.1 基本用法：启动子进程

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

import os



def worker(name: str):

    print(f"子进程 {name} 启动，PID: {os.getpid()}")



if __name__ == "__main__":

    # 创建进程对象

    p = multiprocessing.Process(target=worker, args=("worker-1",))

    p.start()

    print(f"主进程 PID: {os.getpid()}")

    p.join()

    print("子进程已结束")

输出类似：

1
2
3
主进程 PID: 12345

子进程 worker-1 启动，PID: 12346

子进程已结束

关键点：

• 一定要把进程创建代码放在

  1	if __name__ == "__main__"

  下面（Windows上spawn模式必须，Linux/macOS上也推荐）

• 1

  join()

  让主进程等待子进程结束，否则子进程会成为孤儿进程

• 每个进程有独立PID和内存空间

2.2 子类化Process

对于复杂任务，继承Process并重写run()是更优雅的方式：

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
import multiprocessing

import time



class DataProcessor(multiprocessing.Process):

    def __init__(self, data_chunk: list, process_id: int):

        super().__init__()

        self.data_chunk = data_chunk

        self.process_id = process_id

        self.result = []



    def run(self):

        print(f"处理器 #{self.process_id} 开始处理 {len(self.data_chunk)} 条数据")

        for item in self.data_chunk:

            # 模拟耗时计算

            time.sleep(0.1)

            self.result.append(item * 2)



if __name__ == "__main__":

    data = list(range(100))

    chunk_size = 25

    workers = []



    for i in range(4):

        chunk = data[i * chunk_size : (i + 1) * chunk_size]

        w = DataProcessor(chunk, i)

        w.start()

        workers.append(w)



    for w in workers:

        w.join()

        print(f"处理器 #{w.process_id} 完成，结果前5个: {w.result[:5]}")

三、进程间通信（IPC）

进程内存不共享，但multiprocessing提供了多种通信方式。

3.1 Queue（队列）

Queue是生产者-消费者模式的核心工具，线程安全且进程安全：

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
34
35
import multiprocessing

import time

import random



def producer(queue: multiprocessing.Queue):

    """生产者：生成任务"""

    for i in range(10):

        item = f"任务-{i}"

        queue.put(item)

        print(f"[生产者] 放入: {item}")

        time.sleep(random.uniform(0.1, 0.3))

    # 发送结束信号

    queue.put(None)



def consumer(queue: multiprocessing.Queue):

    """消费者：处理任务"""

    while True:

        item = queue.get()

        if item is None:  # 收到结束信号

            break

        print(f"[消费者] 处理: {item}")

        time.sleep(random.uniform(0.2, 0.5))



if __name__ == "__main__":

    q = multiprocessing.Queue()



    p1 = multiprocessing.Process(target=producer, args=(q,))

    p2 = multiprocessing.Process(target=consumer, args=(q,))



    p1.start()

    p2.start()



    p1.join()

    p2.join()

    print("所有任务完成")

3.2 Pipe（管道）

Pipe适用于两个进程间的双向通信：

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



def sender(conn: multiprocessing.Connection):

    """发送者"""

    conn.send("Hello from sender!")

    response = conn.recv()

    print(f"发送者收到: {response}")

    conn.close()



def receiver(conn: multiprocessing.Connection):

    """接收者"""

    msg = conn.recv()

    print(f"接收者收到: {msg}")

    conn.send("Reply from receiver!")

    conn.close()



if __name__ == "__main__":

    parent_conn, child_conn = multiprocessing.Pipe()



    p = multiprocessing.Process(target=receiver, args=(child_conn,))

    p.start()



    sender(parent_conn)

    p.join()

四、进程池（Pool）—— 最常用的并行工具

对于批量并行任务，手动管理进程费时费力。Pool提供了完美的封装。

4.1 map与imap（最常用）

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
import multiprocessing

import time



def cpu_intensive(n: int) -> int:

    """模拟CPU密集型计算"""

    total = 0

    for i in range(n):

        total += i * i

    return total



if __name__ == "__main__":

    nums = [10_000_000] * 8  # 8个大任务



    # 顺序执行（单进程）

    start = time.time()

    results_seq = [cpu_intensive(n) for n in nums]

    seq_time = time.time() - start

    print(f"顺序执行: {seq_time:.2f}s")



    # 并行执行（进程池）

    start = time.time()

    with multiprocessing.Pool(processes=4) as pool:

        results_par = pool.map(cpu_intensive, nums)

    par_time = time.time() - start

    print(f"并行执行 (4进程): {par_time:.2f}s")

    print(f"加速比: {seq_time / par_time:.2f}x")

在我的4核机器上，加速比通常在3.2x~3.8x之间（受进程创建开销和内存带宽影响）。

和

是懒加载版本，适合大数据集：

1
2
3
4
with multiprocessing.Pool(processes=4) as pool:

    # imap_unordered: 结果按完成顺序返回（不一定和输入顺序一致）

    for result in pool.imap_unordered(cpu_intensive, nums):

        print(f"得到一个结果: {result}")

4.2 apply_async：异步提交任务

当任务参数不同或需要更精细的控制时，使用apply_async：

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

import time



def process_item(item: dict) -> dict:

    time.sleep(0.5)

    return {"id": item["id"], "result": item["value"] * 2}



if __name__ == "__main__":

    tasks = [{"id": i, "value": i * 10} for i in range(20)]



    with multiprocessing.Pool(processes=4) as pool:

        # 提交所有任务

        async_results = [

            pool.apply_async(process_item, (task,))

            for task in tasks

        ]



        # 收集结果（保持顺序）

        results = [res.get() for res in async_results]



    print(f"处理完成，共 {len(results)} 个结果")

    print(f"前3个: {results[:3]}")

五、共享内存与锁

虽然multiprocessing推荐使用Queue/Pipe通信，但某些场景下共享内存更高效。

5.1 Value和Array

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



def increment(counter: multiprocessing.Value, n: int):

    for _ in range(n):

        counter.value += 1



if __name__ == "__main__":

    # 'i' 表示有符号整数类型

    counter = multiprocessing.Value("i", 0)



    processes = []

    for _ in range(4):

        p = multiprocessing.Process(target=increment, args=(counter, 100000))

        p.start()

        processes.append(p)



    for p in processes:

        p.join()



    print(f"期望值: {400000}, 实际值: {counter.value}")

你可能会发现实际值小于期望值。这是因为多个进程同时修改counter产生了竞态条件。解决方案是使用锁：

1
2
3
4
def increment_safe(counter: multiprocessing.Value, lock: multiprocessing.Lock, n: int):

    for _ in range(n):

        with lock:  # 加锁

            counter.value += 1

5.2 Manager：更高级的共享

Manager可以共享更复杂的数据结构（list, dict, set等）：

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

from multiprocessing.managers import BaseManager

import time



def worker(shared_dict: dict, worker_id: int):

    time.sleep(0.1 * worker_id)

    shared_dict[worker_id] = f"Worker-{worker_id} done"



if __name__ == "__main__":

    with multiprocessing.Manager() as manager:

        shared_dict = manager.dict()



        workers = [

            multiprocessing.Process(target=worker, args=(shared_dict, i))

            for i in range(5)

        ]



        for w in workers:

            w.start()

        for w in workers:

            w.join()



        print(f"共享字典内容: {dict(shared_dict)}")

注意：Manager的共享效率比原始共享内存低（通过IPC代理实现），适合低频读写场景。

六、实战案例：批量图片处理

下面是一个完整的生产级示例——用multiprocessing并行压缩图片：

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
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
import multiprocessing

import os

import time

from pathlib import Path

from PIL import Image



def compress_image(args: tuple) -> dict:

    """压缩单张图片"""

    src_path, dst_dir, quality = args

    try:

        filename = Path(src_path).name

        dst_path = Path(dst_dir) / filename



        with Image.open(src_path) as img:

            # 如果图片太大，先缩小尺寸

            if max(img.size) > 1920:

                img.thumbnail((1920, 1920), Image.LANCZOS)



            # 保存压缩版

            img.save(dst_path, optimize=True, quality=quality)



        original_size = os.path.getsize(src_path)

        compressed_size = os.path.getsize(dst_path)

        ratio = compressed_size / original_size



        return {

            "filename": filename,

            "original": original_size,

            "compressed": compressed_size,

            "ratio": f"{ratio:.1%}",

            "status": "ok",

        }

    except Exception as e:

        return {"filename": Path(src_path).name, "status": "error", "error": str(e)}



def batch_compress(src_dir: str, dst_dir: str, quality: int = 85, workers: int = None):

    """批量并行压缩图片"""

    src = Path(src_dir)

    dst = Path(dst_dir)

    dst.mkdir(parents=True, exist_ok=True)



    image_files = [

        str(f) for f in src.glob("*")

        if f.suffix.lower() in (".jpg", ".jpeg", ".png", ".webp")

    ]



    if not image_files:

        print("未找到图片文件")

        return



    if workers is None:

        workers = multiprocessing.cpu_count()



    print(f"找到 {len(image_files)} 张图片，使用 {workers} 个进程并行处理")



    start = time.time()

    with multiprocessing.Pool(processes=workers) as pool:

        results = list(pool.imap_unordered(

            compress_image,

            [(f, str(dst), quality) for f in image_files],

            chunksize=4,  # 每个进程一次取4个任务

        ))



    elapsed = time.time() - start



    # 统计结果

    ok = [r for r in results if r["status"] == "ok"]

    errors = [r for r in results if r["status"] == "error"]



    print(f"\n处理完成！")

    print(f"成功: {len(ok)}, 失败: {len(errors)}")

    print(f"耗时: {elapsed:.2f}s")



    if ok:

        total_original = sum(r["original"] for r in ok)

        total_compressed = sum(r["compressed"] for r in ok)

        print(f"原始大小: {total_original / 1024 / 1024:.1f} MB")

        print(f"压缩后: {total_compressed / 1024 / 1024:.1f} MB")

        print(f"压缩率: {total_compressed / total_original:.1%}")



    if errors:

        print("\n错误详情:")

        for e in errors:

            print(f"  {e['filename']}: {e['error']}")



if __name__ == "__main__":

    batch_compress(

        src_dir="/path/to/original/images",

        dst_dir="/path/to/compressed/images",

        quality=80,

        workers=4

    )

这个示例展示了：

• 1	imap_unordered

  +

  1

  chunksize

  提升效率

• 任务参数的元组打包传递
• 结果收集与状态统计
• 异常处理（单任务失败不影响整体）
• 自动检测CPU核数

七、分布式任务调度——延伸生产级能力

multiprocessing.Pool虽然强大，但有局限：所有进程必须在同一台机器上。当数据量达到TB级别或需要跨机器调度时，就需要分布式任务队列。

7.1 Celery：工业级分布式任务队列

Celery是最流行的Python分布式任务框架：

1
2
3
4
5
6
7
8
9
10
11
12
13
# tasks.py

from celery import Celery



app = Celery("tasks", broker="redis://localhost:6379/0")



@app.task

def process_data_chunk(chunk_id: int, data_range: tuple):

    """在任意worker节点上执行"""

    start, end = data_range

    total = 0

    for i in range(start, end):

        total += i * i

    return {"chunk_id": chunk_id, "result": total}

1
2
3
4
5
6
7
8
# 启动worker

celery -A tasks worker --concurrency=4 -l info



# 提交任务

from tasks import process_data_chunk



result = process_data_chunk.delay(1, (0, 1000000))

print(result.get())  # 阻塞等待结果

7.2 TaskQueue：轻量替代方案

对小型项目来说Celery太重，可以用Redis + multiprocessing自建轻量队列：

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

import redis

import json



class DistributedWorker(multiprocessing.Process):

    def __init__(self, queue_name: str, worker_id: int):

        super().__init__()

        self.queue_name = queue_name

        self.worker_id = worker_id

        self.redis_client = redis.Redis(host="localhost", port=6379, db=0)



    def run(self):

        print(f"Worker-{self.worker_id} 启动，监听队列: {self.queue_name}")

        while True:

            # 阻塞获取任务（BRPOP支持超时）

            task_data = self.redis_client.brpop(self.queue_name, timeout=5)

            if task_data is None:

                continue



            _, task_json = task_data

            task = json.loads(task_json)



            if task.get("shutdown"):

                print(f"Worker-{self.worker_id} 收到关闭信号")

                break



            # 处理任务

            result = self.process(task)

            print(f"Worker-{self.worker_id} 完成任务: {task['id']}, 结果: {result}")



    def process(self, task: dict):

        """实际的任务处理逻辑"""

        return {k: v * 2 for k, v in task.get("data", {}).items()}

八、性能调优与最佳实践

8.1 进程数选择

并非进程越多越好：

• CPU密集：设为

  1	multiprocessing.cpu_count()

• IO密集+CPU：设为

  1	cpu_count() * 2

• 内存密集：需要根据可用内存反推（每个进程复制了Python环境）

8.2 chunksize 调优

对于

和

，chunksize决定了每个进程一次领取的任务数：

8.3 避免常见陷阱

• 不要在Windows上使用fork（默认spawn，对象序列化问题）
• 避免传递大对象：进程间传递对象需要序列化（pickle），大对象造成巨大开销
• 日志文件冲突：多个进程写入同一日志文件需要进程安全锁或独立文件
• 数据库连接池：每个进程需要独立的数据库连接，不能共享
• 子线程中的子进程：从线程中创建进程可能有诡异行为，尽量避免

8.4 使用concurrent.futures更现代的API

Python 3.2+ 提供了更简洁的ProcessPoolExecutor：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from concurrent.futures import ProcessPoolExecutor

import time



def compute(n: int) -> int:

    return sum(i * i for i in range(n))



if __name__ == "__main__":

    with ProcessPoolExecutor(max_workers=4) as executor:

        nums = [5_000_000] * 8



        # map方式

        results = list(executor.map(compute, nums))



        # submit方式（更灵活）

        futures = [executor.submit(compute, n) for n in nums]

        results2 = [f.result() for f in futures]



    print(results[:3])

ProcessPoolExecutor底层封装了multiprocessing.Pool，API更现代，推荐新项目使用。

总结

本文从进程与线程的本质区别出发，系统介绍了Python multiprocessing模块的五大核心能力：

1. Process：基础进程创建与管理
2. Queue/Pipe：进程间通信
3. Pool：批量并行任务处理
4. 共享内存与锁：高效数据共享
5. 分布式扩展：从Pool到Celery的生产级方案

在实际项目中，80%的并行场景使用

或

即可解决。遇到更复杂的任务编排、跨机器调度或失败重试需求时，再考虑引入Celery或自建Redis队列。

记住一个核心原则：先测单进程性能，再考虑并行；先确认是CPU瓶颈，再使用多进程。用对工具比用花哨工具更重要。