本文介绍在 synapse notebook 等环境中，通过分批 + 线程池方式精准限制对 azure function 的并发调用（如每批 200 个请求），避免因超载导致函数崩溃或限流失败。
本文介绍在 synapse notebook 等环境中，通过分批 + 线程池方式精准限制对 azure function 的并发调用（如每批 200 个请求），避免因超载导致函数崩溃或限流失败。
Azure Function 默认有严格的并发和资源限制（例如消费计划下典型并发上限约 200 个实例），若在客户端（如 Synapse Notebook）使用 ThreadPoolExecutor 无节制并发发送请求，极易触发 HTTP 429（Too Many Requests）、503（Service Unavailable）甚至导致函数冷启动失败或进程崩溃——这正是您遇到 time.sleep() 无效、请求持续涌出的根本原因：sleep 仅阻塞当前线程，而线程池中其他线程仍在独立发起请求。
✅ 正确做法是分批串行调度 + 批内并行执行：将全部数据切分为固定大小的批次（如每批 200 条），确保前一批所有请求完成后再启动下一批。这种方式既尊重 Azure Function 的承载能力，又充分利用了单批内的并发效率。
以下是推荐实现方案（已适配 Synapse Notebook 环境）：
import time
import requests
import concurrent.futures
# 配置项（请替换为您的实际值）
AZURE_FUNCTION_URL = "https://your-function-app.azurewebsites.net/api/your-trigger?code=xxx"
BATCH_SIZE = 200
TIMEOUT_SECONDS = 30 # 单次请求超时，防止挂起
def call_azure_function(payload):
"""封装单次 Azure Function 调用，含错误处理与日志"""
try:
response = requests.post(
AZURE_FUNCTION_URL,
json=payload,
timeout=TIMEOUT_SECONDS
)
response.raise_for_status() # 抛出 4xx/5xx 异常
return {"status": "success", "status_code": response.status_code}
except requests.exceptions.Timeout:
return {"status": "timeout", "error": "Request timed out"}
except requests.exceptions.RequestException as e:
return {"status": "failed", "error": str(e)}
def process_batch(batch_data):
"""同步执行单个批次：内部并行，外部阻塞等待全部完成"""
with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
# 提交所有任务
futures = [executor.submit(call_azure_function, item) for item in batch_data]
# 等待本批次全部完成
results = [future.result() for future in concurrent.futures.as_completed(futures)]
# 统计本批结果（可选）
success_count = sum(1 for r in results if r["status"] == "success")
print(f"✅ Batch completed: {len(batch_data)} items → {success_count} succeeded, {len(batch_data)-success_count} failed")
return results
# 主执行逻辑：分批串行调度
array_of_objects = [...] # 替换为您的实际数据列表（如字典数组）
print(f"Starting batched submission for {len(array_of_objects)} items (batch size = {BATCH_SIZE})")
start_time = time.time()
for i in range(0, len(array_of_objects), BATCH_SIZE):
current_batch = array_of_objects[i:i + BATCH_SIZE]
print(f"➡️ Processing batch {i//BATCH_SIZE + 1}: items {i+1}–{min(i+len(current_batch), len(array_of_objects))}")
process_batch(current_batch)
# 【关键】批次间强制暂停（可选，增强稳定性）
if i + BATCH_SIZE < len(array_of_objects): # 非末尾批次
print("⏳ Pausing 1 second before next batch...")
time.sleep(1)
end_time = time.time()
print(f"? All batches completed in {end_time - start_time:.2f} seconds")? 重要注意事项：
立即学习“Python免费学习笔记（深入）”；
- 不要在 readarray 中使用 time.sleep() 模拟延迟：它无法控制并发数，仅延长单线程耗时，反而降低吞吐且掩盖真实瓶颈；真正的节流必须在调度层（即批次之间）实现。
- max_workers 建议设为 5–15：过高会加剧客户端资源争抢（尤其 Synapse Notebook 内存受限），过低则浪费 Azure 函数的并发容量；根据函数响应时间实测调整。
- 务必添加 timeout 和异常捕获：Azure Function 可能因冷启动、网络抖动返回延迟或失败，无保护的请求会阻塞整个批次。
- 避免嵌套 ThreadPoolExecutor：原答案中 send_requests_to_azure 内部新建 executor 是正确做法，但需确保外层无 executor 包裹该调用（否则形成线程爆炸）。
- 进阶建议：生产环境可结合 retry 库实现指数退避重试，并将失败项写入 Blob 或 Delta 表供后续补发。
通过该模式，您既能严格遵守 Azure Function 的 200 并发阈值，又能保持高吞吐（每批内充分并发），同时保障执行稳定性和可观测性。
