3、进程与并发：多进程架构、进程池的使用、进程间通信（IPC）

3.1 多进程架构——为什么我偏爱它？

说实话，做任务调度这些年，我踩过不少坑。单线程跑任务，CPU利用率低得可怜。多线程呢？GIL锁一卡，性能直接打回原形。

我个人习惯，遇到CPU密集型的任务，直接上多进程。为什么？每个进程有独立的Python解释器，没有GIL限制。你想想看，4核机器开4个进程，理论上能跑满400%的CPU。

多进程架构的核心思路很简单：

主进程：负责任务分发、结果收集、异常监控
工作进程：只干活，不操心别的
守护进程：定期检查工作进程是否还活着

我在项目中遇到过一个问题：某个工作进程因为第三方库的内存泄漏，悄悄死了。主进程还在傻等结果。后来我加了个心跳机制，每5秒检查一次，发现问题立刻重启进程。

核心原则：工作进程要设计成无状态的。有状态？那你就等着头疼吧。

3.2 进程池——别自己造轮子

很多人喜欢手动创建进程，一个任务一个进程。嗯，我刚开始也这么干。结果呢？进程创建销毁的开销，比任务本身还大。

Python的multiprocessing.Pool就是干这个的。它维护一组工作进程，任务来了直接丢进去。

from multiprocessing import Pool

def worker(task_id):
    # 模拟耗时计算
    result = task_id * 2
    return result

if __name__ == '__main__':
    # 创建4个工作进程的进程池
    with Pool(processes=4) as pool:
        tasks = [1, 2, 3, 4, 5, 6, 7, 8]
        results = pool.map(worker, tasks)
    print(results)  # [2, 4, 6, 8, 10, 12, 14, 16]

这里有个坑，我必须要说。进程池的大小不是越大越好。我曾经在一个32核的机器上，开了32个进程。结果呢？磁盘IO成了瓶颈，性能反而下降了。

我的经验：进程池大小 = CPU核心数 + 1。如果任务有IO等待，可以适当增加，但别超过核心数的2倍。

进程池还有几个实用方法：

apply_async：异步提交，不阻塞主进程
map_async：批量异步提交
imap：懒加载结果，适合大量任务

举个例子，如果你有10万个任务，用map会一次性把所有结果加载到内存。用imap呢？边处理边返回，内存占用小得多。

3.3 进程间通信（IPC）——数据怎么传？

进程之间不共享内存，这是多进程和多线程最大的区别。那数据怎么传？

Python提供了几种IPC方式：

方式	适用场景	性能	我踩过的坑
Queue	生产者-消费者模式	中等	队列满了会阻塞，小心死锁
Pipe	两个进程间通信	高	同时读写容易出问题
共享内存	大数据量传输	极高	需要自己处理同步
Manager	复杂数据结构共享	低	性能开销大，别用于高频场景

我个人最常用的是Queue。为什么？简单、安全、不容易出错。

from multiprocessing import Process, Queue

def producer(q):
    for i in range(5):
        q.put(f'任务{i}')
        print(f'生产: 任务{i}')

def consumer(q):
    while True:
        task = q.get()
        if task is None:  # 结束信号
            break
        print(f'消费: {task}')

if __name__ == '__main__':
    q = Queue()
    p1 = Process(target=producer, args=(q,))
    p2 = Process(target=consumer, args=(q,))
    
    p1.start()
    p2.start()
    
    p1.join()
    q.put(None)  # 发送结束信号
    p2.join()

注意：Queue的get()方法是阻塞的。如果生产者挂了，消费者会一直等下去。我曾经因为这个原因，线上任务卡了一整夜。解决方案：加超时或者心跳检测。

3.4 实战：一个简单的任务调度器

说了这么多，我们来写个实际能用的东西。一个简单的任务调度器，支持并发执行和结果收集。

from multiprocessing import Pool, Queue, Process
import time
import random

class TaskScheduler:
    def __init__(self, num_workers=4):
        self.num_workers = num_workers
        self.task_queue = Queue()
        self.result_queue = Queue()
        
    def worker(self):
        """工作进程：从队列取任务，执行，放结果"""
        while True:
            task = self.task_queue.get()
            if task is None:
                break
            # 模拟任务执行
            time.sleep(random.uniform(0.1, 0.5))
            result = f"任务{task}完成"
            self.result_queue.put(result)
    
    def start(self, tasks):
        """启动调度器"""
        # 创建工作进程
        workers = []
        for _ in range(self.num_workers):
            p = Process(target=self.worker)
            p.start()
            workers.append(p)
        
        # 分发任务
        for task in tasks:
            self.task_queue.put(task)
        
        # 发送结束信号
        for _ in range(self.num_workers):
            self.task_queue.put(None)
        
        # 等待所有工作进程结束
        for p in workers:
            p.join()
        
        # 收集结果
        results = []
        while not self.result_queue.empty():
            results.append(self.result_queue.get())
        
        return results

# 使用示例
if __name__ == '__main__':
    scheduler = TaskScheduler(num_workers=4)
    tasks = list(range(10))
    results = scheduler.start(tasks)
    print(results)

优化建议：实际项目中，我会把结果队列改成批量写入数据库，而不是攒在内存里。不然任务一多，内存就爆了。

3.5 避坑指南

做多进程开发，有几个坑我反复踩过：

子进程不能创建新的进程：Windows上会报错。解决办法：把创建逻辑放在if __name__ == '__main__'里。
日志要加进程ID：不然你根本不知道日志是谁打的。我习惯用logging模块加%(process)d。
资源清理：进程退出时，要确保数据库连接、文件句柄都关了。不然你会看到一堆僵尸进程。
信号处理：主进程收到SIGTERM，要优雅地通知子进程退出。别直接kill -9。

我曾经在一个项目里，因为忘了关数据库连接，导致连接池被占满。数据库直接挂了。嗯，那次被老板骂得不轻。

3.6 知识体系总览

下面这张图，是我自己总结的多进程并发知识体系。你看一眼，心里就有数了。

嗯，多进程这块内容不少。但说白了，核心就三点：架构设计要清晰、进程池大小要合理、IPC方式要选对。把这三点吃透了，大部分并发场景你都能搞定。

想要更多资料，联系可可

📚 课程目录

3、进程与并发：多进程架构、进程池的使用、进程间通信（IPC）

3.1 多进程架构——为什么我偏爱它？

3.2 进程池——别自己造轮子

3.3 进程间通信（IPC）——数据怎么传？

3.4 实战：一个简单的任务调度器

3.5 避坑指南

3.6 知识体系总览