Python限制线程数处理DataFrame并获取驾驶时间结果的问题

你的问题核心出在手动线程管理的线程安全漏洞和异步任务结果无法同步回DataFrame这两个点上，下面一步步拆解问题并给出完整的解决方案：

问题根源分析 #

1. 线程数飙升至数千的原因 #

• 手动维护的threadCount没有线程安全保护：多个线程同时修改这个变量时会出现竞态条件（比如两个线程同时执行threadCount +=1，最终可能只加了1），导致计数完全不准确。
• Pandas的apply是同步遍历每一行，每一行都会调用closureDriveTimeCalc，而你写的while循环只是在当前调用的主线程里短暂等待，根本阻止不了apply继续处理下一行——瞬间就会启动数千个线程，完全突破了你想控制的3-5个线程上限。

2. 无法将结果返回至DataFrame的原因 #

drivetimeCalc是在子线程中执行的，而apply期望被调用的函数立刻返回值，子线程的异步结果根本无法传递回apply的返回值，所以你的DRIVETIME列全是None。

解决方案：用线程池替代手动线程管理 #

Python标准库的concurrent.futures.ThreadPoolExecutor可以完美解决这两个问题：它会自动帮你控制线程数量，还能同步收集任务结果，完全不需要手动维护线程计数。同时我们还要优化ArcGIS服务的调用逻辑，减少不必要的开销。

修改后的完整代码 #

import pandas as pd
import concurrent.futures
import arcgis

class GeoAnalysis:
    def __init__(self, username, pw):
        # 初始化ArcGIS连接和RouteLayer，只做一次！
        self.my_gis = arcgis.gis.GIS("https://www.arcgis.com", username, pw)
        route_service_url = self.my_gis.properties.helperServices.route.url
        self.route_layer = arcgis.network.RouteLayer(route_service_url, gis=self.my_gis)
    
    def drivetimeCalc(self, coordsString):
        try:
            # 解析坐标字符串（注意格式要和你的COORDS_COL完全匹配）
            points = coordsString.split(", ")
            # ArcGIS要求坐标格式是 经度,纬度;经度,纬度
            point_pair = f"{points[1]}, {points[0]}; {points[3]}, {points[2]}"
            # 调用路线服务
            result = self.route_layer.solve(
                stops=point_pair,
                return_directions=False,
                return_routes=True,
                output_lines='esriNAOutputLineNone',
                return_barriers=False,
                return_polygon_barriers=False,
                return_polyline_barriers=False
            )
            # 提取驾驶时间
            return result['routes']['features'][0]['attributes']['Total_TravelTime']
        except Exception as e:
            # 处理异常，避免单个任务失败导致整个进程崩溃
            print(f"计算失败（坐标：{coordsString}）：{str(e)}")
            return pd.NA  # 返回缺失值，方便后续处理

class MainFunction:
    def __init__(self, username, pw):
        self.ga = GeoAnalysis(username, pw)
    
    def driveTimeAnalysis(self, someFileName):
        # 读取CSV数据
        df = pd.read_csv(someFileName)
        # 用线程池控制并发数（这里设为5，你可以根据需求调整）
        max_workers = 5
        with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
            # 批量提交任务，按顺序收集结果（和原DataFrame行顺序一致）
            drivetimes = list(executor.map(self.ga.drivetimeCalc, df['COORDS_COL']))
        # 将结果赋值给新列
        df['DRIVETIME'] = drivetimes
        # 返回处理后的DataFrame（也可以直接保存为CSV）
        return df

# 使用示例
if __name__ == "__main__":
    main = MainFunction("你的用户名", "你的密码")
    result_df = main.driveTimeAnalysis("你的数据文件.csv")
    result_df.to_csv("带驾驶时间的结果.csv", index=False)

关键优化点说明 #

1. 线程池自动控容：ThreadPoolExecutor(max_workers=5)会确保同时运行的线程不超过5个，自动管理线程的创建、复用和销毁，完全避免手动计数的线程安全问题。
2. 复用ArcGIS资源：原来每次调用drivetimeCalc都创建新的RouteLayer，现在在GeoAnalysis初始化时只创建一次，大幅减少服务连接的开销。
3. 同步收集结果：executor.map会按输入的顺序返回每个任务的结果，直接转成列表就能对应到DataFrame的每一行，完美解决结果无法返回的问题。
4. 异常处理：添加try-except块，单个坐标计算失败不会影响整个任务，还会打印错误信息方便排查。

额外注意事项 #

• 确认ArcGIS服务的并发限制：如果5个线程触发了服务的请求频率限制，可以适当降低max_workers的值。
• 坐标格式校验：确保COORDS_COL的格式确实是lat_1,long_1,lat_2,long_2（注意分隔符是, 还是其他，代码里的split(", ")要和实际格式匹配）。

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