Python多进程应用案例 - 摩柯技术社区

多进程基础概念

在深入探讨Python多进程应用案例之前，我们先来回顾一下多进程的基本概念。进程是计算机中程序的一次执行实例，是系统进行资源分配和调度的基本单位。与线程不同，每个进程都有自己独立的内存空间，这意味着它们之间的数据不会相互干扰。

在操作系统层面，多进程机制允许计算机同时运行多个任务，充分利用多核CPU的性能优势。例如，在一台4核CPU的计算机上，理论上可以同时运行4个进程，使每个核心都得到充分利用，从而提高系统整体的处理能力。

Python提供了multiprocessing模块来支持多进程编程。这个模块允许我们创建和管理多个进程，实现并行计算，提高程序的执行效率。

简单的多进程示例

下面通过一个简单的示例来展示如何使用Python的multiprocessing模块创建和启动进程。

import multiprocessing


def worker():
    print('Worker function')


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

在上述代码中，我们首先导入了multiprocessing模块。然后定义了一个worker函数，这就是我们要在新进程中执行的任务。接下来，在if __name__ == '__main__':语句块中，我们创建了一个Process对象p，并将worker函数作为目标函数传递给它。调用p.start()方法启动进程，p.join()方法则等待进程执行完毕。

这里需要注意if __name__ == '__main__':语句的作用。在Windows系统下，multiprocessing模块在创建新进程时会重新导入主模块。如果没有这个语句，就会导致无限递归创建新进程的问题。而在Unix - like系统下，虽然不是必须的，但为了代码的跨平台兼容性，最好也加上这个语句。

多进程间通信

在实际应用中，多个进程之间往往需要进行数据交换，也就是通信。Python的multiprocessing模块提供了多种进程间通信的方式，如Queue、Pipe等。

使用Queue进行进程间通信

Queue是一个线程和进程安全的队列，可用于在多个进程之间传递数据。

import multiprocessing


def producer(queue):
    for i in range(5):
        queue.put(i)
        print(f'Produced {i}')


def consumer(queue):
    while True:
        item = queue.get()
        if item is None:
            break
        print(f'Consumed {item}')


if __name__ == '__main__':
    q = multiprocessing.Queue()
    p1 = multiprocessing.Process(target=producer, args=(q,))
    p2 = multiprocessing.Process(target=consumer, args=(q,))
    p1.start()
    p2.start()
    p1.join()
    q.put(None)
    p2.join()

在上述代码中，producer函数负责向Queue中放入数据，consumer函数则从Queue中取出数据。注意在producer进程结束后，我们向Queue中放入了一个None值作为结束信号，consumer函数通过判断取出的值是否为None来决定是否结束循环。

使用Pipe进行进程间通信

Pipe用于创建一个管道，两个进程可以通过这个管道进行通信。

import multiprocessing


def sender(pipe):
    conn, _ = pipe
    for i in range(5):
        conn.send(i)
        print(f'Sent {i}')
    conn.close()


def receiver(pipe):
    _, conn = pipe
    while True:
        try:
            item = conn.recv()
            print(f'Received {item}')
        except EOFError:
            break
    conn.close()


if __name__ == '__main__':
    parent_conn, child_conn = multiprocessing.Pipe()
    p1 = multiprocessing.Process(target=sender, args=((parent_conn, child_conn),))
    p2 = multiprocessing.Process(target=receiver, args=((parent_conn, child_conn),))
    p1.start()
    p2.start()
    p1.join()
    p2.join()

这里Pipe创建了两个连接对象parent_conn和child_conn。sender函数通过conn.send()方法向管道中发送数据，receiver函数则使用conn.recv()方法从管道中接收数据。当发送端关闭连接后，接收端会收到EOFError异常，从而结束循环。

多进程共享数据

虽然进程之间有独立的内存空间，但在某些情况下，我们需要多个进程共享一些数据。multiprocessing模块提供了Value和Array来实现基本数据类型和数组的共享。

使用Value共享单个值

import multiprocessing


def increment(value):
    with value.get_lock():
        value.value += 1


if __name__ == '__main__':
    num = multiprocessing.Value('i', 0)
    processes = []
    for _ in range(10):
        p = multiprocessing.Process(target=increment, args=(num,))
        processes.append(p)
        p.start()
    for p in processes:
        p.join()
    print(f'Final value: {num.value}')

在上述代码中，我们使用multiprocessing.Value创建了一个共享的整数变量num。increment函数通过value.get_lock()获取锁，确保在对num.value进行操作时不会出现数据竞争。

使用Array共享数组

import multiprocessing


def square_array(arr):
    for i in range(len(arr)):
        arr[i] = arr[i] * arr[i]


if __name__ == '__main__':
    numbers = multiprocessing.Array('i', [1, 2, 3, 4, 5])
    p = multiprocessing.Process(target=square_array, args=(numbers,))
    p.start()
    p.join()
    print(f'Squared array: {list(numbers)}')

这里我们使用multiprocessing.Array创建了一个共享的整数数组numbers。square_array函数对数组中的每个元素进行平方操作，由于数组是共享的，所以在进程执行完毕后，主进程可以看到数组的变化。

多进程应用案例：并行计算

并行计算是多进程的一个重要应用场景。通过将一个大的计算任务分解成多个子任务，分配到不同的进程中并行执行，可以显著提高计算效率。

计算列表元素的平方和

假设我们有一个包含大量数字的列表，需要计算这些数字的平方和。如果使用单进程计算，会花费较长时间。下面我们使用多进程来加速这个计算过程。

import multiprocessing


def square_sum_part(numbers, start, end):
    total = 0
    for num in numbers[start:end]:
        total += num * num
    return total


if __name__ == '__main__':
    numbers = list(range(1, 1000001))
    num_processes = multiprocessing.cpu_count()
    chunk_size = len(numbers) // num_processes
    processes = []
    results = multiprocessing.Queue()
    for i in range(num_processes):
        start = i * chunk_size
        end = start + chunk_size if i < num_processes - 1 else len(numbers)
        p = multiprocessing.Process(target=lambda q, s, e: q.put(square_sum_part(numbers, s, e)), args=(results, start, end))
        processes.append(p)
        p.start()
    total_sum = 0
    for _ in range(num_processes):
        total_sum += results.get()
    for p in processes:
        p.join()
    print(f'Total sum of squares: {total_sum}')

在上述代码中，我们首先根据CPU核心数确定进程数量num_processes，然后将列表numbers分成num_processes个部分。每个进程负责计算其中一部分的平方和，并将结果放入Queue中。最后，主进程从Queue中取出所有结果并累加，得到最终的平方和。

多进程应用案例：文件处理

在处理大量文件时，多进程可以提高处理效率。例如，我们有一批文本文件，需要统计每个文件中单词的出现次数，并汇总结果。

import multiprocessing
import os
import re


def count_words_in_file(file_path):
    word_count = {}
    with open(file_path, 'r', encoding='utf - 8') as file:
        for line in file:
            words = re.findall(r'\w+', line.lower())
            for word in words:
                if word in word_count:
                    word_count[word] += 1
                else:
                    word_count[word] = 1
    return word_count


def merge_word_counts(counts_list):
    total_count = {}
    for count in counts_list:
        for word, count_num in count.items():
            if word in total_count:
                total_count[word] += count_num
            else:
                total_count[word] = count_num
    return total_count


if __name__ == '__main__':
    folder_path = 'your_folder_path'
    file_paths = [os.path.join(folder_path, file) for file in os.listdir(folder_path) if file.endswith('.txt')]
    num_processes = multiprocessing.cpu_count()
    processes = []
    results = multiprocessing.Queue()
    for file_path in file_paths:
        p = multiprocessing.Process(target=lambda q, f: q.put(count_words_in_file(f)), args=(results, file_path))
        processes.append(p)
        p.start()
    all_counts = []
    for _ in range(len(file_paths)):
        all_counts.append(results.get())
    for p in processes:
        p.join()
    total_word_count = merge_word_counts(all_counts)
    for word, count in total_word_count.items():
        print(f'{word}: {count}')

在这个例子中，count_words_in_file函数负责统计单个文件中单词的出现次数。merge_word_counts函数用于合并多个文件的统计结果。主进程首先获取指定文件夹下所有文本文件的路径，然后为每个文件创建一个进程来统计单词数量，最后合并所有结果并输出。

多进程应用案例：网络爬虫

网络爬虫在处理大量网页时，多进程可以加快数据获取速度。以下是一个简单的多进程网络爬虫示例，用于获取多个网页的标题。

import multiprocessing
import requests
from bs4 import BeautifulSoup


def get_page_title(url):
    try:
        response = requests.get(url)
        soup = BeautifulSoup(response.content, 'html.parser')
        title = soup.title.string if soup.title else 'No title'
        return title
    except Exception as e:
        return f'Error: {e}'


if __name__ == '__main__':
    urls = [
        'https://www.example.com',
        'https://www.google.com',
        'https://www.github.com'
    ]
    num_processes = multiprocessing.cpu_count()
    processes = []
    results = multiprocessing.Queue()
    for url in urls:
        p = multiprocessing.Process(target=lambda q, u: q.put(get_page_title(u)), args=(results, url))
        processes.append(p)
        p.start()
    for _ in range(len(urls)):
        print(results.get())
    for p in processes:
        p.join()

在上述代码中，get_page_title函数通过requests库获取网页内容，并使用BeautifulSoup解析出网页标题。主进程为每个URL创建一个进程来获取标题，并从Queue中获取并输出结果。

多进程的性能优化与注意事项

性能优化

合理分配任务：确保每个进程的任务量相对均衡，避免某个进程负载过重，而其他进程空闲。例如在并行计算案例中，我们根据CPU核心数合理划分了列表计算的任务。
减少进程间通信开销：进程间通信会带来一定的性能开销，尽量减少不必要的通信。如果必须通信，可以选择更高效的通信方式，如Queue在大多数情况下比Pipe更适合数据传递。
避免资源竞争：对于共享数据，要正确使用锁机制，防止多个进程同时访问和修改导致数据不一致。

注意事项

内存消耗：每个进程都有自己独立的内存空间，创建过多进程会导致内存消耗过大。在实际应用中，要根据系统内存情况合理控制进程数量。
调试困难：多进程程序的调试比单进程程序更困难，因为进程间的执行顺序是不确定的。可以使用日志记录每个进程的执行情况，帮助定位问题。
操作系统差异：不同操作系统对多进程的支持和实现略有不同，如前面提到的Windows系统下需要if __name__ == '__main__':语句来避免进程创建递归问题。在编写跨平台代码时要特别注意这些差异。

通过以上对Python多进程的基础概念、通信方式、共享数据以及多个应用案例的介绍，相信你对Python多进程编程有了更深入的理解和掌握。在实际应用中，可以根据具体需求灵活运用多进程技术，提高程序的执行效率和性能。