简介：Neo4jOSM 是一套面向 Java 开发者与图数据库学习者的开源路由服务示例，将高性能图数据库 Neo4j 与开放地图数据 OpenStreetMap 结合，用于构建基于地理位置的最短路径与路线规划功能。项目演示了从 OSM 文件解析路网、映射为 Neo4j 节点与关系，再到用 Cypher 查询计算路线的完整链路，适合想入门空间数据建模或为应用添加导航能力的开发者参考。资源包共 35 个文件，约 314KB，以 22 个 Java 源码为主体，辅以 3 个 osm 与 1 个 pbf 地图数据、2 个 properties 配置、2 个 md 说明文档，以及 Gradle 构建脚本、许可证和忽略文件，结构清晰便于按模块阅读。目前已有 169 人学习下载。通过源码可了解 OSM 数据预处理、图模型构建、Cypher 路径查询与 API 接口设计等关键实现，是研究图数据库处理复杂空间关系的实用起点。
1. 用 Neo4j 和 OpenStreetMap 搭一套能跑的路由服务，到底在解决什么问题
导航软件里输入起点终点，几百毫秒内弹出一条带转向指引的路线，背后其实是一张巨大的图在跑最短路。Neo4jOSM 这个方向，讲的就是把 OpenStreetMap 的路网数据塞进 Neo4j 图数据库，再在上面做路径规划。它解决的不是「造一个高德」，而是让你在自己的业务里拥有一套可控、可改、可解释的路由能力——比如给配送调度算里程、给景区做步行导航、给园区做内部路网分析。适合谁？手里有 OSM 数据、需要自定义权重（限高、限重、单行、时段禁行），又不想被商业地图 API 的配额和黑盒算法卡住的后端和算法同学。Neo4j 的图遍历天然贴合路网结构，OSM 又是免费且覆盖全球的底图，两者拼起来，就是一套能自己掌控的轻量路由服务。
2. 路网数据怎么进 Neo4j：从 OSM 原始文件到可查询的图
2.1 为什么选 Neo4j 而不是内存图或 PostGIS
做路由，第一反应往往是 NetworkX 或者 pgRouting。NetworkX 适合单机小图，几万个节点还行，上了百万级边就吃内存；pgRouting 依赖 PostGIS，SQL 写起来递归查询可读性差，动态权重改起来也麻烦。Neo4j 的优势在于：节点和关系都是一等公民，路口的转向限制、道路的通行属性可以直接挂成关系属性，Cypher 查最短路时还能顺手过滤条件。更关键的是，Neo4j 支持在关系上建索引，A* 或 Dijkstra 遍历时能快速定位邻接边。常见做法是：用 osm2pgsql 或 osmium 先把 OSM 的 .pbf 转成结构化数据，再批量导入 Neo4j。如果你只是做城市级路网（几十万节点），Neo4j 社区版单机足够；全国级路网建议先做区域裁剪，别一上来就全量灌。
2.2 用 osmium 抽取目标区域并转成 CSV
OSM 的原始格式是 .osm.pbf，直接读需要解析 XML，效率低。我一般先用 osmium 按边界裁出目标城市，再导出成节点和边的 CSV，方便 Neo4j 的 LOAD CSV 批量导入。
# 1. 下载目标区域的 pbf，例如某个城市的 extract # 2. 用 osmium 按 bbox 裁剪，减少数据量 osmium extract -b 116.20,39.80,116.60,40.05 china-latest.osm.pbf -o beijing.osm.pbf # 3. 导出所有节点（id, lon, lat） osmium cat beijing.osm.pbf -f opl -o beijing.opl # 4. 更推荐：用 osmium export 转成 GeoJSON，再用脚本拆成 nodes.csv 和 edges.csv osmium export beijing.osm.pbf -f geojson -o beijing.geojson上面命令里，-b后面的四个数字是 min_lon,min_lat,max_lon,max_lat，顺序别写反，否则裁出来是空文件。osmium export输出的 GeoJSON 里，每条 LineString 就是一段道路，属性里带 highway 类型、单行标志、名称等。接下来用 Python 把 GeoJSON 拆成两个 CSV：节点表存路口坐标，边表存路段起止节点和权重。
import json import csv with open('beijing.geojson', 'r', encoding='utf-8') as f: data = json.load(f) nodes = {} # node_id -> (lon, lat) edges = [] # (from_id, to_id, length, highway, oneway) def get_node_id(coord): # 用坐标字符串做去重键，实际项目建议用 geohash 或四舍五入到 1e-6 key = f"{coord[0]:.6f},{coord[1]:.6f}" if key not in nodes: nodes[key] = (coord[0], coord[1]) return key for feat in data['features']: geom = feat['geometry'] props = feat['properties'] if geom['type'] != 'LineString': continue coords = geom['coordinates'] highway = props.get('highway', 'unclassified') oneway = props.get('oneway', 'no') for i in range(len(coords) - 1): a = get_node_id(coords[i]) b = get_node_id(coords[i + 1]) # 粗略算长度，单位度，后续在 Neo4j 里再转米 length = ((coords[i][0] - coords[i+1][0])**2 + (coords[i][1] - coords[i+1][1])**2) ** 0.5 edges.append((a, b, length, highway, oneway)) with open('nodes.csv', 'w', newline='', encoding='utf-8') as f: w = csv.writer(f) w.writerow(['node_id', 'lon', 'lat']) for nid, (lon, lat) in nodes.items(): w.writerow([nid, lon, lat]) with open('edges.csv', 'w', newline='', encoding='utf-8') as f: w = csv.writer(f) w.writerow(['from_id', 'to_id', 'length', 'highway', 'oneway']) for e in edges: w.writerow(e)这段脚本的关键点：节点去重用的是坐标字符串，精度取到小数点后 6 位，大约 0.1 米，足够城市路网用。length这里算的是欧氏距离，单位是度，导入 Neo4j 后要乘上纬度修正系数转成米。oneway字段先原样保留，后面建关系时决定是否反向也建边。实际跑的时候，一个中等城市会生成几十万行 edges.csv，用LOAD CSV导入时记得加USING PERIODIC COMMIT，否则事务太大容易崩。
2.3 在 Neo4j 里建约束、导节点、导关系
CSV 准备好后，进 Neo4j Browser 或 cypher-shell 执行导入。先建唯一约束，保证 node_id 不重复，这样后续 MATCH 能走索引。
// 建唯一约束，加速节点查找 CREATE CONSTRAINT node_id_unique IF NOT EXISTS FOR (n:Node) REQUIRE n.node_id IS UNIQUE; // 导入节点，每 5000 行提交一次 LOAD CSV WITH HEADERS FROM 'file:///nodes.csv' AS row CALL { WITH row CREATE (:Node { node_id: row.node_id, lon: toFloat(row.lon), lat: toFloat(row.lat) }) } IN TRANSACTIONS OF 5000 ROWS; // 建关系索引，加速邻接边遍历 CREATE INDEX edge_from_idx IF NOT EXISTS FOR ()-[r:ROAD]-() ON (r.from_id); // 导入边，注意 oneway 处理 LOAD CSV WITH HEADERS FROM 'file:///edges.csv' AS row CALL { WITH row MATCH (a:Node {node_id: row.from_id}) MATCH (b:Node {node_id: row.to_id}) CREATE (a)-[:ROAD { length: toFloat(row.length), highway: row.highway, oneway: row.oneway }]->(b) } IN TRANSACTIONS OF 5000 ROWS;IN TRANSACTIONS OF 5000 ROWS是 Neo4j 5 的语法，4.x 用USING PERIODIC COMMIT 5000。导入完成后，跑一句MATCH ()-[r:ROAD]->() RETURN count(r)确认边数。如果边数明显少于 CSV 行数，多半是某些 from_id 或 to_id 在节点表里找不到，检查坐标精度是否一致。单行道的处理：如果oneway是yes或true，只建正向边；如果是no，正反都建。这一步我一般放在导入后单独跑一条 Cypher 补反向边，避免导入脚本里逻辑太绕。
3. 在 Neo4j 上跑最短路：Cypher 写法与性能调参
3.1 用 GDS 库跑 Dijkstra 和 A*
Neo4j 自带的最短路算法在 APOC 和 GDS（Graph Data Science）里都有。APOC 的apoc.algo.dijkstra适合快速验证，GDS 的gds.shortestPath.dijkstra性能更好，支持投影和权重属性。我一般先用 APOC 跑通逻辑，再换 GDS 做压测。
// 用 APOC 跑 Dijkstra，起点终点用 node_id 指定 MATCH (start:Node {node_id: '116.400000,39.900000'}) MATCH (end:Node {node_id: '116.500000,39.950000'}) CALL apoc.algo.dijkstra(start, end, 'ROAD>', 'length') YIELD path, weight RETURN path, weight;'ROAD>'表示只沿 ROAD 关系的出方向走，'length'是权重属性。如果路网里正反边都建了，这里不用改；如果只建了单向边，反向路径就搜不到。跑出来weight是度单位的累加值，要转成米得乘 111000 再乘纬度余弦。GDS 的写法多一步投影：
// 投影路网到内存图 CALL gds.graph.project( 'roadGraph', 'Node', 'ROAD', { relationshipProperties: 'length' } ); // 跑 Dijkstra MATCH (start:Node {node_id: '116.400000,39.900000'}) MATCH (end:Node {node_id: '116.500000,39.950000'}) CALL gds.shortestPath.dijkstra.stream('roadGraph', { sourceNode: start, targetNode: end, relationshipWeightProperty: 'length' }) YIELD index, sourceNode, targetNode, totalCost, nodeIds, costs, path RETURN totalCost, [nodeId IN nodeIds | gds.util.asNode(nodeId).node_id] AS route;GDS 投影后图常驻内存，重复查询不用每次扫磁盘，这是它比 APOC 快的主要原因。relationshipWeightProperty指定权重字段，totalCost就是路径总长度。注意 GDS 投影是快照，导入新数据后要重新投影或删掉旧图。
3.2 权重怎么设：距离、时间还是自定义代价
默认用length当权重，算出来的是最短距离，不是最快时间。真实路由里，时间权重更常用：time = length / speed，speed 按 highway 类型查表。常见做法是在导入边的时候多存一个speed属性，或者用 Cypher 的CASE WHEN动态算。
| highway 类型 | 默认速度 km/h | 说明 |
|---|---|---|
| motorway | 100 | 高速 |
| trunk | 80 | 快速路 |
| primary | 60 | 主干道 |
| secondary | 50 | 次干道 |
| residential | 30 | 居住区道路 |
| service | 20 | 内部道路 |
| footway | 5 | 步行道 |
// 给每条边补 speed 属性 MATCH ()-[r:ROAD]->() SET r.speed = CASE r.highway WHEN 'motorway' THEN 100 WHEN 'trunk' THEN 80 WHEN 'primary' THEN 60 WHEN 'secondary' THEN 50 WHEN 'residential' THEN 30 WHEN 'service' THEN 20 ELSE 5 END; // 用时间当权重跑最短路 MATCH (start:Node {node_id: '116.400000,39.900000'}) MATCH (end:Node {node_id: '116.500000,39.950000'}) CALL apoc.algo.dijkstra(start, end, 'ROAD>', 'length') YIELD path, weight WITH path, weight, relationships(path) AS rels RETURN weight AS distance, reduce(t = 0.0, r IN rels | t + r.length / r.speed) AS timeHours;上面这段先用距离跑出路径，再在结果里算时间，适合验证。真正要按时间最优，得把权重表达式直接传给算法，APOC 不支持表达式，GDS 可以用relationshipWeightProperty指向一个预先算好的timeCost属性。我一般导入时就生成timeCost = length / speed，省得查询时再算。
3.3 参数调优：索引、内存和并发
Neo4j 跑最短路，性能瓶颈通常在邻接边遍历和内存。几个必调参数：dbms.memory.heap.initial_size和dbms.memory.heap.max_size设成一样，避免动态扩缩；dbms.memory.pagecache.size给到数据文件大小的 1.5 倍。GDS 投影时用gds.graph.project的nodeProperties只带必要属性，别把 lon/lat 也投影进去，省内存。并发查询时，GDS 的concurrency参数控制线程数，默认是 CPU 核数，压测时从 4 开始往上调，观察 GC 日志。如果查询延迟突然飙高，先看dbms.logs.query.threshold设的多少，把慢查询打出来，多半是某个起点没走索引，全图扫了。
4. 避坑与排查：路网导入和路由查询里最容易翻车的 5 个点
4.1 现象：导入后边数远少于 CSV 行数，MATCH 找不到节点
原因：节点 CSV 里的 node_id 是坐标字符串，精度和边 CSV 里的不一致，比如一个保留 6 位一个保留 7 位，MATCH 时对不上。解决：生成 CSV 时统一用同一个格式化函数，导入前用sort -u比对两边 node_id 集合，差集就是问题节点。
4.2 现象：Dijkstra 跑出来路径绕远，明明有直连边
原因：单行道处理反了，或者oneway字段是-1（表示反向单行）没识别。OSM 里oneway=-1表示只能从 to 到 from 走。解决：导入时判断oneway为yes、true、1建正向，为-1建反向，为no、false、0建双向。跑一条MATCH ()-[r:ROAD]->() RETURN DISTINCT r.oneway看看有哪些值。
4.3 现象：GDS 投影报内存不足，图加载到一半失败
原因：全国路网节点上千万，GDS 默认堆内存不够。解决：先按区域裁剪，或者用gds.graph.project的nodeFilter只投影目标子图。另一个办法是调大gds.model.store相关配置，但最稳的还是别一次投影全量。
4.4 现象：查询延迟从几十毫秒涨到几秒，重启后恢复
原因：GDS 投影图没释放，多个查询叠加占满内存。解决：每次查询完调gds.graph.drop('roadGraph')，或者用gds.graph.exists判断后再投影。生产环境建议把投影做成定时任务，查询走缓存。
4.5 现象：路径权重是度，换算成米后里程明显偏大
原因：欧氏距离在经度方向没乘纬度余弦，高纬度地区误差能到 30%。解决：导入时用 Haversine 公式算真实距离，或者简单点，length_m = length_deg * 111000 * cos(lat)，lat 取路段中点纬度。别小看这个修正，做配送里程结算时差几公里就是真金白银。
5. 让路由服务真正可用：从 Cypher 查询到 HTTP 接口的最后一公里
跑通 Cypher 只是第一步，业务系统要的是 HTTP 接口。我一般用 FastAPI 包一层，把 Neo4j 驱动和查询逻辑封进去。下面是一个最小可用的路由接口，接收起点终点坐标，返回路径节点列表和总距离。
from fastapi import FastAPI, HTTPException from neo4j import GraphDatabase from pydantic import BaseModel import math app = FastAPI() driver = GraphDatabase.driver("bolt://localhost:7687", auth=("neo4j", "password")) class RouteRequest(BaseModel): start_lon: float start_lat: float end_lon: float end_lat: float def nearest_node(tx, lon, lat): # 找最近的节点，实际项目建议用空间索引或网格预计算 query = """ MATCH (n:Node) WITH n, point.distance(point({longitude: n.lon, latitude: n.lat}), point({longitude: $lon, latitude: $lat})) AS dist ORDER BY dist ASC LIMIT 1 RETURN n.node_id AS node_id """ result = tx.run(query, lon=lon, lat=lat) record = result.single() return record["node_id"] if record else None @app.post("/route") def route(req: RouteRequest): with driver.session() as session: start_id = session.execute_read(nearest_node, req.start_lon, req.start_lat) end_id = session.execute_read(nearest_node, req.end_lon, req.end_lat) if not start_id or not end_id: raise HTTPException(status_code=404, detail="no nearby node") query = """ MATCH (start:Node {node_id: $start_id}) MATCH (end:Node {node_id: $end_id}) CALL apoc.algo.dijkstra(start, end, 'ROAD>', 'length') YIELD path, weight RETURN [n IN nodes(path) | {lon: n.lon, lat: n.lat}] AS coords, weight * 111000 * cos(radians($mid_lat)) AS distance_m """ mid_lat = (req.start_lat + req.end_lat) / 2 result = session.run(query, start_id=start_id, end_id=end_id, mid_lat=mid_lat) record = result.single() if not record: raise HTTPException(status_code=404, detail="no route") return {"coords": record["coords"], "distance_m": record["distance_m"]}这段代码里，nearest_node用point.distance算球面距离找最近节点，数据量大时这个全表扫描会很慢，常见优化是预先按 geohash 建索引，或者用 Neo4j 的空间函数配合网格。/route接口返回坐标数组和米制距离，前端可以直接画线。注意apoc.algo.dijkstra要求 APOC 插件已安装，Neo4j Desktop 里在插件市场勾一下就行，社区版手动放 jar 包到 plugins 目录。
验证接口是否靠谱，我习惯用几个固定 case 跑回归：同一条路正反方向距离应该接近，跨区域长路径不能断，单行道逆行要绕路。把这些 case 写成 pytest，每次改完查询逻辑跑一遍，比手工点强。最后说个血泪经验：别在 Cypher 里做坐标转换和单位换算，能提前算好的属性就存成字段，查询时只做遍历和累加，这样延迟能压到几十毫秒。路由服务这东西，算法选型占三成，数据预处理和索引占七成，把导入环节做扎实，后面基本不用后悔药。希望帮到你。
本文还有配套的精品资源，点击获取
