Python 并行计算

12. Python 并行计算#

在进入并行计算前，我们首先需要了解一下一些计算机操作系统的知识。

12.1. 进程与线程#

  • 进程是一个程序的运行实例，每个进程有独立的内存空间，相互隔离，不会直接共享数据。

  • 线程是进程内部的一个执行单元，一个进程可以有多个线程，它们共享同一个进程的内存。

现代浏览器（如 Chrome）普遍采用多进程架构，打开一些网页时，可能包含多个进程：

  • 主进程：管理 UI、窗口、用户输入等。

  • 渲染进程：负责渲染网页（HTML、CSS、JavaScript），通常每个标签页一个渲染进程。

  • 插件进程：用于运行 Flash、PDF 之类的插件。

  • GPU 进程：负责 GPU 加速绘图。

当我们打开两个网页时，浏览器运行的进程结构如下：

├── 浏览器主进程
├── 渲染进程（网页A）
├── 渲染进程（网页B）
├── GPU 进程
├── 插件进程

进程的特点是：

  • 进程之间相互隔离，例如一个网页崩溃不会影响其他网页。

  • 进程有独立的内存空间，所以更安全，但开销较大（占用更多内存）。

在浏览器的渲染进程中，负责执行 JavaScript、渲染页面的任务实际上是由多个线程完成的：

  • 主线程（Main Thread）：执行 JavaScript、解析 HTML、CSS、布局计算、渲染等。

  • 工作线程（Worker Thread）：通过 Web Worker 执行耗时任务（如数据处理），避免阻塞主线程。

  • 网络线程：负责处理 HTTP 请求、下载资源。

  • GPU 线程：处理页面渲染加速。

其中，渲染进程的线程结构如下

渲染进程（网页A）
├── 主线程（执行 JavaScript、渲染页面）
├── 工作线程（Web Worker 处理数据）
├── 网络线程（加载图片、请求 API）
├── GPU 线程（处理绘图）

线程的特点：

  • 线程共享同一个进程的内存，可以快速通信，但可能会引起数据竞争问题（需要同步）。

  • 线程的开销比进程小，但如果主线程被阻塞，其他线程可能受影响，例如整个网页可能会卡死。

下面这个例子显著地描述了进程与线程的区别。

  • 进程 = 一家餐厅，每个进程是一个独立的厨房

    • 每个厨房（进程）都独立工作，互不影响，即使一个厨房着火（崩溃），其他厨房还能正常运作。

  • 线程 = 厨房里的厨师

    • 一个厨房（进程）里有多个厨师（线程），他们共享同一套厨房用具（内存）。

    • 如果厨师合作良好，出餐效率高；但如果一个厨师卡住（主线程阻塞），整个厨房可能会停滞。

12.2. multirpocessing 包#

multiprocessing 是 Python 标准库中的一个包，提供了多进程（并行）执行能力，允许 Python 代码充分利用多核 CPU，提高计算效率。

另外一个并行计算的包是threading，提供了多线程（multithreading） 支持，允许多个任务并发（concurrent）执行。

  • multiprocessing 适用于 CPU 密集型任务，如矩阵运算、加密、图像处理。

  • threading 适用于 I/O 密集型任务，如文件读写、爬取网页。

12.2.1. Pool（自动管理进程池）#

使用多进程最便捷的工具是使用multiprocessing.Pool自动管理进程池，不需要向其他进程工具那样需要手动开启关闭进程。Pool 提供了多个方法来调用进入并行计算的函数。

12.2.1.1. map()（适用于单参数函数，自动收集返回值）#

import multiprocessing

def worker(x):
    return x * x

if __name__ == '__main__':
    with multiprocessing.Pool(processes=4) as pool:
        results = pool.map(worker, range(10))  # 并行计算
    print(results)

上面的代码中：

  • Pool(processes=4) 创建 4 个进程，自动分配任务。

  • map(worker, range(10)) 将 0~9 作为参数传递给 worker(x) 并行执行。

使用multiprocessing.cpu_count()可以获取当前电脑的 cpu 的核数。

Note

并行计算的程序必须放在if __name__ == '__main__' 里，防止 Windows/macOS 递归创建进程导致崩溃。

12.2.1.2. starmap()（适用于多参数函数）#

def worker(x, y):
    return x + y

if __name__ == '__main__':
    with multiprocessing.Pool(multiprocessing.cpu_count()) as pool:
        results = pool.starmap(worker, [(1, 2), (3, 4), (5, 6)])
    print(results)

对于 starmap()：

  • 适用于需要多个参数的函数。

  • 参数必须以 [( )] 形式传递，小括号()内不能只有一个参数。

  • 可传递空参数，此时小括号()内无任何内容。

12.2.2. Process（手动创建进程）#

可以通过 Process 手动创建进程。

import multiprocessing

def worker(num):
    print(f"Process {num} is running")

if __name__ == '__main__':
    processes = []
    for i in range(5):
        p = multiprocessing.Process(target=worker, args=(i,))
        processes.append(p)
        p.start() # start the process

    for p in processes:
        p.join()  # stop the process

  • target 后面跟的是目标函数名字

  • args 是传递到目标函数中的参数值

  • args 必须传递一个元祖，因此单参数时后面必须跟一个逗号

  • start()开启进程

  • join()关闭进程

12.2.3. Queue（进程间通信）#

进程间不能直接共享变量，但可以用Queue传递数据。

import multiprocessing

def worker(queue, num):
    queue.put(num * num)  # 进程安全地存入队列

if __name__ == '__main__':
    queue = multiprocessing.Queue()  # 共享队列
    processes = []

    for i in range(5):
        p = multiprocessing.Process(target=worker, args=(queue, i))
        processes.append(p)
        p.start()

    for p in processes:
        p.join()

    results = [queue.get() for _ in range(5)]
    print(results)

  • 使用put()将数据推进 Queue 里

  • 使用get()取出数据

因此，Process 经常结合Queue一起使用，获取函数返回的数据。

12.2.4. Lock（控制进程）#

在 Python 的 multiprocessing 模块中，Lock 是用于同步进程之间访问共享资源的工具。它确保一次只有一个进程可以访问受保护的代码区域，从而防止竞争条件 (race condition) 和数据不一致问题。

import multiprocessing
import time

def worker(lock, num):
    with lock:  # 使用 with 语句自动管理锁的获取和释放
        print(f'Process {num} is starting...')
        time.sleep(1)
        print(f'Process {num} is done.')

if __name__ == '__main__':
    lock = multiprocessing.Lock()  # 创建锁对象

    processes = []
    for i in range(5):
        p = multiprocessing.Process(target=worker, args=(lock, i))
        processes.append(p)
        p.start()

    for p in processes:
        p.join()

  • multiprocessing.Lock() 创建了一个进程锁。

  • 在 worker() 函数中，使用 with lock: 保护关键代码区域，确保一次只有一个进程能执行这部分代码。

  • 由于锁的作用，五个进程会依次执行，而不是同时输出信息。

multiprocessing 包中其他并行计算的工具还有Manager(管理多个进程的共享数据)， pipe (允许两个进程之间通信数据)。