痛点：接口响应慢，但到底慢在哪里？

测试过程中，你有没有遇到过这些情况：

很多时候，”接口能通”和”接口快”是两回事。平均值容易掩盖问题——5 个请求 100ms 和 1 个请求 500ms 的平均值看起来都是 100ms，但用户体验天差地别。

今天分享的工具，就是为了解决这个问题：批量压测接口，自动计算 P50/P90/P99，自动出图表。

工具能做什么

功能
说明
批量接口测试
一次配置多个接口，自动发起 N 次请求
性能指标计算
响应时间 P50/P90/P95/P99，最大/最小/平均值
趋势图表生成
自动生成折线图，展示响应时间分布
异常率统计
超时请求数及占比
结果对比
支持与历史结果对比，发现性能回归
测试报告导出
生成 HTML 报告，可直接发邮件

完整代码

① 接口配置文件 apis.yaml

  1. # 接口监控配置
  2. apis:
  3. - name:"用户登录接口"
  4.     method: POST
  5.     url:"https://test.example.com/api/user/login"
  6.     headers:
  7. Content-Type:"application/json"
  8.     body:
  9.       username:"test_user"
  10.       password:"test_pass"
  11.     timeout:30

  13. - name:"获取用户信息"
  14.     method: GET
  15.     url:"https://test.example.com/api/user/info/10001"
  16.     headers:
  17. Authorization:"Bearer {{token}}"
  18.     timeout:15

  20. - name:"商品列表查询"
  21.     method: GET
  22.     url:"https://test.example.com/api/product/list?page=1&size=20"
  23.     timeout:15

  25. - name:"提交订单"
  26.     method: POST
  27.     url:"https://test.example.com/api/order/create"
  28.     headers:
  29. Content-Type:"application/json"
  30.     body:
  31.       user_id:10001
  32.       product_ids:[101,102]
  33.       total_amount:299.00
  34.     timeout:30

  36. # 测试参数
  37. settings:
  38.   requests_per_api:100# 每个接口跑多少次
  39.   concurrency:10# 并发数
  40.   think_time:0.1# 每次请求间隔（秒）
  41.   base_history_file:"history_results.json"# 历史数据文件（用于对比）

② 主脚本 api_monitor.py

  1. # -*- coding: utf-8 -*-
  2. """
  3. 接口性能监控工具
  4. 批量测试接口，统计 P50/P90/P99，生成图表和 HTML 报告
  5. 依赖：pip install requests pyyaml matplotlib numpy
  6. """
  7. import os
  8. import sys
  9. import json
  10. import time
  11. import datetime
  12. import argparse
  13. import statistics
  14. import requests
  15. import yaml
  16. import matplotlib
  17. matplotlib.use('Agg')# 无 GUI 环境也能生成图片
  18. import matplotlib.pyplot as plt
  19. import numpy as np
  20. plt.rcParams['font.sans-serif']=['SimHei','Microsoft YaHei']
  21. plt.rcParams['axes.unicode_minus']=False
  22. # ─────────────────────────────────────────────
  23. # 第一部分：请求执行器（支持并发）
  24. # ─────────────────────────────────────────────
  25. def send_request(name, api_config):
  26. """发起单次 HTTP 请求，返回响应时间（毫秒）和状态"""
  27.     start = time.time()
  28.     method = api_config.get('method','GET').upper()
  29.     url = api_config['url']
  30.     headers = api_config.get('headers',{})
  31.     body = api_config.get('body')
  32.     timeout = api_config.get('timeout',30)
  33. try:
  34. if method =='GET':
  35.             resp = requests.get(url, headers=headers, timeout=timeout, verify=False)
  36. elif method =='POST':
  37.             resp = requests.post(url, json=body, headers=headers, timeout=timeout, verify=False)
  38. elif method =='PUT':
  39.             resp = requests.put(url, json=body, headers=headers, timeout=timeout, verify=False)
  40. elif method =='DELETE':
  41.             resp = requests.delete(url, headers=headers, timeout=timeout, verify=False)
  42. else:
  43. returnNone,None, f"不支持的方法: {method}"
  44.         elapsed_ms = round((time.time()- start)*1000,2)
  45. return elapsed_ms, resp.status_code,None
  46. except requests.exceptions.Timeout:
  47. returnNone,None,"请求超时"
  48. except requests.exceptions.ConnectionError:
  49. returnNone,None,"连接失败"
  50. exceptExceptionas e:
  51. returnNone,None, str(e)
  52. def run_batch_test(name, api_config, total_requests=100, concurrency=10, think_time=0.1):
  53. """
  54.     批量执行接口测试
  55.     :param name: 接口名称
  56.     :param api_config: 接口配置
  57.     :param total_requests: 总请求次数
  58.     :param concurrency: 并发数（实际为串行+间隔，非真正并发）
  59.     :param think_time: 每次请求间隔
  60.     :return: 响应时间列表、错误信息列表
  61.     """
  62.     response_times =[]
  63.     errors =[]
  64. print(f"\n  📡 测试接口: {name}")
  65. print(f"     URL: {api_config['url']}")
  66. print(f"     总请求: {total_requests} | 并发: {concurrency}")
  67. for i in range(total_requests):
  68.         elapsed, status, error = send_request(name, api_config)
  69. if elapsed isnotNone:
  70.             response_times.append(elapsed)
  71.             bar_len = min(int(elapsed /5),40)# 简单可视化
  72.             bar ='█'* bar_len
  73. # 超2秒用红色标记
  74.             color_flag ='🔴'if elapsed >2000else('🟡'if elapsed >1000else'🟢')
  75. print(f"     [{i+1:3d}/{total_requests}] {color_flag} {elapsed:7.2f}ms | {bar}", flush=True)
  76. else:
  77.             errors.append(error)
  78. print(f"     [{i+1:3d}/{total_requests}] ❌ {error}", flush=True)
  79. # 控制请求频率
  80. if i < total_requests -1:
  81.             time.sleep(think_time)
  82. return response_times, errors
  83. # ─────────────────────────────────────────────
  84. # 第二部分：性能指标计算
  85. # ─────────────────────────────────────────────
  86. def calc_percentile(data, percentile):
  87. """计算指定百分位数"""
  88. ifnot data:
  89. returnNone
  90.     sorted_data = sorted(data)
  91.     idx = int(len(sorted_data)* percentile /100)
  92.     idx = min(idx, len(sorted_data)-1)
  93. return round(sorted_data[idx],2)
  94. def calc_performance_metrics(response_times):
  95. """计算完整性能指标"""
  96. ifnot response_times:
  97. return{}
  98.     sorted_times = sorted(response_times)
  99. return{
  100. 'count': len(response_times),
  101. 'min': round(min(response_times),2),
  102. 'max': round(max(response_times),2),
  103. 'mean': round(statistics.mean(response_times),2),
  104. 'median': round(statistics.median(response_times),2),
  105. 'stdev': round(statistics.stdev(response_times),2)if len(response_times)>1else0,
  106. 'p50': calc_percentile(response_times,50),
  107. 'p90': calc_percentile(response_times,90),
  108. 'p95': calc_percentile(response_times,95),
  109. 'p99': calc_percentile(response_times,99),
  110. }
  111. # ─────────────────────────────────────────────
  112. # 第三部分：可视化图表生成
  113. # ─────────────────────────────────────────────
  114. def generate_charts(metrics_dict, output_dir='output'):
  115. """生成性能图表"""
  116.     os.makedirs(output_dir, exist_ok=True)
  117. # 图1：各接口 P50/P90/P99 对比柱状图
  118.     fig, axes = plt.subplots(1,2, figsize=(16,6))
  119.     names = list(metrics_dict.keys())
  120.     p50_vals =[m.get('p50',0)for m in metrics_dict.values()if m]
  121.     p90_vals =[m.get('p90',0)for m in metrics_dict.values()if m]
  122.     p99_vals =[m.get('p99',0)for m in metrics_dict.values()if m]
  123.     x = np.arange(len(names))
  124.     width =0.25
  125.     bars1 = axes[0].bar(x - width, p50_vals, width, label='P50', color='#4CAF50')
  126.     bars2 = axes[0].bar(x, p90_vals, width, label='P90', color='#FF9800')
  127.     bars3 = axes[0].bar(x + width, p99_vals, width, label='P99', color='#F44336')
  128.     axes[0].set_xlabel('接口名称')
  129.     axes[0].set_ylabel('响应时间 (ms)')
  130.     axes[0].set_title('各接口响应时间 P50/P90/P99 对比')
  131.     axes[0].set_xticks(x)
  132.     axes[0].set_xticklabels(names, rotation=30, ha='right', fontsize=8)
  133.     axes[0].legend()
  134.     axes[0].grid(axis='y', alpha=0.3)
  135. # 给最高值标数字
  136. for bars in[bars1, bars2, bars3]:
  137. for bar in bars:
  138.             h = bar.get_height()
  139. if h >0:
  140.                 axes[0].annotate(f'{h:.0f}',
  141.                                 xy=(bar.get_x()+ bar.get_width()/2, h),
  142.                                 ha='center', va='bottom', fontsize=7)
  143. # 图2：热力图（横轴接口，纵轴百分位）
  144.     heatmap_data =[]
  145.     percentiles =[50,75,90,95,99]
  146. for p in percentiles:
  147.         row =[calc_percentile(metrics_dict[n]['raw_times'], p)if metrics_dict.get(n)else0
  148. for n in names]
  149.         heatmap_data.append(row)
  150.     heatmap_arr = np.array(heatmap_data)
  151.     im = axes[1].imshow(heatmap_arr, cmap='RdYlGn_r', aspect='auto')
  152.     axes[1].set_xticks(range(len(names)))
  153.     axes[1].set_xticklabels(names, rotation=30, ha='right', fontsize=8)
  154.     axes[1].set_yticks(range(len(percentiles)))
  155.     axes[1].set_yticklabels([f'P{p}'for p in percentiles])
  156.     axes[1].set_title('响应时间热力图 (ms)')
  157.     plt.colorbar(im, ax=axes[1], label='响应时间 (ms)')
  158. # 标注数值
  159. for i in range(len(percentiles)):
  160. for j in range(len(names)):
  161.             val = heatmap_arr[i, j]
  162. if val >0:
  163.                 axes[1].text(j, i, f'{val:.0f}', ha='center', va='center',
  164.                             color='white'if val >500else'black', fontsize=7)
  165.     plt.tight_layout()
  166.     chart_path = os.path.join(output_dir,'performance_chart.png')
  167.     plt.savefig(chart_path, dpi=150, bbox_inches='tight')
  168.     plt.close()
  169. print(f"\n  📊 图表已保存: {chart_path}")
  170. return chart_path
  171. # ─────────────────────────────────────────────
  172. # 第四部分：HTML 报告生成
  173. # ─────────────────────────────────────────────
  174. def generate_html_report(metrics_dict, error_summary, chart_path, output_file='report.html'):
  175. """生成 HTML 性能报告"""
  176.     total_requests = sum(m['count']for m in metrics_dict.values()if m)
  177.     total_errors = sum(len(errs)for _, errs in error_summary.values()if errs)
  178.     error_rate = total_errors /(total_requests + total_errors)*100if(total_requests + total_errors)>0else0
  179. # 颜色判定：P99 < 500ms 绿色，500-1000ms 黄色，>1000ms 红色
  180. def p99_color(p99):
  181. if p99 isNone:return'#999'
  182. if p99 <500:return'#4CAF50'
  183. if p99 <1000:return'#FF9800'
  184. return'#F44336'
  185.     rows =""
  186. for name, metrics in metrics_dict.items():
  187. ifnot metrics:
  188. continue
  189.         p99_c = p99_color(metrics.get('p99'))
  190.         rows += f"""
  191.         <tr>
  192.             <td><strong>{name}</strong></td>
  193.             <td>{metrics['count']}</td>
  194.             <td>{metrics['min']} ms</td>
  195.             <td>{metrics['mean']} ms</td>
  196.             <td>{metrics['median']} ms</td>
  197.             <td>{metrics['p90']} ms</td>
  198.             <td style="color:{p99_c};font-weight:bold">{metrics['p99']} ms</td>
  199.             <td>{metrics['max']} ms</td>
  200.             <td style="color:#F44336">{len(error_summary.get(name, [[]])[0]) if error_summary.get(name) else 0}</td>
  201.         </tr>"""
  202.     html = f"""
  203. <!DOCTYPE html>
  204. <html lang="zh">
  205. <head>
  206. <meta charset="UTF-8">
  207. <title>接口性能监控报告</title>
  208. <style>
  209.     body {{ font-family: -apple-system, 'Microsoft YaHei', sans-serif; padding: 20px; background: #f5f5f5; }}
  210.     .card {{ background: white; border-radius: 8px; padding: 20px; margin-bottom: 20px; box-shadow: 0 2px 8px rgba(0,0,0,0.1); }}
  211.     h1 {{ color: #333; border-bottom: 3px solid #4CAF50; padding-bottom: 10px; }}
  212.     .summary-grid {{ display: flex; gap: 20px; margin-bottom: 20px; }}
  213.     .stat-box {{ background: linear-gradient(135deg, #667eea, #764ba2); color: white; padding: 20px; border-radius: 8px; flex: 1; text-align: center; }}
  214.     .stat-box.red {{ background: linear-gradient(135deg, #f093fb, #f5576c); }}"
  215.     .stat-box.orange {{ background: linear-gradient(135deg, #ffecd2, #fcb69f); color: #333; }}"
  216.     .stat-number {{ font-size: 32px; font-weight: bold; }}
  217.     .stat-label {{ font-size: 14px; opacity: 0.9; margin-top: 5px; }}
  218.     table {{ width: 100%; border-collapse: collapse; }}
  219.     th {{ background: #4CAF50; color: white; padding: 12px; text-align: left; }}
  220.     td {{ padding: 10px; border-bottom: 1px solid #eee; }}
  221.     tr:hover {{ background: #f9f9f9; }}
  222.     .chart {{ text-align: center; margin: 20px 0; }}
  223.     .chart img {{ max-width: 100%; border-radius: 8px; box-shadow: 0 2px 8px rgba(0,0,0,0.15); }}
  224. </style>
  225. </head>
  226. <body>
  227. <div class="card">
  228.     <h1>📡 接口性能监控报告</h1>
  229.     <p><b>生成时间:</b> {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p>
  230. </div>
  231. <div class="summary-grid">
  232.     <div class="stat-box">
  233.         <div class="stat-number">{total_requests}</div>
  234.         <div class="stat-label">总请求数</div>
  235.     </div>
  236.     <div class="stat-box red">
  237.         <div class="stat-number">{total_errors}</div>
  238.         <div class="stat-label">异常请求</div>
  239.     </div>
  240.     <div class="stat-box {"orange" if error_rate > 5 else ""}">
  241.         <div class="stat-number">{error_rate:.2f}%</div>
  242.         <div class="stat-label">异常率</div>
  243.     </div>
  244. </div>
  245. <div class="card">
  246.     <h2>📊 性能指标详情</h2>
  247.     <table>
  248.         <thead>
  249.             <tr>
  250.                 <th>接口名称</th>
  251.                 <th>有效请求</th>
  252.                 <th>最小值</th>
  253.                 <th>平均值</th>
  254.                 <th>P50</th>
  255.                 <th>P90</th>
  256.                 <th>P99</th>
  257.                 <th>最大值</th>
  258.                 <th>错误数</th>
  259.             </tr>
  260.         </thead>
  261.         <tbody>
  262.             {rows}
  263.         </tbody>
  264.     </table>
  265. </div>
  266. <div class="card chart">
  267.     <h2>📈 性能图表</h2>
  268.     <img src="{os.path.basename(chart_path)}" alt="性能对比图">
  269. </div>
  270. <p style="text-align:center;color:#999;font-size:12px">
  271.     由 Python API 性能监控工具自动生成 | 依赖: requests + matplotlib + numpy
  272. </p>
  273. </body>
  274. </html>"""
  275. with open(output_file,'w', encoding='utf-8')as f:
  276.         f.write(html)
  277. print(f"\n  📄 HTML 报告已保存: {output_file}")
  278. # ─────────────────────────────────────────────
  279. # 第五部分：历史数据对比
  280. # ─────────────────────────────────────────────
  281. def compare_with_history(current_metrics, history_file):
  282. """与历史数据对比，检测性能回归"""
  283. ifnot os.path.exists(history_file):
  284. returnNone
  285. with open(history_file,'r', encoding='utf-8')as f:
  286.         history = json.load(f)
  287.     history_dict ={item['name']: item['metrics']for item in history}
  288.     comparison =[]
  289. for name, metrics in current_metrics.items():
  290. if name notin history_dict:
  291. continue
  292.         h = history_dict[name]
  293.         p99_delta = metrics.get('p99',0)- h.get('p99',0)
  294.         mean_delta = metrics.get('mean',0)- h.get('mean',0)
  295.         comparison.append({
  296. 'name': name,
  297. 'p99_change': round(p99_delta,2),
  298. 'mean_change': round(mean_delta,2),
  299. 'regressed': p99_delta >100# P99 上升超过 100ms 判定为回归
  300. })
  301. return comparison
  302. # ─────────────────────────────────────────────
  303. # 主入口
  304. # ─────────────────────────────────────────────
  305. def main():
  306. import urllib3
  307.     urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
  308.     parser = argparse.ArgumentParser(description='接口性能监控工具')
  309.     parser.add_argument('--config', default='apis.yaml', help='接口配置文件')
  310.     parser.add_argument('--output', default='output', help='输出目录')
  311.     parser.add_argument('--requests', type=int, default=None, help='每个接口请求次数（覆盖配置）')
  312.     parser.add_argument('--concurrency', type=int, default=None, help='并发数（覆盖配置）')
  313.     args = parser.parse_args()
  314. # 加载配置
  315. with open(args.config,'r', encoding='utf-8')as f:
  316.         config = yaml.safe_load(f)
  317.     apis = config.get('apis',[])
  318.     settings = config.get('settings',{})
  319.     total_requests = args.requests or settings.get('requests_per_api',100)
  320.     concurrency = args.concurrency or settings.get('concurrency',10)
  321.     think_time = settings.get('think_time',0.1)
  322.     history_file = settings.get('base_history_file','history_results.json')
  323.     os.makedirs(args.output, exist_ok=True)
  324. print(f"\n{'='*60}")
  325. print(f"🚀 接口性能监控 | {len(apis)} 个接口 × {total_requests} 次请求")
  326. print(f"{'='*60}")
  327.     all_metrics ={}
  328.     all_errors ={}
  329. for api in apis:
  330.         name = api['name']
  331.         r_times, errors = run_batch_test(name, api, total_requests, concurrency, think_time)
  332.         metrics = calc_performance_metrics(r_times)
  333. if metrics:
  334.             metrics['raw_times']= r_times  # 保存原始数据用于图表
  335.         all_metrics[name]= metrics
  336.         all_errors[name]=[errors]
  337. # 每个接口跑完保存一次（防止中途中断丢数据）
  338.         partial_file = os.path.join(args.output,'partial_results.json')
  339. with open(partial_file,'w', encoding='utf-8')as f:
  340.             json.dump({'metrics':{k:{kk: vv for kk, vv in v.items()if kk !='raw_times'}
  341. for k, v in all_metrics.items()},
  342. 'errors': all_errors}, f, ensure_ascii=False, indent=2)
  343. # 生成图表
  344.     chart_path = generate_charts(all_metrics, args.output)
  345. # 生成 HTML 报告
  346.     report_file = os.path.join(args.output,'api_perf_report.html')
  347.     generate_html_report(all_metrics, all_errors, chart_path, report_file)
  348. # 历史对比
  349.     comparison = compare_with_history(all_metrics, history_file)
  350. if comparison:
  351. print(f"\n{'='*60}")
  352. print(f"📈 与历史数据对比")
  353. print(f"{'='*60}")
  354. for c in comparison:
  355.             icon ="⚠️ 回归!"if c['regressed']else"✅"
  356. print(f"  {icon} {c['name']}: P99 变化 {c['p99_change']:+.2f}ms | 平均变化 {c['mean_change']:+.2f}ms")
  357. # 保存历史数据
  358.     history_entry ={
  359. 'timestamp': datetime.datetime.now().isoformat(),
  360. 'metrics':{k:{kk: vv for kk, vv in v.items()if kk !='raw_times'}
  361. for k, v in all_metrics.items()}
  362. }
  363. if os.path.exists(history_file):
  364. with open(history_file,'r', encoding='utf-8')as f:
  365.             history_data = json.load(f)
  366. else:
  367.         history_data =[]
  368.     history_data.append(history_entry)
  369. with open(history_file,'w', encoding='utf-8')as f:
  370.         json.dump(history_data, f, ensure_ascii=False, indent=2)
  371. print(f"\n{'='*60}")
  372. print(f"✅ 测试完成！")
  373. print(f"   报告: {report_file}")
  374. print(f"   图表: {chart_path}")
  375. print(f"   历史: {history_file}")
  376. print(f"{'='*60}")
  377. if __name__ =='__main__':
  378.     main()

运行效果

运行后输出示例：

  1. ============================================================
  2. 🚀接口性能监控|4个接口×100次请求
  3. ============================================================

  5. 📡测试接口:用户登录接口
  6.      URL: https://test.example.com/api/user/login
  7. 总请求:100|并发:10
  8. [1/100]🟢234.50ms|████████████████████
  9. [2/100]🟢198.30ms|█████████████████
  10. ...
  11. [100/100]🔴2156.00ms|████████████████████████████████████████

  13. 📊图表已保存: output/performance_chart.png
  14. 📄 HTML 报告已保存: output/api_perf_report.html

  16. ============================================================
  17. 📈与历史数据对比
  18. ============================================================
  19. ⚠️回归!用户登录接口: P99 变化+156.32ms|平均变化+43.21ms
  20. ✅获取用户信息: P99 变化-5.12ms|平均变化-2.33ms
  21. ============================================================

生成的 HTML 报告截图效果：包含汇总数据卡片 + 详细指标表格 + P50/P90/P99 柱状对比图 + 热力图。

快速上手步骤

Step 1：安装依赖

  1. pip install requests pyyaml matplotlib numpy

Step 2：编辑接口配置

修改 apis.yaml，填入你要测试的接口 URL、请求方式、参数等。

Step 3：运行测试

  1. python api_monitor.py --config apis.yaml --requests 100

Step 4：查看报告

直接用浏览器打开 output/api_perf_report.html，发给领导也行。

使用技巧

场景
建议
性能回归测试
每次发版前跑一遍，history_results.json 自动累积数据对比
上线前摸底
调到 requests_per_api: 500，更能发现问题
慢接口定位
看热力图中颜色最深的那行，即 P99 最差的接口
并发压测
把 concurrency 调高（真正的并发需要用 concurrent.futures，感兴趣可以改）

总结

这个工具解决了两个核心问题：

  1. 不知道接口的真实性能
     → P50/P90/P99 把”平均正常”背后的真实分布展示出来
  2. 没有数据对比
     → history_results.json 自动积累历史，每次跑完自动和上次比，发现性能回归

代码可以直接用，改改配置文件的 URL 和参数就行。