4. 数据采集基础：使用Python发送HTTP请求、记录响应时间、状态码与延迟数据提取、批量请求与并发控制

数据采集是延迟诊断的基石。在策略即代码的框架下，每一次API调用都必须被精确度量，以形成可对比的基线数据。本节将深入讲解如何使用Python构建可靠的数据采集层，涵盖从单次请求到批量并发的完整技术栈。

4.1 HTTP请求核心：requests库与响应时间记录

Python的requests库是发送HTTP请求的标准工具。在延迟诊断场景中，我们不仅关注响应内容，更关注请求的完整生命周期时间线。

以下代码展示了如何利用requests的钩子（hooks）机制精确捕获各阶段耗时：

import requests
import time

class LatencyTracker:
    def __init__(self):
        self.timings = {}
        self.start = None

    def on_request_start(self, **kwargs):
        self.start = time.perf_counter()
        self.timings['request_start'] = self.start

    def on_response_received(self, **kwargs):
        self.timings['response_received'] = time.perf_counter()

    def on_response_complete(self, **kwargs):
        self.timings['response_complete'] = time.perf_counter()

    def get_latency_report(self):
        if not self.timings:
            return {}
        return {
            'total_latency_ms': round(
                (self.timings['response_complete'] - self.timings['request_start']) * 1000, 2
            ),
            'network_latency_ms': round(
                (self.timings['response_received'] - self.timings['request_start']) * 1000, 2
            ),
            'transfer_latency_ms': round(
                (self.timings['response_complete'] - self.timings['response_received']) * 1000, 2
            )
        }

# 使用示例
tracker = LatencyTracker()
response = requests.get(
    'https://api.example.com/health',
    hooks={
        'pre_request': tracker.on_request_start,
        'response': tracker.on_response_received
    }
)
tracker.on_response_complete()
print(tracker.get_latency_report())

4.2 状态码与延迟数据提取

每次请求完成后，必须提取结构化数据用于后续分析。状态码是判断请求成功与否的第一道关卡，而延迟数据则是性能基线的基础。

4.3 批量请求与并发控制

在发布窗口的基线偏差检查中，通常需要同时向多个端点或同一端点发送大量请求以建立统计基线。Python提供了多种并发方案，其中concurrent.futures与asyncio是最常用的两种。

4.3.1 使用ThreadPoolExecutor实现并发请求

from concurrent.futures import ThreadPoolExecutor, as_completed
import requests

def fetch_url(url: str, timeout: int = 5) -> dict:
    """发送单个请求并返回结构化数据"""
    start = time.perf_counter()
    try:
        resp = requests.get(url, timeout=timeout)
        latency_ms = round((time.perf_counter() - start) * 1000, 2)
        return {
            'url': url,
            'status': resp.status_code,
            'latency_ms': latency_ms,
            'success': True
        }
    except requests.exceptions.RequestException as e:
        return {
            'url': url,
            'status': None,
            'latency_ms': None,
            'success': False,
            'error': str(e)
        }

def batch_request(urls: list, max_workers: int = 10) -> list:
    """批量并发请求"""
    results = []
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        future_to_url = {executor.submit(fetch_url, url): url for url in urls}
        for future in as_completed(future_to_url):
            results.append(future.result())
    return results

# 示例：并发请求10个端点
urls = [f'https://api.example.com/endpoint/{i}' for i in range(10)]
batch_results = batch_request(urls, max_workers=5)

4.3.2 并发控制策略对比

4.4 数据采集架构总览

下图展示了从单次请求到批量并发采集的完整数据流，以及各组件之间的协作关系：

4.5 实战：构建可复用的数据采集器

将上述技术整合为一个可复用的采集器类，便于在发布窗口的基线检查中直接调用：

import time
import logging
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field
from typing import List, Dict, Optional
import requests

logger = logging.getLogger(__name__)

@dataclass
class RequestResult:
    url: str
    status_code: Optional[int]
    total_latency_ms: float
    ttfb_ms: float
    response_size: int
    timestamp: str
    error: Optional[str] = None

class APILatencyCollector:
    """API延迟数据采集器"""
    
    def __init__(self, base_url: str, timeout: int = 10, max_workers: int = 10):
        self.base_url = base_url.rstrip('/')
        self.timeout = timeout
        self.max_workers = max_workers
        self.session = requests.Session()
    
    def single_request(self, endpoint: str) -> RequestResult:
        """发送单次请求并记录延迟"""
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        start = time.perf_counter()
        try:
            resp = self.session.get(url, timeout=self.timeout)
            total_latency = round((time.perf_counter() - start) * 1000, 2)
            ttfb = round(resp.elapsed.total_seconds() * 1000, 2)
            return RequestResult(
                url=url,
                status_code=resp.status_code,
                total_latency_ms=total_latency,
                ttfb_ms=ttfb,
                response_size=len(resp.content),
                timestamp=time.strftime('%Y-%m-%dT%H:%M:%S.%fZ', time.gmtime())
            )
        except requests.exceptions.RequestException as e:
            total_latency = round((time.perf_counter() - start) * 1000, 2)
            logger.warning(f"请求失败: {url} - {str(e)}")
            return RequestResult(
                url=url,
                status_code=None,
                total_latency_ms=total_latency,
                ttfb_ms=0.0,
                response_size=0,
                timestamp=time.strftime('%Y-%m-%dT%H:%M:%S.%fZ', time.gmtime()),
                error=str(e)
            )
    
    def batch_request(self, endpoints: List[str]) -> List[RequestResult]:
        """批量并发请求"""
        results = []
        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            future_map = {
                executor.submit(self.single_request, ep): ep 
                for ep in endpoints
            }
            for future in as_completed(future_map):
                try:
                    results.append(future.result())
                except Exception as e:
                    logger.error(f"批量请求异常: {str(e)}")
        return results

# 使用示例
collector = APILatencyCollector(
    base_url='https://api.example.com',
    timeout=5,
    max_workers=8
)
results = collector.batch_request(['health', 'metrics', 'status'])
for r in results:
    print(f"{r.url} -> {r.status_code} | {r.total_latency_ms}ms")

4.6 数据采集质量检查清单

在将采集器投入基线偏差检查前，请逐项确认以下要点：

1. 时间精度：使用time.perf_counter()而非time.time()，避免系统时钟调整影响
2. 异常覆盖：捕获ConnectionError、Timeout、HTTPError等所有常见异常
3. 数据完整性：每条记录必须包含时间戳、URL、状态码、延迟四个核心字段
4. 并发稳定性：在目标服务可承受范围内测试最大并发数，记录错误率
5. 日志记录：对每次请求的开始和结束打印结构化日志，便于事后审计
6. 资源清理：确保Session在使用完毕后正确关闭

完成上述数据采集基础建设后，下一阶段将进入延迟数据的统计分析与基线建模，为发布窗口中的偏差检测提供数学依据。