在处理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
