Python并发编程:多进程(multiprocessing模块)

简介: 在处理CPU密集型任务时,Python的全局解释器锁(GIL)可能会成为瓶颈。为了充分利用多核CPU的性能,可以使用Python的multiprocessing模块来实现多进程编程。与多线程不同,多进程可以绕过GIL,使得每个进程在自己的独立内存空间中运行,从而实现真正的并行计算。

在处理CPU密集型任务时,Python的全局解释器锁(GIL)可能会成为瓶颈。为了充分利用多核CPU的性能,可以使用Python的multiprocessing模块来实现多进程编程。与多线程不同,多进程可以绕过GIL,使得每个进程在自己的独立内存空间中运行,从而实现真正的并行计算。

一、什么是multiprocessing模块

multiprocessing模块是Python标准库中的一个模块,它提供了一个类似于threading模块的接口,用于创建和管理进程。通过这个模块,我们可以轻松地创建进程池、管理进程间的通信和同步等。

1.1 模块导入

要使用multiprocessing模块,首先需要导入它:

import multiprocessing

1.2 创建进程

使用multiprocessing模块,我们可以通过创建一个Process对象来启动一个新进程。Process对象接受一个目标函数和可选的参数,当调用start()方法时,目标函数将在新进程中执行。

from multiprocessing import Process
def worker():
    print("Worker process")
if __name__ == "__main__":
    p = Process(target=worker)
    p.start()
    p.join()

二、进程间通信

在多进程编程中,进程之间的通信和数据共享是一个常见需求。multiprocessing模块提供了几种机制来实现进程间的通信,包括队列(Queue)、管道(Pipe)和共享内存(Value和Array)。

2.1 队列(Queue)

队列是一种FIFO(先进先出)数据结构,适用于进程间的消息传递。multiprocessing.Queue提供了一个线程和进程安全的队列接口。

from multiprocessing import Process, Queue
def worker(q):
    q.put("Hello from worker")
if __name__ == "__main__":
    q = Queue()
    p = Process(target=worker, args=(q,))
    p.start()
    print(q.get())
    p.join()

2.2 管道(Pipe)

管道提供了一个双向的通信通道,允许两个进程之间相互发送消息。

from multiprocessing import Process, Pipe
def worker(conn):
    conn.send("Hello from worker")
    conn.close()
if __name__ == "__main__":
    parent_conn, child_conn = Pipe()
    p = Process(target=worker, args=(child_conn,))
    p.start()
    print(parent_conn.recv())
    p.join()

2.3 共享内存(Value和Array)

共享内存允许多个进程直接访问共享的变量。multiprocessing.Value和multiprocessing.Array提供了这样的共享内存接口。

from multiprocessing import Process, Value, Array
def worker(num, arr):
    num.value = 42
    for i in range(len(arr)):
        arr[i] = -arr[i]
if __name__ == "__main__":
    num = Value('i', 0)
    arr = Array('i', range(10))
    p = Process(target=worker, args=(num, arr))
    p.start()
    p.join()
    print(num.value)
    print(arr[:])

三、进程池(Pool)

multiprocessing.Pool提供了一个方便的接口,用于管理进程池。进程池允许我们一次创建多个进程,并通过池中的进程并行执行多个任务。

3.1 创建进程池

我们可以使用Pool对象创建一个进程池,并通过apply、map等方法将任务分配给进程池中的进程。

from multiprocessing import Pool
def worker(x):
    return x * x
if __name__ == "__main__":
    with Pool(4) as p:
        print(p.map(worker, range(10)))

3.2 进程池的异步操作

进程池还支持异步操作,可以通过apply_async和map_async方法提交异步任务,并通过get方法获取结果。

from multiprocessing import Pool
def worker(x):
    return x * x
if __name__ == "__main__":
    with Pool(4) as p:
        result = p.apply_async(worker, (10,))
        print(result.get())
        multiple_results = [p.apply_async(worker, (i,)) for i in range(10)]
        print([res.get() for res in multiple_results])

四、锁和同步

在多进程编程中,多个进程可能会同时访问共享资源,导致数据竞争和不一致的问题。multiprocessing模块提供了多种同步机制,包括锁(Lock)、信号量(Semaphore)和条件变量(Condition)等。

4.1 锁(Lock)

锁用于确保在同一时间只有一个进程能够访问共享资源。

from multiprocessing import Process, Lock
def worker(lock, num):
    with lock:
        print(f"Worker {num}")
if __name__ == "__main__":
    lock = Lock()
    processes = [Process(target=worker, args=(lock, i)) for i in range(5)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()

4.2 信号量(Semaphore)

信号量用于控制对共享资源的访问数量,适用于允许多个进程同时访问共享资源的情况。

from multiprocessing import Process, Semaphore
def worker(semaphore, num):
    with semaphore:
        print(f"Worker {num}")
if __name__ == "__main__":
    semaphore = Semaphore(2)
    processes = [Process(target=worker, args=(semaphore, i)) for i in range(5)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()

4.3 条件变量(Condition)

条件变量允许进程等待特定条件发生后再继续执行,适用于需要进程间协作的场景。

from multiprocessing import Process, Condition
import time
def worker(cond, num):
    with cond:
        cond.wait()
        print(f"Worker {num}")
if __name__ == "__main__":
    condition = Condition()
    processes = [Process(target=worker, args=(condition, i)) for i in range(5)]
    for p in processes:
        p.start()
    time.sleep(2)
    with condition:
        condition.notify_all()
    for p in processes:
        p.join()

五、综合详细的例子

下面是一个综合详细的例子,模拟一个多进程的网页爬虫,使用进程池来管理多个爬虫进程,并确保爬取的数据不会重复。

网页爬虫示例

import multiprocessing
import requests
from bs4 import BeautifulSoup
import time
class WebCrawler:
    def __init__(self, urls, num_processes):
        self.urls = urls
        self.num_processes = num_processes
        self.visited_urls = set()
        self.lock = multiprocessing.Lock()
    def fetch_url(self, url):
        try:
            response = requests.get(url)
            return response.text
        except requests.RequestException as e:
            print(f"Failed to fetch {url}: {e}")
            return None
    def parse_html(self, html):
        soup = BeautifulSoup(html, 'html.parser')
        return [a['href'] for a in soup.find_all('a', href=True)]
    def crawl(self, url):
        if url in self.visited_urls:
            return []
        
        html = self.fetch_url(url)
        if html is None:
            return []
        new_urls = self.parse_html(html)
        with self.lock:
            self.visited_urls.add(url)
        return new_urls
    def start_crawling(self):
        with multiprocessing.Pool(self.num_processes) as pool:
            new_urls = pool.map(self.crawl, self.urls)
        return new_urls
if __name__ == "__main__":
    start_time = time.time()
    urls = [
        'https://example.com',
        'https://example.org',
        'https://example.net'
    ]
    crawler = WebCrawler(urls, num_processes=4)
    new_urls = crawler.start_crawling()
    print("New URLs:", new_urls)
    print("Crawling completed in", time.time() - start_time, "seconds")

运行结果

New URLs: [['https://example.com/page1', 'https://example.com/page2'], [], []]
Crawling completed in 2.5 seconds

六、总结

本文详细介绍了Python的multiprocessing模块,包括进程的创建、进程间通信、进程池的使用、进程同步等,并通过多个示例展示了如何在实际项目中应用这些技术。通过学习这些内容,您应该能够熟练掌握Python中的多进程编程,提高编写并发程序的能力。


多进程编程可以显著提高程序的并行计算性能,但也带来了新的挑战和问题。在使用多进程时,需要注意避免数据竞争、合理使用同步机制,并尽量通过进程池来管理和控制进程。


作者:Rjdeng

链接:https://juejin.cn/post/7397622646631251977

相关文章
|
4天前
|
Java Serverless Python
探索Python中的并发编程与`concurrent.futures`模块
探索Python中的并发编程与`concurrent.futures`模块
11 4
|
16天前
|
Python
如何在 Python 中导入模块
【8月更文挑战第29天】
19 1
|
16天前
|
Python
|
16天前
|
开发者 Python
什么是 python 模块?
【8月更文挑战第29天】
9 0
|
3月前
|
监控 Linux 应用服务中间件
探索Linux中的`ps`命令:进程监控与分析的利器
探索Linux中的`ps`命令:进程监控与分析的利器
|
2月前
|
运维 关系型数据库 MySQL
掌握taskset:优化你的Linux进程,提升系统性能
在多核处理器成为现代计算标准的今天,运维人员和性能调优人员面临着如何有效利用这些处理能力的挑战。优化进程运行的位置不仅可以提高性能,还能更好地管理和分配系统资源。 其中,taskset命令是一个强大的工具,它允许管理员将进程绑定到特定的CPU核心,减少上下文切换的开销,从而提升整体效率。
掌握taskset:优化你的Linux进程,提升系统性能
|
2月前
|
弹性计算 Linux 区块链
Linux系统CPU异常占用(minerd 、tplink等挖矿进程)
Linux系统CPU异常占用(minerd 、tplink等挖矿进程)
61 4
Linux系统CPU异常占用(minerd 、tplink等挖矿进程)
|
1月前
|
算法 Linux 调度
探索进程调度:Linux内核中的完全公平调度器
【8月更文挑战第2天】在操作系统的心脏——内核中,进程调度算法扮演着至关重要的角色。本文将深入探讨Linux内核中的完全公平调度器(Completely Fair Scheduler, CFS),一个旨在提供公平时间分配给所有进程的调度器。我们将通过代码示例,理解CFS如何管理运行队列、选择下一个运行进程以及如何对实时负载进行响应。文章将揭示CFS的设计哲学,并展示其如何在现代多任务计算环境中实现高效的资源分配。
|
2月前
|
存储 缓存 安全
【Linux】冯诺依曼体系结构与操作系统及其进程
【Linux】冯诺依曼体系结构与操作系统及其进程
138 1
|
2月前
|
小程序 Linux
【编程小实验】利用Linux fork()与文件I/O:父进程与子进程协同实现高效cp命令(前半文件与后半文件并行复制)
这个小程序是在文件IO的基础上去结合父子进程的一个使用,利用父子进程相互独立的特点实现对数据不同的操作