API测试脚本在EC2上成功率远低于本地的排查及优化方案咨询

核心问题分析 #

本地与EC2的成功率差异，核心原因集中在网络延迟/API异步同步差、中间层缓存、EC2网络环境限制三类，针对性优化即可解决。

具体优化措施 #

1. 实现POST后的验证重试机制（解决异步同步延迟） #

EC2到API服务器的网络延迟通常高于本地，API写操作可能是异步落地的，单次GET大概率拿到旧值。改用循环验证+指数退避等待的方式：

def verify_support_setup_steps(api_url, expected_value, max_retries=5):
    delay = 1
    for _ in range(max_retries):
        current_value = fetch_support_setup_steps(api_url)
        if current_value == expected_value:
            return True
        time.sleep(delay)
        delay *= 1.5  # 指数增加等待时间，避免频繁请求
    return False

在原脚本POST成功后，调用该函数替代单次GET：

if data:
    # 验证变更是否生效
    if verify_support_setup_steps(api_url_fetch_support_setup_steps, support_setup_steps_value):
        successful_calls += 1
    else:
        failures_calls += 1
        print(f"同步失败: 预期{support_setup_steps_value}，实际{fetch_support_setup_steps(api_url_fetch_support_setup_steps)}")

2. 禁用GET请求缓存（解决中间缓存问题） #

EC2所在VPC或中间代理可能缓存GET响应，强制添加缓存控制头绕过缓存：
修改fetch_support_setup_steps函数：

def fetch_support_setup_steps(api_url):
    try:
        headers = {
            "Cache-Control": "no-cache",
            "Pragma": "no-cache"
        }
        response = requests.get(api_url, headers=headers, timeout=10)
        response.raise_for_status()
        data = response.json()
        return data['payload']['supportSetupSteps']
    except requests.exceptions.RequestException as e:
        error_msg = f"GET请求错误: {e}"
        if 'response' in locals():
            error_msg += f", 状态码: {response.status_code}, 响应内容: {response.text[:200]}"
        print(error_msg)
        return None

3. 给所有请求添加超时设置（避免网络挂起） #

原脚本未设置超时，EC2网络波动时请求会无限等待导致失败，给POST和GET都添加超时限制：
修改set_support_setup_steps中的POST请求：

response = requests.post(api_url, json=payload, headers=headers, timeout=10)

4. 增强错误日志定位问题 #

原日志无法区分是POST失败还是GET同步失败，细化日志便于排查：
修改set_support_setup_steps的异常处理：

except requests.exceptions.RequestException as e:
    error_msg = f"POST请求错误: {e}"
    if 'response' in locals():
        error_msg += f", 状态码: {response.status_code}, 响应内容: {response.text[:200]}"
    print(error_msg)
    return None

5. 优化EC2网络环境 #

• 检查EC2安全组/网络ACL是否允许访问API的目标端口（如443/80）
• 选择与API服务器同区域的EC2实例，降低跨区域网络延迟
• 若使用NAT网关，检查是否存在带宽瓶颈或连接数限制

修改后的完整脚本示例 #

整合所有优化后的脚本：

import requests
import time

access_token = ""
api_url_fetch_support_setup_steps = ""
api_url_set_support_setup_steps = ""

def set_support_setup_steps(api_url, payload, access_token):
    try:
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {access_token}"
        }
        response = requests.post(api_url, json=payload, headers=headers, timeout=10)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        error_msg = f"POST错误: {e}"
        if 'response' in locals():
            error_msg += f", 状态码: {response.status_code}, 响应: {response.text[:200]}"
        print(error_msg)
        return None

def fetch_support_setup_steps(api_url):
    try:
        headers = {"Cache-Control": "no-cache", "Pragma": "no-cache"}
        response = requests.get(api_url, headers=headers, timeout=10)
        response.raise_for_status()
        data = response.json()
        return data['payload']['supportSetupSteps']
    except requests.exceptions.RequestException as e:
        error_msg = f"GET错误: {e}"
        if 'response' in locals():
            error_msg += f", 状态码: {response.status_code}, 响应: {response.text[:200]}"
        print(error_msg)
        return None

def verify_support_setup_steps(api_url, expected_value, max_retries=5):
    delay = 1
    for _ in range(max_retries):
        current_val = fetch_support_setup_steps(api_url)
        if current_val == expected_value:
            return True
        time.sleep(delay)
        delay *= 1.5
    return False

payload = {"supportSetupSteps": True}
cnt = 0
successful_calls = 0
failures_calls = 0
N = 100

while cnt < N:
    support_setup_steps_value = cnt % 2 == 0
    payload['supportSetupSteps'] = support_setup_steps_value
    print(f"迭代次数: {cnt}")
    cnt += 1
    
    data = set_support_setup_steps(api_url_set_support_setup_steps, payload, access_token)
    if not data:
        print(f"设置supportSetupSteps为{support_setup_steps_value}失败")
    else:
        if verify_support_setup_steps(api_url_fetch_support_setup_steps, support_setup_steps_value):
            successful_calls += 1
        else:
            failures_calls += 1
            print(f"同步失败: 预期{support_setup_steps_value}，实际{fetch_support_setup_steps(api_url_fetch_support_setup_steps)}")

with open("/home/ubuntu/cache/new_file.txt", "w") as file:
    file.write("详情\n")
    file.write(f"成功调用总数: {successful_calls}\n")
    file.write(f"同步失败次数: {failures_calls}\n")

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