如何限制ThreadPoolExecutor的队列最大容量？技术咨询

嘿，这个问题我太熟了！你完全不用折腾queue.Queue和threading那套底层工具，用ThreadPoolExecutor本身加上一点点小技巧就能轻松限制队列大小，避免一次性生成几百万个future对象。给你分享两个实用的方案：

这种方法会先提交一批任务，之后每有一个任务完成就立刻补上一个新任务，让系统中同时存在的任务数始终保持在你设定的阈值内，既不会闲置线程，也不会一次性创建过多future。

from concurrent.futures import ThreadPoolExecutor, as_completed

def some_func(arg):
    # 这里放你的密集型任务逻辑
    # 比如模拟耗时操作：import time; time.sleep(0.01)
    return arg * 2

def main():
    max_workers = 4  # 线程池大小
    batch_size = max_workers * 2  # 控制同时存在的任务数，可根据机器性能调整
    total_args = range(10000000)
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # 初始化第一批任务
        futures = {executor.submit(some_func, arg): arg for arg in total_args[:batch_size]}
        remaining_args = list(total_args[batch_size:])
        
        while futures:
            # 等待任意一个任务完成
            for completed_future in as_completed(futures):
                # 可以在这里处理任务结果
                # result = completed_future.result()
                
                # 移除已完成的任务
                del futures[completed_future]
                
                # 提交下一个任务（如果还有剩余）
                if remaining_args:
                    next_arg = remaining_args.pop(0)
                    new_future = executor.submit(some_func, next_arg)
                    futures[new_future] = next_arg

这种方式的好处是能实时处理每个完成的任务结果，而且对系统资源的占用非常平稳。

如果你的需求只是简单限制队列大小，不想手动维护任务池，用threading.Semaphore来做流量控制是更简洁的选择：

from concurrent.futures import ThreadPoolExecutor
import threading

def some_func(arg):
    # 你的密集型任务逻辑
    return arg * 2

def main():
    max_workers = 4
    max_pending_tasks = max_workers * 2  # 限制待处理+执行中任务的总数
    semaphore = threading.Semaphore(max_pending_tasks)
    
    # 包装任务函数，确保任务完成后释放信号量
    def wrapped_task(arg):
        try:
            return some_func(arg)
        finally:
            semaphore.release()
    
    total_args = range(10000000)
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = []
        for arg in total_args:
            semaphore.acquire()  # 超过限制时会阻塞，直到有任务完成释放信号量
            future = executor.submit(wrapped_task, arg)
            futures.append(future)
        
        # 后续可以遍历futures获取所有结果
        # for future in futures:
        #     print(future.result())

信号量就像一个“通行证”，每次提交任务前要先拿到通行证，任务完成后归还，这样就能保证同时在队列里的任务数不会超过你设定的max_pending_tasks。

这两种方案都不需要切换到底层的线程队列实现，完全基于concurrent.futures就能搞定，选哪种看你是否需要实时处理任务结果啦～

内容的提问来源于stack exchange，提问作者Bob