并行下载THREDDS数据服务器文件失败，串行正常求助

问题说明 #

从UCAR THREDDS服务器批量提取格网参数时，串行执行代码可正常运行，但使用ThreadPoolExecutor并行执行时触发如下连接错误：

ConnectionError: ('Connection aborted.', RemoteDisconnected('Remote end closed connection without response'))

需要找到比串行更快的可靠下载方式。

错误原因 #

1. 服务器并发限制：公共数据服务器通常会限制单IP的并发连接数，过多并行请求会触发服务器断开连接。
2. 会话线程不安全：requests的Session对象并非线程安全，多线程共享会话可能导致连接异常。
3. 缺少重试机制：临时网络波动或服务器繁忙会导致连接中断，无重试逻辑则直接失败。

解决方案 #

1. 限制并发数 + 添加请求重试 #

降低线程数避免触发服务器限流，同时给请求添加重试逻辑处理临时连接问题：

import pandas as pd
import xarray as xr
import concurrent.futures
from functools import partial
from tqdm import tqdm
from siphon.catalog import TDSCatalog
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import requests

# 创建带重试策略的会话
def create_retry_session():
    session = requests.Session()
    retry = Retry(
        total=3,          # 最大重试次数
        backoff_factor=1, # 重试间隔递增系数
        status_forcelist=[429, 500, 502, 503, 504], # 需要重试的状态码
        allowed_methods=["GET"]
    )
    adapter = HTTPAdapter(max_retries=retry)
    session.mount("https://", adapter)
    session.mount("http://", adapter)
    return session

# 配置参数
bounds = [-10,-10,10,10]
cat_url = 'https://thredds.rda.ucar.edu/thredds/catalog/files/g/ds633.0/e5.oper.an.pl/200001/catalog.xml'
dataset_names = ['e5.oper.an.pl.128_131_u.ll025uv.2000010100_2000010123.nc',
                 'e5.oper.an.pl.128_132_v.ll025uv.2000010100_2000010123.nc',
                 'e5.oper.an.pl.128_130_t.ll025sc.2000010100_2000010123.nc',
                 'e5.oper.an.pl.128_133_q.ll025sc.2000010100_2000010123.nc',
                 'e5.oper.an.pl.128_129_z.ll025sc.2000010100_2000010123.nc']
var_strs = ['U', 'V', 'T', 'Q', 'Z']
start_date = pd.to_datetime('2000-01-01 00:00:00')
end_date = pd.to_datetime('2000-01-02 00:00:00')

def pull_data_from_url(dataset_name, var, cat_url, bounds, start_date, end_date):
    session = create_retry_session()
    # 用重试会话加载Catalog
    catalog = TDSCatalog(cat_url, session=session)
    ds = catalog.datasets[dataset_name]
    
    ncss = ds.subset()
    # 给NCSS绑定重试会话
    ncss._session = session
    
    query = ncss.query()
    query.lonlat_box(east=bounds[2], west=bounds[0], south=bounds[1], north=bounds[3])
    query.time_range(start_date, end_date)
    query.variables(var)
    
    nc = ncss.get_data(query)
    return xr.open_dataset(xr.backends.NetCDF4DataStore(nc))

# 限制线程数为2-3，避免触发服务器限流
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
    partial_func = partial(pull_data_from_url, cat_url=cat_url, bounds=bounds, start_date=start_date, end_date=end_date)
    results = list(tqdm(executor.map(partial_func, dataset_names, var_strs), total=len(dataset_names)))

2. 改用进程池隔离请求环境 #

如果线程池的会话共享问题无法解决，改用ProcessPoolExecutor隔离每个请求的运行环境：

# 其他代码与上述一致，仅修改执行部分
with concurrent.futures.ProcessPoolExecutor(max_workers=2) as executor:
    partial_func = partial(pull_data_from_url, cat_url=cat_url, bounds=bounds, start_date=start_date, end_date=end_date)
    results = list(tqdm(executor.map(partial_func, dataset_names, var_strs), total=len(dataset_names)))

3. 预加载Catalog减少重复请求 #

原代码每个线程都重新请求Catalog，浪费资源且增加服务器压力，可提前加载一次Catalog复用：

# 提前加载Catalog
catalog = TDSCatalog(cat_url)

def pull_data_from_url(dataset_name, var, catalog, bounds, start_date, end_date):
    session = create_retry_session()
    ds = catalog.datasets[dataset_name]
    
    ncss = ds.subset()
    ncss._session = session
    
    query = ncss.query()
    query.lonlat_box(east=bounds[2], west=bounds[0], south=bounds[1], north=bounds[3])
    query.time_range(start_date, end_date)
    query.variables(var)
    
    nc = ncss.get_data(query)
    return xr.open_dataset(xr.backends.NetCDF4DataStore(nc))

with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
    partial_func = partial(pull_data_from_url, catalog=catalog, bounds=bounds, start_date=start_date, end_date=end_date)
    results = list(tqdm(executor.map(partial_func, dataset_names, var_strs), total=len(dataset_names)))

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