Agent天气系统开发实战：从零构建高可用气象数据服务

1次阅读

共计 2375 个字符，预计需要花费 6 分钟才能阅读完成。

构建气象服务时，开发者通常会遇到三个主要挑战：

数据源异构性 ：不同气象数据提供商（如中国气象局、OpenWeatherMap、AccuWeather）返回的数据结构和更新频率各不相同，需要统一处理。
实时性要求 ：气象数据具有强时效性，过时的数据可能完全失去价值。
服务稳定性 ：第三方 API 可能不稳定，需要设计完善的降级和容错机制。

优点：开发简单，快速上线
缺点：
依赖外部服务稳定性
难以定制数据格式
可能产生高昂的 API 调用费用

优点：
完全掌控数据流程
可以优化数据存储和查询
长期成本更低
缺点：
初期开发成本高
需要维护基础设施

class WeatherAgent:
    def __init__(self):
        self.cache = RedisCache()
        self.data_sources = [ChinaMeteorologicalAdmin(),
            OpenWeatherMap(),
            AccuWeather()]

    def get_weather(self, location):
        # 先检查缓存
        cached_data = self.cache.get(location)
        if cached_data:
            return cached_data

        # 多数据源采集
        for source in self.data_sources:
            try:
                data = source.fetch(location)
                if data:
                    self.cache.set(location, data)
                    return data
            except Exception as e:
                logging.error(f"数据源 {source} 获取失败: {e}")

        raise WeatherDataUnavailableError("所有数据源均不可用")

from flask import Flask, jsonify
from flask_jwt_extended import JWTManager, jwt_required
from flask_limiter import Limiter

app = Flask(__name__)

# JWT 配置
app.config['JWT_SECRET_KEY'] = 'your-secret-key'
jwt = JWTManager(app)

# 限流配置
limiter = Limiter(
    app=app,
    key_func=lambda: request.remote_addr,
    default_limits=["200 per day", "50 per hour"]
)

weather_agent = WeatherAgent()

@app.route('/weather/<location>')
@jwt_required()
@limiter.limit("10 per minute")
def get_weather(location):
    try:
        data = weather_agent.get_weather(location)
        return jsonify(data)
    except WeatherDataUnavailableError:
        return jsonify({"error": "服务暂时不可用"}), 503

def merge_weather_data(sources_data):
    """
    合并多个数据源的天气数据
    优先级：中国气象局 > OpenWeatherMap > AccuWeather
    """base_data = {'temperature': None,'humidity': None,'wind_speed': None,
        # 其他字段
    }

    for source in ['ChinaMeteorologicalAdmin', 'OpenWeatherMap', 'AccuWeather']:
        if source in sources_data and sources_data[source]:
            for key in base_data.keys():
                if base_data[key] is None and key in sources_data[source]:
                    base_data[key] = sources_data[source][key]

    return base_data

使用 Locust 进行测试，配置：
– 100 并发用户
– 持续 5 分钟

结果：
– 平均响应时间：78ms
– 95% 响应时间：120ms
– 失败率：0.2%

使用更合理的数据结构：
采用 Hash 存储天气数据，而非简单 String
为常用查询建立索引
优化缓存过期策略：
静态数据（如城市信息）：长期缓存
实时天气数据：5 分钟过期
天气预报数据：1 小时过期

商用气象数据通常需要授权
注意数据使用限制条款
敏感地区数据可能需要特殊处理

常见错误：

未考虑夏令时
将 UTC 时间直接转为本地时间而未考虑时区
不同数据源使用不同时区标准

解决方案：

import pytz
from datetime import datetime

def convert_timezone(dt, from_tz, to_tz):
    from_zone = pytz.timezone(from_tz)
    to_zone = pytz.timezone(to_tz)

    localized = from_zone.localize(dt)
    return localized.astimezone(to_zone)