本文介绍在 synapse notebook 等环境中，通过分批 + 线程池方式精准限制对 azure function 的并发调用数（如每批 200 个），避免因瞬时请求过载导致函数崩溃或限流失败。
本文介绍在 synapse notebook 等环境中，通过分批 + 线程池方式精准限制对 azure function 的并发调用数（如每批 200 个），避免因瞬时请求过载导致函数崩溃或限流失败。
Azure Function 默认具有严格的并发与资源限制（例如 Consumption Plan 下典型并发上限约 200 个实例），若客户端未加节制地发起大量并发请求（如使用 ThreadPoolExecutor 直接全量提交），极易触发 HTTP 429（Too Many Requests）、503（Service Unavailable）或冷启动超时，最终导致函数不可用或数据丢失。
关键误区在于：线程池（ThreadPoolExecutor）本身不提供跨批次的流量节制能力。您原代码中 max_workers=5 仅限制本地线程数，但每个线程仍会立即发起一个独立 HTTP 请求——若 array_of_objects 含数千项，实际并发请求数仍可能远超 Azure Function 承受阈值。
✅ 正确方案是采用 “分批串行 + 批内并发” 的双层控制策略：
- 外层逻辑：将数据切分为固定大小批次（如 batch_size = 200），严格按顺序执行；
- 内层并发：每批内部启用线程池（如 max_workers=10），在单批约束下提升吞吐效率；
- 显式等待：确保当前批全部完成（含错误处理）后，再进入下一批。
以下是完整、健壮的实现示例（适配 Synapse Notebook）：
SkillSub Pro - Python 题解与代码注释双功能技能功能概述SkillSub Pro - Python 题解与代码注释双功能技能是一项面向实际任务的技能，主要用于SkillSub Pro 是一个 Python 题解生成与代码注释的 双功能合体技能 ，专为学生、算法学习者和开发者设计；✅ 一个技能，两种用途 ：；核心要点📝 题解模式 ：输入题目/题号，自动生成完整 Python 题解（含详细注释、解题思路、复杂度分析）；💬 注释模式 ：输入 Python 代码，自动添加详细中。它将相关步骤、
立即学习“Python免费学习笔记（深入）”；
import time
import requests
import concurrent.futures
from typing import List, Dict, Any
# 配置参数（根据实际调整）
AZURE_FUNCTION_URL = "https://your-function-app.azurewebsites.net/api/your-endpoint?code=xxx"
BATCH_SIZE = 200
BATCH_CONCURRENCY = 10 # 每批最多并行 10 个请求，避免单批压垮函数
RETRY_DELAY_SEC = 1 # 请求失败后重试前等待时间
def call_azure_function(payload: Dict[str, Any]) -> Dict[str, Any]:
"""封装单次 Azure Function 调用，含基础错误处理"""
try:
response = requests.post(
AZURE_FUNCTION_URL,
json=payload,
timeout=30 # 防止长阻塞
)
response.raise_for_status() # 抛出 4xx/5xx 异常
return {"success": True, "status": response.status_code, "data": response.json()}
except requests.exceptions.RequestException as e:
return {"success": False, "error": str(e), "status": getattr(e.response, 'status_code', 'N/A')}
def process_batch(batch: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""处理单个批次：并发调用 + 统一结果收集"""
results = []
with concurrent.futures.ThreadPoolExecutor(max_workers=BATCH_CONCURRENCY) as executor:
# 提交所有任务
future_to_payload = {executor.submit(call_azure_function, item): item for item in batch}
# 按完成顺序收集结果（可选：改为 as_completed 以实时反馈）
for future in concurrent.futures.as_completed(future_to_payload):
payload = future_to_payload[future]
try:
result = future.result()
results.append(result)
except Exception as e:
results.append({"success": False, "error": f"Future exception: {str(e)}", "payload": payload})
return results
# 主执行流程
array_of_objects = [...] # 替换为您的实际数据列表（每个元素为 dict）
print(f"Total items: {len(array_of_objects)}, Batch size: {BATCH_SIZE}")
for i in range(0, len(array_of_objects), BATCH_SIZE):
current_batch = array_of_objects[i:i + BATCH_SIZE]
batch_num = i // BATCH_SIZE + 1
print(f"\n--- Processing Batch {batch_num} ({len(current_batch)} items) ---")
start_time = time.time()
batch_results = process_batch(current_batch)
elapsed = time.time() - start_time
# 统计本批结果
success_count = sum(1 for r in batch_results if r.get("success"))
print(f"Batch {batch_num} completed in {elapsed:.2f}s: {success_count}/{len(current_batch)} succeeded")
# 可选：检查失败项并记录（便于后续重试）
failed_items = [r for r in batch_results if not r.get("success")]
if failed_items:
print(f"⚠️ {len(failed_items)} failures in batch {batch_num}. First error: {failed_items[0].get('error', 'Unknown')}")
# 强制批次间间隔（防止连续高压，可选）
if batch_num < (len(array_of_objects) + BATCH_SIZE - 1) // BATCH_SIZE:
time.sleep(0.5) # 轻量缓冲，避免下一批立即触发
print("\n✅ All batches processed.")? 关键注意事项：
- 不要依赖 time.sleep() 模拟限流：它无法解决并发竞争问题，且易受系统调度影响；
- 务必设置 timeout：Azure Function 执行时间有限（Consumption Plan 默认 10 分钟），超时需主动中断；
- 启用重试机制（生产环境建议）：可在 call_azure_function 中集成 tenacity 库实现指数退避重试；
- 监控与日志：在 Synapse 中建议将 batch_results 写入临时表或 Log Analytics，便于故障追溯；
- 函数端配合：确认 Azure Function 的 host.json 中 extensions.http.maxConcurrentRequests 和 maxOutstandingRequests 配置合理（尤其在 Premium/ASE 计划中）。
通过该模式，您既能充分利用 Azure Function 的弹性伸缩能力，又能严格保障调用稳定性——真正实现「可控并发、安全批量」。
