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

wen python案例 2026-06-27 30

本文目录导读：

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()