【Python学习】进程和线程

对比维度	进程（Process）	线程（Thread）
资源分配	独立资源（内存、PID等）	共享所属进程的资源
调度单位	操作系统调度（资源分配单位）	CPU调度（执行单元）
启动速度	慢（需分配资源）	快（仅分配栈空间）
通信难度	难（需IPC机制）	易（共享内存、全局变量）
稳定性	高（进程崩溃不影响其他进程）	低（线程崩溃可能导致进程崩溃）
资源消耗	大	小

三、Python中的进程实现：multiprocessing模块

Python内置multiprocessing模块，用于实现多进程编程，支持创建进程、进程间通信、进程池等功能，解决了Python中“GIL锁”导致的多线程无法利用多核CPU的问题（后续会讲解GIL）。

3.1 基础用法：创建单个进程

使用multiprocessing.Process类创建进程，核心参数：

target：进程要执行的函数（必填）。
args：传递给函数的参数（元组类型）。
kwargs：传递给函数的关键字参数（字典类型）。
name：进程名称（可选）。

示例代码（可直接运行）：

import multiprocessing
import time

# 定义进程要执行的函数
def task(name, delay):
    for i in range(3):
        print(f"进程{name}：执行第{i+1}次，当前时间：{time.ctime()}")
        time.sleep(delay)  # 模拟任务耗时

if __name__ == "__main__":
    # 创建进程（target指定函数，args传递参数）
    p1 = multiprocessing.Process(target=task, args=("进程1", 1), name="p1")
    p2 = multiprocessing.Process(target=task, args=("进程2", 2), name="p2")

    # 启动进程
    p1.start()
    p2.start()

    # 等待进程执行完成（阻塞主进程，直到子进程结束）
    p1.join()
    p2.join()

    print("所有进程执行完毕！")

运行结果说明：两个进程会交替执行（因为CPU调度），进程1每1秒执行一次，进程2每2秒执行一次，主进程会等待两个子进程执行完再输出“所有进程执行完毕”。

关键注意点：if __name__ == "__main__": 必须加！因为Windows系统中，创建子进程时会重新导入当前模块，不加这句话会导致无限递归创建进程，报错。

3.2 进阶用法：进程池（Pool）

当需要创建大量进程时，手动创建多个Process对象效率低，且会消耗大量资源。此时可以使用multiprocessing.Pool创建进程池，实现进程的复用和管理。

进程池的核心方法：

Pool(n)：创建一个包含n个进程的进程池（n通常设为CPU核心数，可通过multiprocessing.cpu_count()获取）。
apply_async(func, args)：异步提交任务（非阻塞，主进程继续执行，子进程并行执行）。
close()：关闭进程池，不再接受新任务。
join()：等待进程池内所有任务执行完毕。

示例代码（批量处理任务）：

import multiprocessing
import time

# 定义任务函数（处理单个数据）
def process_data(data):
    time.sleep(1)  # 模拟数据处理耗时
    return f"数据{data}处理完成，结果：{data * 2}"

if __name__ == "__main__":
    # 获取CPU核心数，创建对应数量的进程池
    cpu_num = multiprocessing.cpu_count()
    pool = multiprocessing.Pool(cpu_num)

    # 批量提交任务（处理10个数据）
    data_list = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
    # 异步提交，返回结果对象列表
    result_list = [pool.apply_async(process_data, args=(data,)) for data in data_list]

    # 关闭进程池，等待所有任务完成
    pool.close()
    pool.join()

    # 获取所有任务结果
    for result in result_list:
        print(result.get())  # get()方法获取任务返回值

    print("所有数据处理完毕！")

运行结果说明：进程池会根据CPU核心数分配进程，并行处理10个数据，总耗时约1秒（而非10秒），大幅提升效率。

3.3 进程间通信（IPC）

进程之间相互独立，无法直接访问对方的内存，因此需要借助IPC机制实现通信。Python中常用的IPC方式有3种：队列（Queue）、管道（Pipe）、共享内存（Value/Array）。

3.3.1 队列（Queue）：最常用，线程安全

队列是“先进先出（FIFO）”的数据结构，multiprocessing.Queue支持多进程间的安全通信，适合多个进程之间传递数据。

import multiprocessing
import time

# 生产者进程：向队列中放入数据
def producer(queue):
    for i in range(5):
        data = f"数据{i+1}"
        queue.put(data)  # 放入队列
        print(f"生产者：放入{data}，当前队列大小：{queue.qsize()}")
        time.sleep(0.5)

# 消费者进程：从队列中取出数据
def consumer(queue):
    while True:
        if not queue.empty():  # 判断队列是否为空
            data = queue.get()  # 取出数据
            print(f"消费者：取出{data}，当前队列大小：{queue.qsize()}")
            time.sleep(1)
        else:
            break  # 队列空了，退出循环

if __name__ == "__main__":
    # 创建队列（默认无界队列，可指定maxsize设置最大容量）
    queue = multiprocessing.Queue()

    # 创建生产者和消费者进程
    p_producer = multiprocessing.Process(target=producer, args=(queue,))
    p_consumer = multiprocessing.Process(target=consumer, args=(queue,))

    # 启动进程
    p_producer.start()
    p_producer.join()  # 等待生产者放入所有数据

    p_consumer.start()
    p_consumer.join()

    print("通信完成！")

3.3.2 管道（Pipe）：适合两个进程间通信

管道通过multiprocessing.Pipe()创建，返回两个连接对象（conn1, conn2），两个进程分别通过这两个对象发送和接收数据，支持双向通信。

import multiprocessing
import time

# 进程1：发送数据
def send_data(conn):
    data_list = ["a", "b", "c", "d"]
    for data in data_list:
        conn.send(data)  # 发送数据
        print(f"进程1：发送{data}")
        time.sleep(0.5)
    conn.close()  # 关闭连接

# 进程2：接收数据
def recv_data(conn):
    while True:
        try:
            data = conn.recv()  # 接收数据
            print(f"进程2：接收{data}")
            time.sleep(1)
        except EOFError:
            break  # 管道关闭，退出循环

if __name__ == "__main__":
    # 创建管道（默认双向通信，duplex=False表示单向）
    conn1, conn2 = multiprocessing.Pipe()

    # 创建两个进程
    p1 = multiprocessing.Process(target=send_data, args=(conn1,))
    p2 = multiprocessing.Process(target=recv_data, args=(conn2,))

    p1.start()
    p2.start()

    p1.join()
    p2.join()

    print("管道通信完成！")

四、Python中的线程实现：threading模块

Python内置threading模块，用于实现多线程编程，线程共享进程资源，启动速度快，适合处理IO密集型任务（如网络请求、文件读写）。

注意：Python存在“GIL锁（全局解释器锁）”，导致同一时刻只有一个线程能执行Python字节码，因此多线程无法利用多核CPU，适合IO密集型任务，而非CPU密集型任务（CPU密集型任务建议用多进程）。

4.1 基础用法：创建单个线程

使用threading.Thread类创建线程，核心参数与Process类类似：target（执行函数）、args（参数元组）、kwargs（关键字参数）、name（线程名称）。

import threading
import time

# 定义线程要执行的函数
def task(name, delay):
    for i in range(3):
        print(f"线程{name}：执行第{i+1}次，当前时间：{time.ctime()}")
        time.sleep(delay)

if __name__ == "__main__":
    # 创建线程
    t1 = threading.Thread(target=task, args=("线程1", 1), name="t1")
    t2 = threading.Thread(target=task, args=("线程2", 2), name="t2")

    # 启动线程
    t1.start()
    t2.start()

    # 等待线程执行完成（阻塞主进程）
    t1.join()
    t2.join()

    print("所有线程执行完毕！")

运行结果说明：两个线程交替执行，与多进程类似，但启动速度更快，资源消耗更小。

4.2 线程的核心特性：守护线程（Daemon）

默认情况下，主进程会等待所有子线程执行完毕后才退出；如果将线程设置为“守护线程”（daemon=True），主进程退出时，守护线程会被强制终止，无论是否执行完毕。

应用场景：后台任务（如日志记录、心跳检测），主进程退出后，后台任务无需继续执行。

import threading
import time

def daemon_task():
    while True:
        print("守护线程：正在运行...")
        time.sleep(1)

def normal_task():
    for i in range(3):
        print("普通线程：执行第{i+1}次")
        time.sleep(1)

if __name__ == "__main__":
    # 创建守护线程（daemon=True）
    daemon_thread = threading.Thread(target=daemon_task, daemon=True)
    # 创建普通线程
    normal_thread = threading.Thread(target=normal_task)

    daemon_thread.start()
    normal_thread.start()

    # 等待普通线程执行完毕（主进程会等待普通线程，但不会等待守护线程）
    normal_thread.join()
    print("主进程退出！")

运行结果说明：普通线程执行3次后结束，主进程退出，此时守护线程被强制终止，不再打印“守护线程：正在运行...”。

4.3 线程安全问题：锁（Lock）

多个线程共享进程的全局变量、资源时，可能出现“竞态条件”（多个线程同时修改同一个资源，导致数据错误）。此时需要使用threading.Lock（锁）来保证线程安全，确保同一时刻只有一个线程能访问共享资源。

示例：未加锁导致的数据错误 vs 加锁后的数据正确

import threading
import time

# 全局变量（共享资源）
count = 0

# 未加锁的任务函数（会出现数据错误）
def task_without_lock():
    global count
    for _ in range(100000):
        count += 1  # 多个线程同时修改count，会出现竞态条件

# 加锁的任务函数（线程安全）
def task_with_lock(lock):
    global count
    for _ in range(100000):
        lock.acquire()  # 获取锁（阻塞，直到拿到锁）
        count += 1
        lock.release()  # 释放锁（必须释放，否则会导致死锁）

if __name__ == "__main__":
    # 1. 未加锁测试
    count = 0
    t1 = threading.Thread(target=task_without_lock)
    t2 = threading.Thread(target=task_without_lock)
    t1.start()
    t2.start()
    t1.join()
    t2.join()
    print(f"未加锁：count = {count}")  # 结果通常小于200000（数据错误）

    # 2. 加锁测试
    count = 0
    lock = threading.Lock()  # 创建锁对象
    t3 = threading.Thread(target=task_with_lock, args=(lock,))
    t4 = threading.Thread(target=task_with_lock, args=(lock,))
    t3.start()
    t4.start()
    t3.join()
    t4.join()
    print(f"加锁：count = {count}")  # 结果一定是200000（数据正确）

关键注意点：锁的acquire()和release()必须成对出现，否则会导致死锁（线程一直等待锁释放，无法继续执行）。也可以使用with lock:上下文管理器，自动释放锁，更安全：

def task_with_lock(lock):
    global count
    for _ in range(100000):
        with lock:  # 自动acquire()和release()
            count += 1

4.4 线程池：concurrent.futures.ThreadPoolExecutor

与进程池类似，线程池用于管理大量线程，避免频繁创建和销毁线程带来的开销。Python 3.2+ 提供concurrent.futures.ThreadPoolExecutor，用法与进程池类似。

from concurrent.futures import ThreadPoolExecutor
import time

# 定义任务函数
def process_data(data):
    time.sleep(1)
    return f"数据{data}处理完成，结果：{data * 2}"

if __name__ == "__main__":
    # 创建线程池（max_workers指定线程数量）
    with ThreadPoolExecutor(max_workers=4) as executor:
        # 批量提交任务
        data_list = [1, 2, 3, 4, 5, 6, 7, 8]
        # 异步提交，返回Future对象列表
        futures = [executor.submit(process_data, data) for data in data_list]

        # 获取所有任务结果
        for future in futures:
            print(future.result())  # result()方法获取返回值

    print("所有任务处理完毕！")

说明：with语句会自动关闭线程池，无需手动调用close()和join()，更简洁。

五、关键补充：GIL锁（全局解释器锁）

5.1 GIL锁是什么？

GIL锁是Python解释器（CPython）的一个全局锁，用于保证同一时刻只有一个线程能执行Python字节码。这是CPython的设计限制，目的是简化内存管理（避免多线程同时操作内存导致的冲突）。

5.2 GIL锁的影响

对IO密集型任务：影响不大。因为IO操作（如网络请求、文件读写）会阻塞线程，此时GIL锁会释放，其他线程可以执行，因此多线程能提升效率。
对CPU密集型任务：影响很大。因为CPU密集型任务（如计算、循环）会一直占用CPU，GIL锁无法释放，导致多线程只能串行执行，无法利用多核CPU，效率甚至不如单线程。此时建议使用多进程（进程有独立的GIL锁，可利用多核）。

5.3 如何规避GIL锁的限制？

CPU密集型任务：使用multiprocessing多进程，或使用Cython、NumPy等避开Python字节码的库。
IO密集型任务：使用threading多线程，或使用asyncio异步编程（后续可扩展）。

六、进程与线程的选择场景（实战指南）

实际开发中，选择进程还是线程，核心看任务类型和需求，总结如下：

6.1 选择多进程的场景

CPU密集型任务：如数据计算、图像处理、复杂算法（多进程可利用多核CPU，提升效率）。
任务之间相互独立，无需频繁通信：如批量处理多个独立文件、并行计算多个任务。
需要高稳定性：一个任务崩溃不影响其他任务（进程独立，互不干扰）。

6.2 选择多线程的场景

IO密集型任务：如网络爬虫、文件读写、接口请求（线程切换快，能利用IO阻塞的时间执行其他任务）。
任务之间需要频繁通信、共享资源：如实时数据处理、UI界面响应（线程共享内存，通信方便）。
对资源消耗敏感，需要快速启动：如后台轻量任务、实时监控。

6.3 总结对比

任务类型	推荐方案	原因
CPU密集型（计算多）	多进程	规避GIL锁，利用多核CPU提升效率
IO密集型（等待多）	多线程	线程切换快，利用等待时间并行执行
任务独立、高稳定	多进程	进程独立，一个崩溃不影响其他
频繁通信、共享资源	多线程	共享内存，通信方便、开销小

七、实战案例：多进程+多线程综合应用

需求：批量下载10张图片，使用多进程分配任务（每个进程处理2张图片），每个进程内部使用多线程并行下载图片（提升单进程内的下载效率）。

依赖库：requests（用于下载图片），需提前安装：pip install requests

import multiprocessing
from concurrent.futures import ThreadPoolExecutor
import requests
import os

# 图片下载函数（线程执行）
def download_image(url, save_path):
    try:
        response = requests.get(url, timeout=10)
        response.raise_for_status()  # 抛出HTTP错误
        with open(save_path, "wb") as f:
            f.write(response.content)
        print(f"成功下载：{os.path.basename(save_path)}")
    except Exception as e:
        print(f"下载失败：{url}，错误：{str(e)}")

# 进程任务函数（每个进程处理一批图片）
def process_batch(image_urls, save_dir):
    # 创建线程池，每个进程内启动2个线程
    with ThreadPoolExecutor(max_workers=2) as executor:
        # 批量提交下载任务
        futures = []
        for i, url in enumerate(image_urls):
            save_path = os.path.join(save_dir, f"image_{i+1}.jpg")
            futures.append(executor.submit(download_image, url, save_path))
        # 等待所有线程完成
        for future in futures:
            future.result()

if __name__ == "__main__":
    # 模拟10张图片URL（实际可替换为真实URL）
    image_urls = [
        "https://example.com/image1.jpg",
        "https://example.com/image2.jpg",
        "https://example.com/image3.jpg",
        "https://example.com/image4.jpg",
        "https://example.com/image5.jpg",
        "https://example.com/image6.jpg",
        "https://example.com/image7.jpg",
        "https://example.com/image8.jpg",
        "https://example.com/image9.jpg",
        "https://example.com/image10.jpg"
    ]

    # 创建保存目录
    save_dir = "downloaded_images"
    os.makedirs(save_dir, exist_ok=True)

    # 拆分任务：5个进程，每个进程处理2张图片
    batch_size = 2
    batches = [image_urls[i:i+batch_size] for i in range(0, len(image_urls), batch_size)]

    # 创建进程池，启动5个进程
    with multiprocessing.Pool(processes=5) as pool:
        # 批量提交进程任务
        futures = [pool.apply_async(process_batch, args=(batch, save_dir)) for batch in batches]
        # 等待所有进程完成
        for future in futures:
            future.get()

    print("所有图片下载完成！")

案例说明：结合多进程和多线程的优势，多进程利用多核CPU分配任务，多线程在每个进程内并行处理IO密集型的下载任务，大幅提升整体下载效率。