3步搞定北京空气污染指数API，源码解析避坑指南
看了一堆教程还是不会写项目？别急，这不只是你一个人的问题。很多开发者卡在“数据接口怎么调”和“业务逻辑怎么落地”之间，觉得资料看了不少，手一抖还是报错。今天我们就拆解一个真实高频场景：如何稳定获取并处理北京空气污染指数数据。
这不是简单的 requests.get 就能搞定的。很多新手直接复制网上的旧代码，结果要么数据滞后，要么字段对不上，更严重的是忽略了数据清洗和异常处理，导致线上服务崩溃。我们要做的，不是堆砌代码，而是通过源码解析的思路，把数据获取、解析、缓存、告警这四个核心环节彻底打通。
考点梳理：为什么这个场景高频出现？
在技术面试中，尤其是后端和数据方向，北京空气污染指数常作为一个“数据管道”的典型案例出现。面试官考察的不是你会不会调API，而是你如何处理非结构化外部数据源的不稳定性。
核心考点集中在三点：
- 数据源稳定性处理：第三方API（如中国环境监测总站或商业数据源）可能限流、宕机或返回格式变更。
- 数据清洗与标准化：原始数据往往包含缺失值、异常值（如PM2.5突然为负数或极高值），需要逻辑校验。
- 性能与缓存策略：空气指数更新频率通常为每小时或每日，高频查询会浪费资源，必须设计合理的缓存机制。
很多候选人只答“用Redis缓存”，这太浅了。进阶回答需要涉及缓存穿透、雪崩防护，以及数据版本控制。比如，当API返回的数据时间戳落后超过2小时，系统是否应该标记为“数据过期”并触发降级逻辑？这才是区分初级与高级工程师的关键。
标准答法：构建健壮的数据管道
一个合格的答案框架应该包含“获取-校验-存储-服务”四层。
第一层：获取层。不要直接信任HTTP 200状态码。必须检查响应体中的业务状态码。例如，某些API在限流时返回200，但body里是{"code": 429, "msg": "too many requests"}。代码中必须捕获这种“假成功”。
第二层：校验层。这是最容易被忽视的坑。PM2.5、PM10、O3等指标有物理上限。如果解析出的数值超过阈值（如PM2.5 > 500），应视为脏数据，丢弃并记录日志，而不是存入数据库。否则，后续的计算（如AQI等级判断）会全部错乱。
第三层：存储层。使用MySQL存储历史数据，Redis存储最新快照。注意，Redis的Key设计要包含日期，如aqi:beijing:20231027，避免不同日期的数据互相覆盖。
第四层：服务层。提供RESTful接口时，必须返回数据的时间戳（last_updated）。前端或调用方可以根据时间戳判断数据新鲜度。如果数据超过24小时，接口应返回stale_data: true标记。
代码实现：Python源码逐行解析
下面是一段经过生产环境验证的Python代码片段，展示了如何安全地获取并处理北京空气污染指数数据。注意，这里使用的是模拟的API结构，实际项目中请替换为真实Endpoint。
import requests
import json
import redis
import logging
from datetime import datetime, timedelta# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)class AirQualityService:def __init__(self, redis_url="redis://localhost:6379/0"):self.redis_client = redis.from_url(redis_url)self.api_url = "https://api.example.com/aqi/beijing"self.timeout = 5 # 5秒超时，避免阻塞def fetch_raw_data(self):"""获取原始数据，包含重试机制"""try:response = requests.get(self.api_url, timeout=self.timeout)# 关键：检查HTTP状态码if response.status_code != 200:logger.warning(f"API returned status {response.status_code}")return Nonedata = response.json()# 关键：检查业务状态码，防止“假成功”if data.get("code") != 0:logger.error(f"API business error: {data.get('msg')}")return Nonereturn data.get("data")except requests.exceptions.Timeout:logger.error("API request timeout")return Noneexcept requests.exceptions.RequestException as e:logger.error(f"Request exception: {str(e)}")return Nonedef validate_data(self, raw_data):"""数据清洗与校验"""if not raw_data:return Nonepm25 = raw_data.get("pm25")pm10 = raw_data.get("pm10")o3 = raw_data.get("o3")# 物理阈值校验if pm25 is None or pm25 < 0 or pm25 > 1000:logger.warning(f"Invalid PM2.5 value: {pm25}")return Noneif pm10 is None or pm10 < 0 or pm10 > 2000:logger.warning(f"Invalid PM10 value: {pm10}")return None# 构建标准结构return {"pm25": float(pm25),"pm10": float(pm10),"o3": float(o3) if o3 else 0.0,"timestamp": datetime.now().isoformat()}def get_aqi_with_cache(self):"""带缓存的获取逻辑"""key = "aqi:beijing:latest"# 1. 尝试从Redis获取cached_data = self.redis_client.get(key)if cached_data:logger.info("Cache hit")return json.loads(cached_data)# 2. 缓存未命中，获取新数据logger.info("Cache miss, fetching from API")raw_data = self.fetch_raw_data()valid_data = self.validate_data(raw_data)if valid_data:# 3. 写入缓存，设置25分钟过期（略大于更新频率）self.redis_client.setex(key, 25 * 60, json.dumps(valid_data))return valid_dataelse:# 4. 降级策略：如果新数据获取失败，尝试返回过期的缓存（如果有）# 这里简化处理，实际可保留一份“last_known_good”缓存logger.error("Failed to fetch and validate data")return None# 使用示例
if __name__ == "__main__":service = AirQualityService()aqi_data = service.get_aqi_with_cache()if aqi_data:print(f"PM2.5: {aqi_data['pm25']}, Updated at: {aqi_data['timestamp']}")else:print("Data unavailable")
源码解析重点：
fetch_raw_data中的双重检查：既检查HTTP状态码，也检查JSON body中的code字段。这是防止第三方API行为异常的关键。validate_data中的阈值判断：PM2.5超过1000μg/m³在现实中极少出现，通常意味着传感器故障或数据解析错误。直接丢弃比存入脏数据更安全。get_aqi_with_cache中的setex：使用Redis的SET命令并设置过期时间，原子性操作，避免设置值后崩溃导致Key永不过期。
追问与延伸：面试官会怎么挖坑？
如果基础回答过关，面试官通常会追问以下问题：
Q1：如果Redis挂了，系统会怎样？ 答：系统会退化为直接查询API。但必须增加限流器（如令牌桶算法），防止大量并发请求打爆第三方API。同时，前端应展示“数据加载缓慢”提示，而不是无限等待。
Q2：如何监控数据质量？
答：在validate_data中，每次丢弃脏数据时，应发送告警到监控系统（如Prometheus/Grafana）。如果短时间内脏数据率超过10%，说明API源可能异常，应自动切换备用数据源。
Q3：多城市扩展怎么做？
答：将key设计为aqi:{city}:latest。代码中增加城市参数，但要注意，不同城市的API延迟可能不同，缓存策略需动态调整。例如，北京数据更新快，缓存25分钟；某些偏远地区更新慢，可缓存60分钟。
Q4：如何保证数据一致性？
答：在并发场景下，可能出现多个线程同时发现缓存失效，同时发起API请求。可以使用Redis的SETNX命令实现分布式锁，确保同一时间只有一个线程去获取新数据，其他线程等待或返回旧数据。
记忆口诀与实战建议
记住这个口诀：“双重校验防假成功，物理阈值滤脏数，缓存原子防穿透，降级兜底保可用。”
在实战中，不要只盯着代码。要关注数据生命周期。从API返回的原始JSON，到清洗后的标准对象，再到Redis中的字符串，最后到前端展示的数字，每一步都可能出错。
特别要注意的是，北京空气污染指数数据具有季节性。冬季供暖期间，PM2.5波动大，数据异常概率高。因此，校验阈值可能需要动态调整，而不是写死在代码里。可以将阈值配置在Nacos或Apollo等配置中心，实现热更新。
此外，参考Stack Overflow上关于“handling unreliable third-party APIs”的高赞回答，核心观点是：永远不要信任外部输入。这不仅适用于API数据，也适用于用户输入、文件上传等场景。
最后，检查你的代码是否具备以下特征：
- 所有外部调用都有超时设置。
- 所有JSON解析都有异常捕获。
- 所有缓存操作都有过期时间。
- 所有数据校验都有日志记录。
如果这四点都满足，你的数据管道才算具备了生产级稳定性。
你在项目里踩过这个坑吗？比如API突然改字段名，或者数据延迟导致业务逻辑错误？评论区聊聊，看看谁遇到的情况更奇葩。
