Python脚本如何控制进程最大数量

wen python案例 2026-06-27 295

本文目录导读：

Python脚本如何控制进程最大数量

使用 multiprocessing 模块的 Pool（推荐）
使用 Process 配合 semaphore
使用 concurrent.futures 模块
手动控制进程队列
使用资源限制（系统级）
实际应用示例：爬虫场景
推荐使用方式

在Python中控制进程最大数量主要有以下几种方法：

使用 multiprocessing 模块的 Pool（推荐）

这是最常用的方法,通过进程池限制并发进程数：

from multiprocessing import Pool
import time
def worker(task_id):
    print(f"进程 {task_id} 开始工作")
    time.sleep(2)
    return f"任务 {task_id} 完成"
# 限制最大进程数为3
with Pool(processes=3) as pool:
    results = pool.map(worker, range(10))
    print(results)

使用 Process 配合 semaphore

使用信号量来控制进程数量：

from multiprocessing import Process, Semaphore
import time
def worker(task_id, semaphore):
    with semaphore:  # 获取信号量，超过限制会阻塞
        print(f"进程 {task_id} 开始工作")
        time.sleep(2)
        print(f"进程 {task_id} 结束工作")
if __name__ == '__main__':
    max_processes = 3
    semaphore = Semaphore(max_processes)
    processes = []
    for i in range(10):
        p = Process(target=worker, args=(i, semaphore))
        p.start()
        processes.append(p)
    for p in processes:
        p.join()

使用 concurrent.futures 模块

高级API,使用方式更简洁：

from concurrent.futures import ProcessPoolExecutor
import time
def worker(task_id):
    print(f"进程 {task_id} 开始工作")
    time.sleep(2)
    return f"任务 {task_id} 完成"
# 限制最大进程数为3
with ProcessPoolExecutor(max_workers=3) as executor:
    futures = [executor.submit(worker, i) for i in range(10)]
    for future in futures:
        print(future.result())

手动控制进程队列

更精细的控制方式：

from multiprocessing import Process
import time
import queue
def worker(task_queue, max_processes):
    while True:
        try:
            task = task_queue.get_nowait()
            print(f"处理任务: {task}")
            time.sleep(2)
        except queue.Empty:
            break
if __name__ == '__main__':
    max_processes = 3
    task_queue = queue.Queue()
    # 添加任务
    for i in range(10):
        task_queue.put(f"task_{i}")
    # 启动控制进程数量的工作
    processes = []
    for _ in range(min(max_processes, task_queue.qsize())):
        p = Process(target=worker, args=(task_queue, max_processes))
        p.start()
        processes.append(p)
    for p in processes:
        p.join()

使用资源限制（系统级）

在某些操作系统上,可以限制进程的全局数量：

import resource
import multiprocessing
# 设置进程数量限制（Linux系统）
def set_process_limit():
    try:
        resource.setrlimit(resource.RLIMIT_NPROC, (5, 5))
    except Exception as e:
        print(f"设置资源限制失败: {e}")
# 使用限制
set_process_limit()

实际应用示例：爬虫场景

from multiprocessing import Pool
import requests
import time
def fetch_url(url):
    try:
        response = requests.get(url, timeout=5)
        print(f"成功获取: {url}, 状态码: {response.status_code}")
        return response.status_code
    except Exception as e:
        print(f"获取失败: {url}, 错误: {e}")
        return None
def main():
    urls = [
        "http://example.com",
        "http://example.org",
        # ... 更多URL
    ]
    # 限制最多5个并发进程
    max_processes = 5
    with Pool(processes=max_processes) as pool:
        results = pool.map(fetch_url, urls)
    print(f"完成处理 {len(urls)} 个URL")
if __name__ == '__main__':
    main()