如何结合ThreadPoolExecutor与BFS实现高效H3单元格API调用？

我尝试用ThreadPoolExecutor加速H3单元格的API调用：类接收逗号分隔的H3单元格字符串，调用API返回结果总数；由于API分页最多返回1000条，我通过H3逐级拆分单元格（广度优先搜索逻辑），直到所有子单元格的结果总数低于1000。

当前问题：

• 运行run方法后，search_queue仍有未处理的子单元格，仅处理了初始传入的单元格
• 交换while与ThreadPoolExecutor的代码位置后，能处理所有单元格，但速度和单线程一致，没起到多线程加速效果

1. 线程安全问题：原代码用普通列表作为search_queue，多线程下pop(0)和extend操作存在竞争条件，可能导致队列状态混乱、任务丢失或重复处理
2. 任务提交逻辑缺陷：主线程的while循环仅在ThreadPoolExecutor块内提交初始队列的任务，当工作线程处理单元格并新增子任务到队列时，主线程可能已经因为当时队列为空而退出循环，不再提交新任务
3. 无等待逻辑：原代码没有等待所有工作线程完成任务就结束，导致部分子任务未被处理

import h3
import math
import requests
from concurrent.futures import ThreadPoolExecutor
from time import sleep
import queue

dummy_results = {
    '85489e37fffffff': {'total': 1001},
    '85489e27fffffff': {'total': 999},
    '86489e347ffffff': {'total': 143},
    '86489e34fffffff': {'total': 143},
    '86489e357ffffff': {'total': 143},
    '86489e35fffffff': {'total': 143},
    '86489e367ffffff': {'total': 143},
    '86489e36fffffff': {'total': 143},
    '86489e377ffffff': {'total': 143},
}

class SearchH3Test(object):

    def __init__(self, origin_cells):
        # 替换为线程安全的队列
        self.search_queue = queue.Queue()
        self.params_list = []
        # 初始化队列
        for cell in filter(None, origin_cells.split(',')):
            self.search_queue.put(cell)
        
    def get_h3_radius(self, cell, buffer=False):
        """获取H3单元格的近似半径"""
        return math.ceil(
            math.sqrt(
                (h3.cell_area(cell))/(1.5*math.sqrt(3))
            )*1000
            + ((100*(h3.h3_get_resolution(cell)/10)) if buffer else 0)
        )
    
    def get_items(self, cell):
        """模拟API调用返回结果总数"""
        sleep(1)
        r = dummy_results[cell]
        return r['total']
    
    def get_hex_params(self, cell):
        """生成单元格对应的请求参数并获取结果总数"""
        lat, long = h3.h3_to_geo(cell)
        radius = self.get_h3_radius(cell, buffer=True)

        params = {
            'latitude': lat,
            'longitude': long,
            'radius': radius,
        }

        total = self.get_items(cell)
        print(total)

        return total, params
    
    def hex_search_worker(self):
        """工作线程：循环处理队列中的单元格"""
        while True:
            try:
                # 非阻塞获取队列元素，队列为空时抛出Empty异常
                cell = self.search_queue.get_nowait()
            except queue.Empty:
                break
            
            total, params = self.get_hex_params(cell)
            if total > 1000:
                # 将子单元格加入队列
                for child_cell in h3.h3_to_children(cell):
                    self.search_queue.put(child_cell)
            elif total > 0:
                # 列表append操作在GIL下是原子性的，线程安全
                self.params_list.append(params)
            
            # 标记当前任务处理完成
            self.search_queue.task_done()
    
    def get_params_list(self):
        """启动多线程处理任务，等待所有任务完成"""
        # 根据API并发限制或CPU核心数设置线程数
        thread_count = 4
        with ThreadPoolExecutor(max_workers=thread_count) as e:
            # 启动指定数量的工作线程
            for _ in range(thread_count):
                e.submit(self.hex_search_worker)
        
        # 等待队列中所有任务（包括新增的子任务）处理完成
        self.search_queue.join()
    
    def run(self):
        self.get_params_list()

• 线程安全队列：用queue.Queue替代普通列表，其get/put操作自带线程锁，避免多线程竞争问题
• 生产者-消费者模式：工作线程主动从队列取任务，处理完后自动将子任务加入队列，无需主线程持续检查提交
• 等待任务完成：调用search_queue.join()确保所有任务（包括动态新增的子任务）都被处理完毕
• 线程数控制：通过max_workers设置合理的线程数，避免过多线程导致API限流或系统资源耗尽

h = SearchH3Test(
    '85489e37fffffff,85489e27fffffff',
)

h.run()
len(h.search_queue)  # 返回0，所有单元格已处理
len(h.params_list)   # 返回8，初始符合条件的1个+7个符合条件的子单元格

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