高德地图API批量距离计算实战：从坐标转换到多线程并发

简介：基于高德地图API开发的批量地理位置距离计算工具，面向物流配送、路线规划与地理数据分析人员，解决成百上千地址点之间距离矩阵的高效计算问题。资源包共16个文件，约695KB，包含Java源码（Main.java、Data.java、Const.java）、编译后的class文件、fastjson-1.2.62.jar依赖、XML与IML工程配置文件，以及README.md、说明文件.txt和附赠资源.docx，源码与说明文档兼顾，便于直接运行和二次开发。已有176人学习记录表明该工具具备一定实用参考价值。文件内提供CSV文件导入导出接口，支持多线程并发处理以缩短计算时间，并可将距离结果可视化展示，适合物流调度中快速筛选最优路径。附带的说明文件与docx文档还介绍了高德地图API调用、地理编码等扩展思路，开发者可基于Distance-master工程进一步定制功能，提升地理位置数据批量处理的效率与精度。

1. 距离矩阵计算在物流里的价值，以及一个没算配额就开始写循环的翻车点

给两千个地址做批量地理位置距离计算，最先被点爆的往往不是代码，而是高德地图 API 的配额和费用。物流场景里的网点覆盖评估、顺路取件排线、干线拼车，本质上都要先算出一张“地址到地址”的距离矩阵。如果按一对一接口硬调，两千个地址就能产生上百万次请求。更常见的做法，是先把地址批量地理编码成坐标，再用距离测量接口一次算一批坐标到同一目的地的距离，配合多线程把吞吐拉起来，最后导回 CSV 并画到地图上。本文顺着这条链路，把参数、坑和能直接抄的写法讲清楚，适合做配送排线和运力测算的工程师。

2. 高德地图 API 距离计算原理与最小可用调用

高德地图 API 并不直接接收“地址”算距离，它接收的是经纬度坐标串。所以一条完整的批处理链路里，至少有两个调用合在一起：先用地理编码接口把文本地址转成坐标，再用距离测量接口基于坐标算路线距离。这两步的顺序、入参和返回结构都不同，分开理解才不会被返回的 status 字段绕晕。

距离测量接口本身是“一对多”的设计：一次请求里只放一个 destination，可以放最多 100 个起点坐标，返回结果里按 OriginID 对应每一个起点。这个设计直接决定了后面切片任务和多线程切分的方式，建议先把它跑通，再考虑并发和可视化。

2.1 地址到坐标：先地理编码再算距离，能省一次请求就省一次

地理编码是高德 Web 服务里的基础接口，负责把“北京市朝阳区阜通东大街 6 号”这类文本解析成“116.xxx,39.xxx”。在批量工具里，最好把地理编码结果落成一张本地坐标表，这样同一地址在后续跑批时直接命中缓存，不重复消耗接口配额。

import requests AMAP_KEY = "在高德开放平台申请的key" def geocode(address: str) -> dict | None: resp = requests.get( "https://restapi.amap.com/v3/geocode/geo", params={"key": AMAP_KEY, "address": address, "output": "json"}, timeout=10, ) payload = resp.json() if payload.get("status") != "1" or not payload.get("geocodes"): return None # geocodes 是候选列表，location 形如 "116.481028,39.989643" location = payload["geocodes"][0]["location"] return {"address": address, "location": location}

params里四个参数各管一件事：key做鉴权，address是要解析的完整地址，output固定 json，timeout是请求超时上限。返回的geocodes是数组，高德会按匹配度排候选，直接取第一个通常够用。如果地址里有明显的大区、商厦名，建议先拼成“省+市+区+详细地址”再提交，命中率比让接口自己猜高一些。

这里有个经常被忽略的细节：地理编码结果要按规范化后的地址文本缓存，而不是按 CSV 行号。去掉首尾空格、把中文数字转阿拉伯数字、统一“省/市/区”后缀，缓存命中率能提高 20% 以上，这对控制企业认证后的账单非常关键。

2.2 用距离测量接口一次算一批坐标到同一个目的地的距离

坐标就位后，就可以走 v3/distance 接口算距离。这个名字在高德文档里归属于路径规划大类，但用法更接近批量测量：origins里用分号拼接一串坐标，destination放目标坐标，返回每个起点到该目标的行驶距离和时间。

def batch_distance(origins: list[str], destination: str, type_code: int = 1): # 一次最多100个起点，超过要分片 if len(origins) > 100: raise ValueError("origins 每批不能超过100个") params = { "origins": ";".join(origins), "destination": destination, "type": type_code, "key": AMAP_KEY, } resp = requests.get("https://restapi.amap.com/v3/distance", params=params, timeout=15) data = resp.json() if data.get("status") != "1": raise RuntimeError(f"distance API error: {data.get('info')}") # results 列表顺序与 origins 传入顺序一一对应 return data["results"]

这段代码的逻辑非常直白：把坐标列表用分号拼成一个字符串，交给接口批量计算。调用方要做好的只有两件事——一是确保每个坐标都是“经度,纬度”的顺序，二是把超过 100 的列表按 100 切片循环调用。切片时要注意保留原始 id，否则多对多矩阵拼回去时会乱。

注意：v3/distance 是一对多接口，只支持一个 destination。如果要做 M 个目的地 × N 个起点，正确姿势是外层循环目的地，内层把起点按 100 分片，分片后形成完整的任务队列。

2.3 响应字段与过滤规则：光看 distance 不够，还要看 duration 和 info

接口返回的results数组里，每个元素都带有单条调用自己的状态。整体调用成功不代表每一条路线都计算成功，这是和地理编码最大的不同，也是最容易埋数据坑的地方。

type_code是影响结果口径的关键参数：1 对应驾车、2 对应骑行、3 对应步行、4 对应公交。物流场景里如果用的是货车，还要看路线规划接口是否有对应的车辆类型参数。直接拿驾车模式给卡车排线，容易低估红绿灯等待和绕行距离，最终算出来的矩阵和实际跑出来的油耗对不上。

3. CSV 文件导入导出与地理位置数据处理：把脏地址洗干净再送进 API

CSV 是这类工具最常见的输入输出形态，因为它兼容 Excel、数据库导入和各家的 WMS 导出。但 CSV 也是最容易藏脏数据的地方：字段带空格、省市区拆分不完整、编码在中文 Windows 下乱掉，都会让地理编码请求变成无效调用。所以第一件事不是写循环，而是先做数据处理。

3.1 用 pandas 读入 CSV 并清洗省市区字段

实际文件里地址往往不是一列，而是省、市、区、街道分开的几列。先读进来，把关键列拼成一个完整地址，同时把明显是空字符串的片段过滤掉。

import pandas as pd df = pd.read_csv("addresses.csv", dtype={"postcode": str}) df = df.fillna("") df["address"] = ( df["province"].str.strip() + df["city"].str.strip() + df["district"].str.strip() + df["street"].str.strip() ).str.replace(r"\s+", "", regex=True)

这个拼接是专门用来提高地理编码命中率的。高德按“省市区+街道”的结构做匹配，比直接拿用户填写的备注文本要稳定。str.strip清掉头尾空格，replace(r"\s+", "", regex=True)把地址中间的换行和制表符也一起去掉，避免一条地址被拆成两段。

导出侧同样有编码问题。默认的df.to_csv("result.csv")在 Excel 里打开会乱码，常见做法是加上encoding="utf-8-sig"，让 Excel 自动识别 UTF-8。如果后端还要继续喂给别的服务，建议同时保留一份不带 BOM 的 UTF-8 版本，防止某些解析器把\ufeff当成字段内容。

3.2 做坐标缓存，只对没坐标的地址调用地理编码

如果一批地址里有一半是重复网点，每次跑批都重新地理编码就是在烧钱。把坐标表做成本地缓存，跑批前先查缓存，查不到再走网络请求，这是最直接的降本手段。

cached = {} def get_location(row): key = row["address"] if key in cached: return cached[key] result = geocode(key) if result: lng, lat = result["location"].split(",") cached[key] = (float(lng), float(lat)) return cached.get(key) df["lng"] = None df["lat"] = None for idx, row in df.iterrows(): loc = get_location(row) if loc: df.at[idx, "lng"], df.at[idx, "lat"] = loc

缓存 key 用清洗后的地址文本，而不是 CSV 里的原始列。这样即使两份文件在“北京市朝阳区”后面一个带空格一个不带，也能命中同一份坐标。缓存的持久化可以简单 dump 成 json 文件，几千个地址也就是几 MB 的事，不需要上数据库。

处理完坐标后，距离计算阶段只用 lng/lat 两列，不再需要带着长地址跑来跑去。把这两列提前转成 float，还能节省后面拼接 origins 时的隐式转换开销。

3.3 坐标系的坑：高德返回 GCJ-02，别直接拿 WGS-84 算距离

GPS 设备直接出来的坐标是 WGS-84，而高德地图 API 使用的是一套加密偏移后的 GCJ-02 坐标系。两者在城市区域大概有几十到几百米的偏移，直接混用，轻则距离多算几百米，重则路线被误判到隔壁马路上。这个问题在纯地址导入时不会出现，但只要涉及到车载终端回传坐标，就必须处理。

常见做法是提供一套 WGS-84 到 GCJ-02 的转换函数，在导入 CSV 时就把所有坐标统一到 GCJ-02。下面是公开的偏移转换核心逻辑：

import math def _transform_lat(x, y): ret = -100.0 + 2.0 * x + 3.0 * y + 0.2 * y * y + 0.1 * x * y + 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, y): ret = 300.0 + x + 2.0 * y + 0.1 * x * x + 0.1 * x * y + 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

转换时只需要在导入侧统一一次：凡是 GPS 设备给的 WGS-84 坐标，进矩阵前先过一遍上述函数；凡是街道地址地理编码出来的坐标，保持原样，因为高德已经返回 GCJ-02。用同坐标系数据算出来的距离矩阵，才能在高德地图上画图时严丝合缝。

4. 多线程并发处理与配额控制：把上万对距离压缩进配额窗口

单线程跑完几万个距离请求，时间会很难看。假设一个分片请求耗时 0.8 秒，5000 个任务就是接近一个小时，这在调线路时完全不可接受。但多线程也不是把max_workers调到 32 就完事，核心矛盾是 API 的 QPS 限制和日配额。并发调度的目标，是在不触发限流的前提下尽量把配额窗口填满。

4.1 用 ThreadPoolExecutor 把任务按“目的地×100个起点”切分

任务切分粒度不应该是一对一的地址对，而应该是“一个目的地 + 一批最多 100 个起点”。理由很简单：每次调用都包含 HTTP 连接建立和鉴权开销，把调用次数压到最少，就是变相延长配额寿命。

from concurrent.futures import ThreadPoolExecutor, as_completed def build_tasks(destinations: list[str], origins: list[str], batch_size: int = 100): tasks = [] for dest in destinations: for i in range(0, len(origins), batch_size): tasks.append((dest, origins[i:i + batch_size])) return tasks def run_tasks(tasks, max_workers=8): results = [] with ThreadPoolExecutor(max_workers=max_workers) as pool: futures = { pool.submit(batch_distance, chunk, dest): (dest, chunk) for dest, chunk in tasks } for fut in as_completed(futures): dest, chunk = futures[fut] try: rows = fut.result() results.extend(rows) except Exception as e: print("task failed", dest, chunk[:2], e) return results

build_tasks先把每个目的地对应的起点列表切成 100 个一组的块，再把(dest, chunk)作为一个独立任务。这样每个任务在 worker 里只做一次 HTTP 调用，返回的最多 100 条结果天然带 OriginID，可以直接映射回原始地址。as_completed保证哪怕部分任务失败，也不阻塞已经完成的结果回写。

线程数选择有经验公式：max_workers约等于“允许的 QPS × 单请求耗时”。如果控制台显示 QPS 10，单请求平均 0.5 秒，那并发设为 5 附近就够用；继续加大只会触发限流，反而把成功的请求也拖进重试循环。

4.2 用令牌桶限流保护 key，避免触发 QPS 上限

高德对单个 key 的并发有明确限制，超了会返回调用频繁的错误码。为了避免把压力全部积压到重试逻辑里，我一般会在调用层加一个简单的令牌桶。它的作用不是减少总调用量，而是把请求均匀地散到每个时间窗口内，防止瞬时峰值把 key 打挂。

import threading import time class RateLimiter: def __init__(self, qps: float): self.interval = 1.0 / qps self.lock = threading.Lock() self.next_time = time.monotonic() def wait(self): with self.lock: now = time.monotonic() run_at = max(self.next_time, now) self.next_time = run_at + self.interval delay = run_at - now if delay > 0: time.sleep(delay)

使用方式是在每个 worker 里调用batch_distance之前先执行limiter.wait()。interval取 QPS 的倒数，线程各自排队领时间片，比粗暴地time.sleep(0.1)更均匀。这里要注意threading.Lock只保护时间片的分配，不保护网络请求本身，多线程之间不会互相阻塞 I/O。

4.3 失败重试与断点续跑：先把结果落盘，再回头追异常

并发跑批时最怕的是跑到一半进程崩掉，前面积累的结果全丢。所以正确顺序是“先落盘，再处理”，任务每完成一个分片就往 CSV 追加一行，不等到全部计算完成再做一次性导出。

import csv import json with open("distance_matrix.csv", "a", newline="", encoding="utf-8-sig") as f: writer = csv.writer(f) writer.writerow(["destination", "origin_id", "distance", "duration"]) for dest, chunk in tasks: try: rows = batch_distance(chunk, dest) for r in rows: writer.writerow([dest, r.get("origin_id"), r.get("distance"), r.get("duration")]) f.flush() except Exception as e: with open("errors.jsonl", "a", encoding="utf-8") as ef: ef.write(json.dumps({"destination": dest, "error": str(e)}) + "\n")

flush()在这里很关键，它把缓冲区的数据强制写进磁盘，进程被杀也能保住已完成的部分。每次重跑时，程序可以先把errors.jsonl里的失败任务读回来重新入队，已经完成的 destination 直接跳过，达到断点续跑的效果。

错误处理按类型分流，常见的有这么几类：

网上总有人说“高德地图 API 收费坑人”，我观察到的原因多半不是单价贵，而是把“调用次数”当成了“地址条数”来规划成本。一个包含 200 个目的地的矩阵，实际 API 调用量是 200 乘以分片数，再加上地理编码和重试系数。想控制账单，就在开头把请求次数估算清楚，别等月底看账单才反应过来。

5. 结果可视化展示与物流调度核查：用地图把距离矩阵画出来

距离矩阵算完，光看 CSV 很难发现哪条路线异常。可视化不是为了好看，而是为了核查：哪条线路距离异常、哪个目的地覆盖区域明显偏移，地图上一眼就能看出来。

5.1 把矩阵导出成前端可读的 JSON

CSV 适合后续系统处理，但网页端地图渲染更适合 JSON。把每一条计算结果整理成独立的记录，每条包含起点、终点、距离和耗时，前端不用再做字段映射。

import json df = pd.read_csv("distance_matrix.csv") records = df.rename(columns={"destination": "dest"}).to_dict(orient="records") with open("matrix.json", "w", encoding="utf-8") as f: json.dump(records, f, ensure_ascii=False)

ensure_ascii=False是为了让 JSON 里保留中文地址原样，不然前端显示出来的都是\u转义符，调试体验很差。如果起点坐标还没有在地理编码阶段存下来，这一步要回原表做一次关联，否则画不了线。

5.2 用高德 JS API 渲染散点与分级连线

浏览器端引入高德 JavaScript API 后，核心绘图逻辑很简单：把每条记录的起点和终点连成一条折线，再按距离阈值做颜色分级。短途用亮色，超过阈值的长途用浅色，这样能避免满屏都是线、看不出主次。

records.forEach(r => { const line = new AMap.Polyline({ path: [[r.origin_lng, r.origin_lat], [r.dest_lng, r.dest_lat]], strokeColor: r.distance > 30000 ? 'rgba(180,180,180,0.4)' : '#2c9fc8', strokeWeight: 1.5 }); map.add(line); });

折线路径的数组顺序是[经度, 纬度]，和高德 API 的坐标格式一致，不要写成[纬度, 经度]，否则画出来的线路会直接横跨半个中国。实际做物流调度可视化时，我还会再叠一层圆点标记，半径按目的地聚合的订单量缩放，这样既能看距离，又能看货量分布。

5.3 随机抽样校验距离矩阵

最后给整个计算流程留一个校验步骤：从 CSV 里随机抽 10 条记录，手动打开高德地图，把同样的起终点输入路线查询，对比 API 返回的 distance 和页面显示的距离。如果两者误差在 5% 以内，说明坐标、type、坐标系全部正确；如果超过 5%，优先查两件事——起点坐标是否被反向赋值，以及经纬度是否混用了其他坐标系的 GPS 数据。这一样本量虽然小，但足够暴露大多数批处理工具的通病，比再跑一次全量便宜得多。

本文还有配套的精品资源，点击获取