构建生产级地理编码客户端：坐标技巧、连接池和速率限制

地理编码看起来很简单：发送地址，获取坐标。但在规模化生产中，会出现三个教程从不提及的隐藏问题：

1. 在中国，坐标被有意混淆
2. HTTP 连接无法正确清理
3. API 配额消耗速度超过预期

我曾见过这三个问题每一个都导致生产系统故障。

架构概览

下面是完整的架构图：

三个模块，三个问题。

问题 #1：300 米坐标漂移

在中国，消费级地图应用不使用原始 GPS 坐标（WGS-84）。取而代之，它们使用 GCJ-02——被戏称为"火星坐标"——这种系统对每个点进行非线性偏移。这不是简单的平移；而是涉及三角函数的有意混淆，根据地点位置，坐标可能偏移 50 到 500 米。

如果你的后端从 GPS 设备接收 WGS-84 坐标，然后直接传给中文地图 API，标记会出现在错误位置。配送区域崩溃。地理围栏失效。用户抱怨"地图错了"。

解决方案：迭代反演

GCJ-02 转换没有封闭解。我们通过数值方法求解：

1. 假设 WGS-84 ≈ GCJ-02（作为初值足够好）
2. 对猜测的坐标进行正向转换得到预测的 GCJ-02
3. 测量误差并调整
4. 重复直到误差 < 10⁻⁶ 度（约 0.1 米）

import math

_A = 6378245.0  # Semi-major axis
_EE = 0.00669342162296594323  # First eccentricity squared


def _out_of_china(lng: float, lat: float) -> bool:
    """Check if point is outside China (skip transformation)."""
    return not (72.004 <= lng <= 137.8347 and 0.8293 <= lat <= 55.8271)


def _transform_lat(x: float, y: float) -> float:
    ret = -100.0 + 2.0 * x + 3.0 * y + 0.2 * y * y + 0.1 * x * y
    ret += 0.2 * math.sqrt(abs(x))
    ret += (20.0 * math.sin(6.0 * x * math.pi) + 
            20.0 * math.sin(2.0 * x * math.pi)) * 2.0 / 3.0
    ret += (20.0 * math.sin(y * math.pi) + 
            40.0 * math.sin(y / 3.0 * math.pi)) * 2.0 / 3.0
    ret += (160.0 * math.sin(y / 12.0 * math.pi) + 
            320.0 * math.sin(y * math.pi / 30.0)) * 2.0 / 3.0
    return ret


def _transform_lng(x: float, y: float) -> float:
    ret = 300.0 + x + 2.0 * y + 0.1 * x * x + 0.1 * x * y
    ret += 0.1 * math.sqrt(abs(x))
    ret += (20.0 * math.sin(6.0 * x * math.pi) + 
            20.0 * math.sin(2.0 * x * math.pi)) * 2.0 / 3.0
    ret += (20.0 * math.sin(x * math.pi) + 
            40.0 * math.sin(x / 3.0 * math.pi)) * 2.0 / 3.0
    ret += (150.0 * math.sin(x / 12.0 * math.pi) + 
            300.0 * math.sin(x / 30.0 * math.pi)) * 2.0 / 3.0
    return ret


def wgs84_to_gcj02(lng: float, lat: float) -> tuple[float, float]:
    """Convert WGS-84 to GCJ-02 (forward transformation)."""
    if _out_of_china(lng, lat):
        return lng, lat
    
    dlat = _transform_lat(lng - 105.0, lat - 35.0)
    dlng = _transform_lng(lng - 105.0, lat - 35.0)
    
    rad_lat = lat / 180.0 * math.pi
    magic = math.sin(rad_lat)
    magic = 1 - _EE * magic * magic
    sqrt_magic = math.sqrt(magic)
    
    dlat = (dlat * 180.0) / ((_A * (1 - _EE)) / (magic * sqrt_magic) * math.pi)
    dlng = (dlng * 180.0) / (_A / sqrt_magic * math.cos(rad_lat) * math.pi)
    
    return lng + dlng, lat + dlat


def gcj02_to_wgs84(lng: float, lat: float, 
                   tol: float = 1e-6, 
                   max_iter: int = 10) -> tuple[float, float]:
    """Convert GCJ-02 back to WGS-84 via iterative refinement."""
    wgs_lng, wgs_lat = lng, lat
    
    for _ in range(max_iter):
        pred_lng, pred_lat = wgs84_to_gcj02(wgs_lng, wgs_lat)
        dlng = pred_lng - lng
        dlat = pred_lat - lat
        
        if abs(dlng) < tol and abs(dlat) < tol:
            break
            
        wgs_lng -= dlng
        wgs_lat -= dlat
    
    return wgs_lng, wgs_lat

迭代通常在 2-3 步内收敛。

问题 #2：TIME_WAIT 风暴

朴素的地理编码请求会为每个查询创建一个新的 TCP 连接：

connect() → send HTTP request → receive response → close()

在 500 QPS 的情况下，每秒创建 500 个连接。每个关闭的连接会进入 TIME_WAIT 状态，持续 60 秒（Linux 默认值）。计算一下：500 × 60 = 30,000 个套接字卡在 TIME_WAIT 中，而典型系统只有约 28,000 个临时端口。

结果：OSError: [Errno 99] Cannot assign requested address

解决方案：连接池 + 指数退避

用持久会话替换无状态请求，重用 TCP 连接：

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry


def create_resilient_session(pool_size: int = 50) -> requests.Session:
    """
    Create an HTTP session with connection pooling and retry logic.
    
    Args:
        pool_size: Max connections to keep alive (tune based on expected QPS)
    """
    session = requests.Session()
    
    retry_strategy = Retry(
        total=5,
        backoff_factor=1,  # delays: 0s, 2s, 4s, 8s, 16s
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["GET"],
    )
    
    adapter = HTTPAdapter(
        max_retries=retry_strategy,
        pool_connections=pool_size,
        pool_maxsize=pool_size,
    )
    
    session.mount("https://", adapter)
    session.mount("http://", adapter)
    
    return session

连接池大小：使用 pool_size = expected_QPS × average_latency。在 500 QPS、100ms 延迟的情况下，需要 50 个连接。

问题 #3：配额消耗殆尽

商业地理编码 API 有严格的速率限制。如果没有客户端限流，一次流量突增会在数分钟内耗尽你的日配额——然后整天返回 HTTP 429。

解决方案：令牌桶速率限制

令牌桶算法强制执行平均速率，同时允许短期突发：

import threading
import time
from dataclasses import dataclass, field


@dataclass
class TokenBucket:
    """
    Local token bucket rate limiter.
    For distributed systems, use Redis with Lua scripts.
    """
    rate: float      # Tokens per second
    capacity: float  # Max burst size
    _tokens: float = field(init=False)
    _timestamp: float = field(init=False)
    _lock: threading.Lock = field(init=False)
    
    def __post_init__(self):
        self._tokens = self.capacity
        self._timestamp = time.time()
        self._lock = threading.Lock()
    
    def acquire(self, tokens: int = 1) -> bool:
        with self._lock:
            now = time.time()
            self._tokens = min(
                self.capacity,
                self._tokens + (now - self._timestamp) * self.rate
            )
            self._timestamp = now
            
            if self._tokens >= tokens:
                self._tokens -= tokens
                return True
            return False

设置 capacity = rate × 2 可以允许 2 秒的突发而不冒配额耗尽的风险。

完整实现

下面是将所有内容整合到生产级类中的代码：

import math
import threading
import time
from dataclasses import dataclass, field
from typing import Any

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry


# === Coordinate Transformation ===

_A = 6378245.0
_EE = 0.00669342162296594323

def _out_of_china(lng: float, lat: float) -> bool:
    return not (72.004 <= lng <= 137.8347 and 0.8293 <= lat <= 55.8271)

def _transform_lat(x: float, y: float) -> float:
    ret = -100.0 + 2.0 * x + 3.0 * y + 0.2 * y * y + 0.1 * x * y
    ret += 0.2 * math.sqrt(abs(x))
    ret += (20.0 * math.sin(6.0 * x * math.pi) + 
            20.0 * math.sin(2.0 * x * math.pi)) * 2.0 / 3.0
    ret += (20.0 * math.sin(y * math.pi) + 
            40.0 * math.sin(y / 3.0 * math.pi)) * 2.0 / 3.0
    ret += (160.0 * math.sin(y / 12.0 * math.pi) + 
            320.0 * math.sin(y * math.pi / 30.0)) * 2.0 / 3.0
    return ret

def _transform_lng(x: float, y: float) -> float:
    ret = 300.0 + x + 2.0 * y + 0.1 * x * x + 0.1 * x * y
    ret += 0.1 * math.sqrt(abs(x))
    ret += (20.0 * math.sin(6.0 * x * math.pi) + 
            20.0 * math.sin(2.0 * x * math.pi)) * 2.0 / 3.0
    ret += (20.0 * math.sin(x * math.pi) + 
            40.0 * math.sin(x / 3.0 * math.pi)) * 2.0 / 3.0
    ret += (150.0 * math.sin(x / 12.0 * math.pi) + 
            300.0 * math.sin(x / 30.0 * math.pi)) * 2.0 / 3.0
    return ret


# === Rate Limiter ===

@dataclass
class TokenBucket:
    rate: float
    capacity: float
    _tokens: float = field(init=False)
    _timestamp: float = field(init=False)
    _lock: threading.Lock = field(init=False)
    
    def __post_init__(self):
        self._tokens = self.capacity
        self._timestamp = time.time()
        self._lock = threading.Lock()
    
    def acquire(self, tokens: int = 1) -> bool:
        with self._lock:
            now = time.time()
            self._tokens = min(
                self.capacity,
                self._tokens + (now - self._timestamp) * self.rate
            )
            self._timestamp = now
            if self._tokens >= tokens:
                self._tokens -= tokens
                return True
            return False


# === Main Engine ===

class SpatialDataEngine:
    
    def __init__(
        self,
        api_key: str,
        pool_size: int = 50,
        qps: float = 100,
        burst: float = 200,
        base_url: str = "https://restapi.amap.com/v3/geocode/geo",
    ):
        self.api_key = api_key
        self.base_url = base_url
        self._session = self._create_session(pool_size)
        self._limiter = TokenBucket(rate=qps, capacity=burst)
    
    def _create_session(self, pool_size: int) -> requests.Session:
        session = requests.Session()
        retry = Retry(
            total=5,
            backoff_factor=1,
            status_forcelist=[429, 500, 502, 503, 504],
            allowed_methods=["GET"],
        )
        adapter = HTTPAdapter(
            max_retries=retry,
            pool_connections=pool_size,
            pool_maxsize=pool_size,
        )
        session.mount("https://", adapter)
        session.mount("http://", adapter)
        return session
    
    def geocode(self, address: str, 
                timeout: tuple[float, float] = (3.0, 10.0)) -> dict[str, Any]:
        if not self._limiter.acquire():
            return {"error": "rate_limited", "retry_after": 1.0}
        
        try:
            resp = self._session.get(
                self.base_url,
                params={"address": address, "key": self.api_key},
                timeout=timeout,
            )
            resp.raise_for_status()
            return resp.json()
        except requests.exceptions.Timeout:
            return {"error": "timeout"}
        except requests.exceptions.ConnectionError as e:
            return {"error": "connection_failed", "detail": str(e)}
        except requests.exceptions.HTTPError as e:
            return {"error": "http_error", "status": e.response.status_code}
    
    @staticmethod
    def wgs84_to_gcj02(lng: float, lat: float) -> tuple[float, float]:
        if _out_of_china(lng, lat):
            return lng, lat
        dlat = _transform_lat(lng - 105.0, lat - 35.0)
        dlng = _transform_lng(lng - 105.0, lat - 35.0)
        rad_lat = lat / 180.0 * math.pi
        magic = 1 - _EE * math.sin(rad_lat) ** 2
        sqrt_magic = math.sqrt(magic)
        dlat = (dlat * 180.0) / ((_A * (1 - _EE)) / (magic * sqrt_magic) * math.pi)
        dlng = (dlng * 180.0) / (_A / sqrt_magic * math.cos(rad_lat) * math.pi)
        return lng + dlng, lat + dlat
    
    @staticmethod
    def gcj02_to_wgs84(lng: float, lat: float, 
                       tol: float = 1e-6, max_iter: int = 10) -> tuple[float, float]:
        wgs_lng, wgs_lat = lng, lat
        for _ in range(max_iter):
            pred_lng, pred_lat = SpatialDataEngine.wgs84_to_gcj02(wgs_lng, wgs_lat)
            dlng, dlat = pred_lng - lng, pred_lat - lat
            if abs(dlng) < tol and abs(dlat) < tol:
                break
            wgs_lng -= dlng
            wgs_lat -= dlat
        return wgs_lng, wgs_lat

关键要点

每一层在失败前都是隐形的。上面的代码解决了全部三个问题。

后续步骤

• 多提供商故障转移：错误时路由到备用 API
• 响应缓存：地址很少变动；应该积极缓存
• 异步处理：对于大批量地理编码，使用 asyncio + aiohttp
• 可观测性：将延迟、错误率、令牌利用率导出到监控系统

#地理编码 #python #分布式系统 #速率限制 #地理空间 #后端工程