Python多进程案例如何创建进程

本文目录导读：

1. 使用Process类（最基础）
2. 继承Process类（面向对象方式）
3. 使用Pool进程池（批量管理）
4. 进程间通信示例
5. 实际应用：多进程下载器
6. 重要注意事项

我来详细介绍Python多进程创建进程的几种主要方式：

使用Process类（最基础）

from multiprocessing import Process
import os
def worker(name):
    print(f'子进程 {name} 运行中，PID: {os.getpid()}')
if __name__ == '__main__':
    print(f'主进程 PID: {os.getpid()}')
    # 创建多个子进程
    processes = []
    for i in range(3):
        p = Process(target=worker, args=(f'进程-{i}',))
        processes.append(p)
        p.start()
    # 等待所有子进程结束
    for p in processes:
        p.join()
    print('所有进程执行完毕')

继承Process类（面向对象方式）

from multiprocessing import Process
import os
import time
class MyProcess(Process):
    def __init__(self, name, delay):
        super().__init__()
        self.name = name
        self.delay = delay
    def run(self):  # 重写run方法
        print(f'{self.name} 开始执行，PID: {os.getpid()}')
        time.sleep(self.delay)
        print(f'{self.name} 执行完成')
if __name__ == '__main__':
    # 创建进程实例
    p1 = MyProcess('进程A', 2)
    p2 = MyProcess('进程B', 3)
    # 启动进程
    p1.start()
    p2.start()
    # 等待完成
    p1.join()
    p2.join()
    print('所有进程执行完毕')

使用Pool进程池（批量管理）

from multiprocessing import Pool
import os
import time
def square(n):
    time.sleep(1)  # 模拟耗时操作
    result = n * n
    print(f'进程 {os.getpid()} 计算: {n}^2 = {result}')
    return result
if __name__ == '__main__':
    # 创建进程池，默认使用CPU核心数
    with Pool(processes=4) as pool:
        # 方法1: 同步执行
        # results = pool.map(square, range(10))
        # 方法2: 异步执行
        results = pool.map_async(square, range(10))
        results.wait()  # 等待所有完成
        print(f'最终结果: {results.get()}')
    print('进程池使用完毕')

进程间通信示例

from multiprocessing import Process, Queue
import os
import time
def producer(queue):
    """生产者进程"""
    for i in range(5):
        data = f'数据-{i}'
        print(f'生产者 {os.getpid()} 发送: {data}')
        queue.put(data)
        time.sleep(0.5)
def consumer(queue):
    """消费者进程"""
    while True:
        data = queue.get()
        if data == 'END':  # 结束信号
            break
        print(f'消费者 {os.getpid()} 接收: {data}')
        time.sleep(1)
if __name__ == '__main__':
    # 创建队列用于进程间通信
    queue = Queue()
    # 创建生产者和消费者进程
    p1 = Process(target=producer, args=(queue,))
    p2 = Process(target=consumer, args=(queue,))
    # 启动进程
    p1.start()
    p2.start()
    # 等待生产者完成
    p1.join()
    # 发送结束信号
    queue.put('END')
    # 等待消费者完成
    p2.join()
    print('通信完成')

实际应用：多进程下载器

from multiprocessing import Process, Queue
import requests
import time
import os
def download_file(url, save_path):
    """下载文件"""
    try:
        print(f'进程 {os.getpid()} 开始下载: {url}')
        response = requests.get(url, stream=True)
        with open(save_path, 'wb') as f:
            for chunk in response.iter_content(chunk_size=8192):
                if chunk:
                    f.write(chunk)
        print(f'下载完成: {url}')
        return True
    except Exception as e:
        print(f'下载失败 {url}: {e}')
        return False
def worker(task_queue):
    """工作进程"""
    while not task_queue.empty():
        try:
            url, save_path = task_queue.get_nowait()
            download_file(url, save_path)
        except:
            break
if __name__ == '__main__':
    # 下载任务列表
    downloads = [
        ('https://example.com/file1.zip', 'downloads/file1.zip'),
        ('https://example.com/file2.zip', 'downloads/file2.zip'),
        ('https://example.com/file3.zip', 'downloads/file3.zip'),
        ('https://example.com/file4.zip', 'downloads/file4.zip'),
    ]
    # 创建任务队列
    task_queue = Queue()
    for url, path in downloads:
        task_queue.put((url, path))
    # 创建进程池
    processes = []
    num_processes = min(3, len(downloads))  # 最多3个进程
    for _ in range(num_processes):
        p = Process(target=worker, args=(task_queue,))
        processes.append(p)
        p.start()
    # 等待所有进程完成
    for p in processes:
        p.join()
    print('所有下载任务完成')

重要注意事项

1. if __name__ == '__main__': - Windows系统必须使用这个保护
2. 进程间不共享全局变量 - 需要使用Queue、Pipe等通信
3. 资源消耗 - 每个进程有自己的内存空间，比线程开销大
4. 调试困难 - 进程间独立，错误传递比较复杂
5. 避免死锁 - 使用join()时注意顺序

选择哪种方式取决于你的具体需求：简单任务用Process类，批量任务用Pool，复杂逻辑可以继承Process类。