Python中多进程编程如何实现？
需求：批量检测 url 有效性
我有一个文本中有 10w 条网址，需要检测所有 url 是否能打开，并将能打开的 url 保存到文本中，不了解 python 的多进程，网上看了一些文章，有些懵逼，求指点
原代码如下：
import requests
url_result_success = []
url_result_failed = []
headers = {
‘Accept’: ‘text/html,application/xhtml+xml,application/xml;q=0.9,/;q=0.8’,
‘Accept-Encoding’: ‘gzip, deflate, compress’,
‘Accept-Language’: ‘en-us;q=0.5,en;q=0.3’,
‘Cache-Control’: ‘max-age=0’,
‘Connection’: ‘keep-alive’,
‘User-Agent’: ‘Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:59.0) Gecko/20100101 Firefox/59.0’
}
with open(r’urls.txt’, ‘r’) as f:
for url in f:
url = url.strip()
print url
try:
response = requests.get(url, headers=headers, allow_redirects=True, timeout=5)
if response.status_code != 200:
raise requests.RequestException(u"Status code error: {}".format(response.status_code))
except requests.RequestException as e:
url_result_failed.append(url)
continue
url_result_success.append(url)
with open(r’valid_urls.txt’, ‘a+’) as f:
for url in url_result_success:
url = url.strip()
f.write(url + ‘\n’)
Python中多进程编程如何实现？
Python多进程编程主要用multiprocessing模块，它绕过了GIL限制，能真正利用多核CPU。核心是Process类，用法和threading类似但更简单。
看个具体例子，计算一个大列表的平方和：
import multiprocessing
import time
def compute_square(numbers, result, index):
"""子进程任务：计算部分数据的平方和"""
partial_sum = 0
for num in numbers:
partial_sum += num * num
result[index] = partial_sum
def main():
# 准备测试数据
data = list(range(1, 1000001))
# 将数据分成4份
chunk_size = len(data) // 4
chunks = [
data[i:i + chunk_size]
for i in range(0, len(data), chunk_size)
]
# 创建共享内存用于结果收集
with multiprocessing.Manager() as manager:
result = manager.list([0] * 4)
processes = []
# 创建并启动4个进程
start_time = time.time()
for i in range(4):
p = multiprocessing.Process(
target=compute_square,
args=(chunks[i], result, i)
)
processes.append(p)
p.start()
# 等待所有进程完成
for p in processes:
p.join()
total_sum = sum(result)
elapsed = time.time() - start_time
print(f"多进程计算结果: {total_sum}")
print(f"耗时: {elapsed:.2f}秒")
# 对比单进程版本
start_time = time.time()
single_result = sum(x*x for x in data)
single_elapsed = time.time() - start_time
print(f"\n单进程计算结果: {single_result}")
print(f"耗时: {single_elapsed:.2f}秒")
print(f"加速比: {single_elapsed/elapsed:.1f}倍")
if __name__ == "__main__":
main()
关键点说明：
Process类创建进程，target指定要执行的函数，args传参数start()启动进程，join()等待进程结束- 用
Manager()管理进程间通信，这里用了共享列表 - 数据分片处理是常见模式，避免进程间竞争
if __name__ == "__main__"在Windows下必须写，防止子进程重复执行
更现代的写法可以用Pool：
from multiprocessing import Pool
def square(x):
return x * x
with Pool(4) as pool:
results = pool.map(square, range(1000))
进程池适合任务均匀的场景，自动管理进程创建和任务分配。
多进程适合CPU密集型任务，进程间通信用Queue、Pipe或共享内存。
生产者 消费者模式的多线程就行,
用 queue 队列存取数据
生产者往 queue 里面存数据,
消费者里面用一个 while True 死循环 从 queue 里面取数据.
设置一个终止的毒丸.
