Google Maps Geocoding API批量调用能否提升13k+城市坐标查询性能？

阿华AIGC实验室

2026-5-20

关于Google Maps Geocoding API批量查询性能提升的解答

首先直接给结论：用你的batcher函数分组+并行请求的方式，肯定能大幅提升查询速度，但要注意正确的实现细节和Google API的限制，下面我给你拆解清楚：

为什么单条查询这么慢？

单条串行查询的问题在于，每一次请求都要经历「建立HTTP连接→发送请求→等待服务器响应→解析结果」的完整流程，100条就是100次重复这个过程，时间是线性累加的。而批量+并行的方式，可以把多个请求的等待时间重叠起来，总耗时会大幅降低。

你的`batcher`函数的作用

这个生成器函数只是帮你把大列表拆成固定大小的批次（比如每10条一组），这是批量处理的第一步，但它本身不会提升性能——关键是你怎么处理这些批次：

如果还是串行处理每个批次（先等前10条全部返回，再处理下10条），性能提升有限；
如果是并行发起每个批次内的请求（用异步IO或多线程），才能真正把速度提上去。

必须注意的Google API限制

Google Maps Geocoding API有严格的配额和速率限制，不注意的话可能会触发限流（429错误）甚至封号：

配额限制：免费版每天最多2500次请求，13000+城市肯定不够用，你需要升级到付费套餐；
速率限制：即使付费，也有每秒请求数（QPS）限制（默认一般是50左右，具体看你的套餐），并行请求的数量不能超过这个阈值；
批量请求的特殊说明：Google其实有专门的「批量地理编码服务」针对超大量地址的场景，如果你要处理13000+条数据，也可以考虑这个服务，效率会更高。

优化实现的建议

结合你的batcher函数，给你几个具体的优化方向：

1. 用异步IO并行处理请求

Python里可以用aiohttp+asyncio来实现异步请求，把批次内的请求同时发出去，示例代码如下：

import aiohttp
import asyncio

def batcher(sequence, batch_size):
    return (sequence[position:position + batch_size] for position in range(0, len(sequence), batch_size))

async def geocode_single_address(session, address, api_key):
    # 注意对地址做URL编码，避免特殊字符出错
    encoded_address = aiohttp.helpers.urlquote(address)
    url = f"https://maps.googleapis.com/maps/api/geocode/json?address={encoded_address}&key={api_key}"
    async with session.get(url) as response:
        if response.status == 429:
            # 遇到限流，等待后重试（简单实现，实际可以加指数退避）
            await asyncio.sleep(2)
            return await geocode_single_address(session, address, api_key)
        return await response.json()

async def process_batch(session, batch, api_key):
    # 并行发起批次内的所有请求
    tasks = [geocode_single_address(session, addr, api_key) for addr in batch]
    return await asyncio.gather(*tasks, return_exceptions=True)

async def main(address_list, api_key, batch_size=10):
    async with aiohttp.ClientSession() as session:
        all_results = []
        for batch in batcher(address_list, batch_size):
            batch_results = await process_batch(session, batch, api_key)
            all_results.extend(batch_results)
            # 加个小延迟，避免触发速率限制
            await asyncio.sleep(0.1)
        return all_results

if __name__ == "__main__":
    YOUR_API_KEY = "你的API密钥"
    city_list = ["北京", "上海", ...]  # 你的13000+城市列表
    final_results = asyncio.run(main(city_list, YOUR_API_KEY))

2. 加入缓存机制

很多城市名称可能重复，提前用字典缓存已经查询过的坐标，避免重复请求：

cache = {}

async def geocode_single_address(session, address, api_key):
    if address in cache:
        return cache[address]
    # 后面的查询逻辑...
    result = await response.json()
    cache[address] = result
    return result