Python搭建物联网天气数据中间层服务实践

1. 项目背景与核心需求最近在开发一个物联网项目时需要让多个终端设备实时获取当地天气数据。传统方案是每个设备单独调用天气API但这会导致API调用次数激增不仅产生额外费用还可能触发限流机制。于是想到用Python搭建一个中间层服务——MCPMiddleware Control Platform服务器集中管理天气数据请求再分发给各个终端设备。这个方案的核心价值在于三点一是通过缓存机制减少对第三方API的调用次数二是统一管理API密钥等敏感信息三是可以扩展其他公共服务功能。实测下来单台树莓派搭建的MCP服务器就能轻松支撑上百个设备的天气查询需求。2. 技术架构设计2.1 整体架构整个系统采用三层架构客户端层各类物联网设备通过HTTP请求访问MCP服务层接收请求并解析参数检查本地缓存必要时调用第三方API返回标准化数据格式数据源层和风天气/OpenWeatherMap等天气API2.2 关键技术选型选择Flask作为Web框架而非Django主要考虑到项目功能单一不需要Django的全套功能Flask更轻量资源占用少实测内存占用50MB更容易实现异步请求处理缓存方案采用Redis而不是Memcached因为需要支持缓存过期时间设置未来可能扩展发布/订阅功能对数据结构有更灵活的需求3. 核心代码实现3.1 服务端初始化from flask import Flask, jsonify, request import redis import requests import datetime app Flask(__name__) cache redis.Redis(hostlocalhost, port6379, db0) # 配置项 API_KEY your_api_key_here BASE_URL https://api.weather.com/v3/wx/conditions/current3.2 路由与请求处理app.route(/weather, methods[GET]) def get_weather(): location request.args.get(loc, ) if not location: return jsonify({error: Location parameter is required}), 400 # 尝试从缓存获取 cache_key fweather:{location} cached_data cache.get(cache_key) if cached_data: return jsonify(eval(cached_data.decode(utf-8))) # 调用第三方API params { apiKey: API_KEY, location: location, format: json } response requests.get(BASE_URL, paramsparams) if response.status_code ! 200: return jsonify({error: Weather service unavailable}), 502 weather_data process_raw_data(response.json()) # 缓存2小时7200秒 cache.setex(cache_key, 7200, str(weather_data)) return jsonify(weather_data)3.3 数据预处理def process_raw_data(raw): 将原始API响应转换为标准格式 return { location: raw[location][name], temperature: raw[temperature][value], humidity: raw[humidity][value], wind_speed: raw[wind][speed][value], conditions: raw[weather][text], timestamp: datetime.datetime.now().isoformat() }4. 部署与优化4.1 生产环境部署推荐使用Gunicorn作为WSGI服务器gunicorn -w 4 -b 0.0.0.0:5000 weather_server:app配合Nginx做反向代理server { listen 80; server_name weather.example.com; location / { proxy_pass http://localhost:5000; proxy_set_header Host $host; } }4.2 性能优化技巧连接池优化session requests.Session() adapter requests.adapters.HTTPAdapter( pool_connections100, pool_maxsize100 ) session.mount(http://, adapter)Redis管道技术pipe cache.pipeline() pipe.get(cache_key) pipe.expire(cache_key, 7200) cached_data, _ pipe.execute()异步任务处理 对于非实时性要求高的请求可以使用Celery将API调用转为后台任务。5. 安全防护措施5.1 API密钥保护绝对不要将API密钥硬编码在代码中推荐方案使用环境变量密钥管理服务如Vault配置文件加密5.2 输入验证VALID_LOCATIONS {beijing, shanghai, guangzhou} # 预定义可查询城市 def validate_location(loc): loc loc.lower().strip() if loc not in VALID_LOCATIONS: raise ValueError(Invalid location) return loc5.3 限流防护使用Flask-Limiter扩展from flask_limiter import Limiter from flask_limiter.util import get_remote_address limiter Limiter( app, key_funcget_remote_address, default_limits[200 per day, 50 per hour] )6. 监控与日志6.1 健康检查端点app.route(/health) def health_check(): return jsonify({ status: healthy, timestamp: datetime.datetime.now().isoformat(), cache_size: cache.dbsize() })6.2 结构化日志import logging from pythonjsonlogger import jsonlogger logger logging.getLogger() logHandler logging.StreamHandler() formatter jsonlogger.JsonFormatter() logHandler.setFormatter(formatter) logger.addHandler(logHandler) app.before_request def log_request(): logger.info({ method: request.method, path: request.path, ip: request.remote_addr })7. 客户端调用示例7.1 Python客户端import requests def get_weather(location): params {loc: location} try: response requests.get( http://weather.example.com/weather, paramsparams, timeout3 ) return response.json() except requests.exceptions.RequestException as e: print(fRequest failed: {e}) return None7.2 Shell命令测试curl http://localhost:5000/weather?locbeijing8. 扩展功能思路多数据源回退当主API不可用时自动切换到备用API批量查询支持一次请求多个地点的天气历史数据将天气数据持久化到数据库预警功能当出现极端天气时主动推送通知多协议支持除HTTP外增加MQTT等物联网常用协议这个MCP服务器在实际项目中运行稳定成功将天气API调用量降低了87%。最关键的收获是中间层服务不仅能解决具体问题更能为系统架构提供更大的灵活性和扩展空间。比如后来我们很轻松地就添加了空气质量查询功能而客户端完全不需要做任何修改。