Python中ThreadPoolExecutor如何实现任务公平轮询分配以提升并发效率？

默认的concurrent.futures.ThreadPoolExecutor基于线程空闲度分配任务：线程完成当前任务后，从任务队列头部取下一个任务执行。这种调度逻辑会导致任务分配不均——若任务执行耗时差异大，可能出现部分线程持续忙碌、部分线程提前闲置的情况，拉低整体效率。

要实现轮询式的任务均分，确保每个线程依次获取任务、负载均衡，可以采用以下两种方案：

方案一：预轮询分组任务（推荐，实现简单） #

核心思路是提前将任务按线程数量轮询分组，每个线程固定处理一组任务。这种方式能保证任务被均匀分配，从根源避免线程闲置。

代码示例 #

import concurrent.futures

def is_prime(n):
    if n <= 1:
        return False
    for i in range(2, int(n**0.5) + 1):
        if n % i == 0:
            return False
    return True

def process_task_group(task_group):
    """处理一组轮询分配的任务"""
    return [is_prime(num) for num in task_group]

if __name__ == "__main__":
    max_workers = 4
    h_range = 100000  # 示例任务范围

    # 按轮询规则拆分任务组
    task_groups = [[] for _ in range(max_workers)]
    for idx, number in enumerate(range(h_range + 1)):
        # 索引取模，实现轮询分配
        task_groups[idx % max_workers].append(number)

    # 提交分组任务到线程池
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = [executor.submit(process_task_group, group) for group in task_groups]
        
        # 收集所有结果
        all_results = []
        for future in concurrent.futures.as_completed(futures):
            all_results.extend(future.result())

这种方式的优势：

• 基于原生ThreadPoolExecutor，无需自定义线程池，兼容性好
• 任务分配完全可控，每个线程的任务量和类型分布均匀，避免闲置

方案二：自定义轮询调度线程池 #

如果需要更灵活的实时轮询调度，可以自己实现一个简单的线程池，给每个线程绑定独立任务队列，提交任务时按轮询顺序放入对应队列。

代码示例 #

import threading
from queue import Queue

class RoundRobinThreadPool:
    def __init__(self, max_workers):
        self.max_workers = max_workers
        self.task_queues = [Queue() for _ in range(max_workers)]
        self._worker_idx = 0
        self._start_workers()

    def _start_workers(self):
        """启动所有工作线程"""
        for idx in range(self.max_workers):
            thread = threading.Thread(target=self._worker_loop, args=(idx,))
            thread.daemon = True
            thread.start()

    def _worker_loop(self, thread_idx):
        """工作线程循环处理任务"""
        queue = self.task_queues[thread_idx]
        while True:
            func, args, kwargs = queue.get()
            try:
                func(*args, **kwargs)
            except Exception as e:
                print(f"Thread {thread_idx}执行出错: {str(e)}")
            finally:
                queue.task_done()

    def submit(self, func, *args, **kwargs):
        """轮询提交任务到不同队列"""
        self.task_queues[self._worker_idx % self.max_workers].put((func, args, kwargs))
        self._worker_idx += 1

    def wait_for_completion(self):
        """等待所有任务完成"""
        for queue in self.task_queues:
            queue.join()

# 使用示例
def is_prime(n):
    if n <= 1:
        return False
    for i in range(2, int(n**0.5) + 1):
        if n % i == 0:
            return False
    return True

if __name__ == "__main__":
    max_workers = 4
    h_range = 100000

    pool = RoundRobinThreadPool(max_workers)
    for number in range(h_range + 1):
        pool.submit(is_prime, number)
    
    pool.wait_for_completion()

注意事项 #

自定义线程池需要自行处理结果回调、异常捕获等逻辑，适合对调度逻辑有特殊需求的场景。

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