feat: 专题图克里金插值支持多线程并发

- ThreadPoolExecutor 替代串行 for 循环,默认 2 线程并发
- 克里金内部 numpy/scipy 运算释放 GIL,线程并行有效
- 通过 kriging_workers 配置项控制并发数(默认 2,设为 1 回退串行)
- 主线程预先设置 matplotlib Agg 后端,避免多线程竞争
- 保留串行路径作为 fallback(单 CSV 或 workers=1 时)
This commit is contained in:
duxin
2026-07-28 16:41:41 +08:00
parent 3c7e735342
commit e8a81008fa

View File

@ -243,32 +243,79 @@ class Step11MapHandler(BaseStepHandler):
context.notify('step11_map', 'warning',
f'共享上下文预计算失败: {e},回退逐个处理')
# ── 顺序生成(避免 Windows spawn 下 ProcessPoolExecutor 死锁)──
# 局部 Kriging 内部已做 16 块顺序分块,每块 ~20-30s
# 每张图约 5-8 分钟。64 张 ≈ 5-8 小时,但进度完全透明可见
# ── 并发生成 ──
# 克里金插值内部为 numpy/scipy 运算(释放 GIL
# 使用 ThreadPoolExecutor 并发处理多个 CSV大幅缩短总耗时
# 注意:不使用 ProcessPoolExecutorWindows spawn 会导致死锁)。
_max_workers = int(config.get('kriging_workers', 2))
_max_workers = max(1, min(_max_workers, total, os.cpu_count() or 4))
generated: List[str] = []
errors: Dict[str, str] = {}
context.notify('step11_map', 'info',
f'顺序生成 {total} 张专题图(局部 Kriging 自适应分块)')
for idx, csv_p in enumerate(csv_paths):
percent = int(idx / total * 100)
context.notify('step11_map', 'info',
f'专题图 [{idx+1}/{total}]: {Path(csv_p).name}')
global_event_bus.publish('ProgressUpdate', {
'percentage': percent,
'message': f'Step11: {idx+1}/{total} {Path(csv_p).stem}',
})
if _max_workers > 1 and total > 1:
# ★ 主线程预先设置 matplotlib Agg 后端(避免多线程竞争)
import matplotlib
try:
result_path, _ = _process_one_map(csv_p, base_kwargs, output_dir)
generated.append(result_path)
except Exception as e:
errors[csv_p] = str(e)
context.notify('step11_map', 'warning',
f'专题图 FAIL: {Path(csv_p).name}{e}')
matplotlib.use('Agg', force=True)
except Exception:
pass
import concurrent.futures
context.notify('step11_map', 'info',
f'并发生成 {total} 张专题图({_max_workers} 线程并行)')
completed = 0
with concurrent.futures.ThreadPoolExecutor(
max_workers=_max_workers) as executor:
future_map = {
executor.submit(
_process_one_map, csv_p, base_kwargs, output_dir
): csv_p
for csv_p in csv_paths
}
for future in concurrent.futures.as_completed(future_map):
csv_p = future_map[future]
completed += 1
try:
result_path, _ = future.result()
generated.append(result_path)
context.notify('step11_map', 'info',
f'专题图 [{completed}/{total}] ✓: '
f'{Path(csv_p).name}')
except Exception as e:
errors[csv_p] = str(e)
context.notify('step11_map', 'warning',
f'专题图 [{completed}/{total}] ✗: '
f'{Path(csv_p).name}{e}')
percent = int(completed / total * 100)
global_event_bus.publish('ProgressUpdate', {
'percentage': percent,
'message': f'Step11: {completed}/{total} '
f'{Path(csv_p).stem}',
})
else:
context.notify('step11_map', 'info',
f'顺序生成 {total} 张专题图(局部 Kriging 自适应分块)')
for idx, csv_p in enumerate(csv_paths):
percent = int(idx / total * 100)
context.notify('step11_map', 'info',
f'专题图 [{idx+1}/{total}]: {Path(csv_p).name}')
global_event_bus.publish('ProgressUpdate', {
'percentage': percent,
'message': f'Step11: {idx+1}/{total} {Path(csv_p).stem}',
})
try:
result_path, _ = _process_one_map(csv_p, base_kwargs, output_dir)
generated.append(result_path)
except Exception as e:
errors[csv_p] = str(e)
context.notify('step11_map', 'warning',
f'专题图 FAIL: {Path(csv_p).name}{e}')
step_end_time = time.time()
elapsed = step_end_time - step_start_time