当前位置：首页 > article >正文

Python 并发新境界：探索 `multiprocessing` 模块的无限可能

article 2024/11/20 14:44:03

引言

随着硬件技术的发展，多核处理器已经成为标准配置。这意味着我们的计算机拥有执行多个任务的能力。然而，默认情况下，Python程序由于全局解释器锁（GIL）的存在，并不能充分利用这些核心资源。这就引出了multiprocessing模块的重要性——它通过创建独立进程来绕过GIL限制，从而实现真正的并行计算。

multiprocessing模块的应用场景非常广泛，从简单的文件处理到复杂的科学计算，甚至是Web爬虫等都可以见到它的身影。接下来，我们将逐步了解如何利用这一强大工具来优化我们的Python程序。

基础语法介绍

核心概念

进程：一个独立的程序实例，有自己的内存空间，可以并行执行。
Pool：用于管理一组工作进程，提供了一个简单的方式来分发任务给不同的进程。
Queue：进程间通信的一种方式，允许不同进程之间传递对象。
Lock/Semaphore：控制多个线程/进程对共享资源的访问，防止竞态条件。

基本语法规则

创建进程的基本语法如下所示：

from multiprocessing import Process

def worker():
    print('I am a worker')

if __name__ == '__main__':
    p = Process(target=worker)
    p.start()
    p.join()

这里，我们定义了一个名为worker的函数，然后通过Process类创建了一个新的进程对象p，并将worker作为目标函数传入。最后调用start()方法启动进程，并通过join()等待其完成。

基础实例

假设我们需要对一批图片进行处理（例如调整大小），如果单靠主线程执行将会非常耗时。这时，我们可以借助multiprocessing模块来加速这一过程。

from multiprocessing import Pool
import os, time, random

def long_time_task(name):
    print(f'Run task {name} (pid: {os.getpid()})')
    start = time.time()
    time.sleep(random.random() * 3)
    end = time.time()
    print(f'Task {name} runs {end-start:.2f} seconds')

if __name__ == '__main__':
    print('Parent process %s.' % os.getpid())
    p = Pool(4)
    for i in range(5):
        p.apply_async(long_time_task, args=(i,))
    print('Waiting for all subprocesses done...')
    p.close()
    p.join()
    print('All subprocesses done.')

在这个例子中，我们创建了一个包含4个子进程的Pool对象，并使用apply_async方法异步地将任务分发给它们。注意close和join的使用，前者告诉Pool不再接受新的任务，后者则会阻塞主进程直到所有子进程完成。

进阶实例

当涉及到更复杂的应用场景时，如大规模数据分析或分布式系统开发，multiprocessing模块同样能够发挥巨大作用。比如下面这个示例展示了如何使用Manager来创建共享对象，从而实现进程间的数据交换。

from multiprocessing import Manager, Process

def f(d, l):
    d[1] = '1'
    d['2'] = 2
    d[0.25] = None
    l.reverse()

if __name__ == '__main__':
    manager = Manager()
    d = manager.dict()
    l = manager.list(range(10))

    p_list = []
    for i in range(10):
        p = Process(target=f, args=(d, l))
        p.start()
        p_list.append(p)
    
    for res in p_list:
        res.join()

    print(d)
    print(l)

这里我们通过Manager创建了两个可以被多个进程共享的对象：字典d和列表l。每个进程都会修改这些共享对象的内容，最终结果表明所有修改都被正确地同步到了所有进程中。

实战案例

在实际工作中，multiprocessing模块往往与其他库结合使用以解决特定问题。以一个典型的Web爬虫项目为例，我们不仅需要抓取网页数据，还需要对其进行解析和存储。这无疑是一个耗时的过程，特别是当目标网站较多时。此时，我们可以考虑将整个流程拆分成几个阶段，并行处理每一部分以提高效率。

from bs4 import BeautifulSoup
from urllib.request import urlopen
from multiprocessing import Process, Queue

def crawl(url, q):
    response = urlopen(url)
    html = response.read().decode('utf-8')
    soup = BeautifulSoup(html, features='html.parser')
    q.put(soup)

def parse(q):
    while True:
        try:
            item = q.get_nowait()
        except Empty:
            break
        else:
            # 对item进行解析操作...

if __name__ == '__main__':
    urls = ['http://example.com/page%d' % page for page in range(1, 11)]
    q = Queue()

    crawlers = [Process(target=crawl, args=(url, q)) for url in urls]
    parsers = [Process(target=parse, args=(q,)) for _ in range(5)]

    for crawler in crawlers:
        crawler.start()

    for parser in parsers:
        parser.start()

    for crawler in crawlers:
        crawler.join()

    for parser in parsers:
        parser.join()

上述代码首先定义了两个函数crawl和parse分别负责抓取和解析工作。接着我们创建了一个队列q用于存放中间结果，并根据需求启动相应数量的爬虫进程和解析进程。通过这种方式，整个爬虫系统的吞吐量得到了显著提升。

扩展讨论

尽管multiprocessing模块提供了丰富的功能，但我们在实际使用过程中仍需注意以下几点：

进程间通信：虽然multiprocessing模块内置了多种通信机制，但在设计时仍然需要谨慎选择合适的方法，以保证数据传输的高效性和安全性。
资源共享：由于每个进程都有自己独立的内存空间，因此对于那些需要在多个进程中共享的数据，必须通过特殊手段（如Manager提供的共享对象）来实现。
异常处理：与普通线程相比，进程之间的异常传播更加复杂，因此在编写多进程程序时应当充分考虑到这一点，并做好相应的容错设计。
死锁问题：当多个进程同时尝试访问同一资源时，可能会导致死锁现象发生。为了避免这种情况，通常需要合理安排各进程的执行顺序或引入锁定机制。
性能考量：虽然多进程可以有效提升程序运行速度，但也并非万能药。特别是在处理小型任务时，频繁创建和销毁进程反而会增加额外开销。因此，在决定是否使用multiprocessing之前，最好先进行性能测试。