如何从Python获取Neo4j中Cypher查询的服务端执行时间？

我明白你要的是排除网络传输、结果处理时间的纯服务器端计算耗时，用来做对比测试——用Python本地计时肯定不准，得从Neo4j本身的查询元数据里拿才靠谱。下面给你两种实用的实现方案：

方案1：利用Neo4j驱动的查询摘要（Summary） #

Neo4j的Python驱动会返回查询的执行摘要，里面包含服务器端的实际执行时间，这是最推荐的方式，不需要额外依赖。

完整修正后的代码 #

from neo4j import GraphDatabase  # 注意：neo4j.v1是旧版本写法，现在直接导入neo4j即可

# 初始化驱动
driver = GraphDatabase.driver('bolt://localhost:7687', auth=('neo4j', '1234'))
n_repeats = 3
graphname = "your_graph_name"  # 替换成你的实际图名
cypher = "MATCH (a)-[:{}*]-(b) WHERE ID(a) < ID(b) RETURN DISTINCT a, b".format(graphname + '_edges')

with driver.session() as session:
    # 先跑一次预热，避免第一次查询的初始化/缓存影响结果
    session.run(cypher).consume()
    
    total_server_time = 0
    for _ in range(n_repeats):
        # 执行查询并获取摘要
        result = session.run(cypher)
        summary = result.consume()  # 必须调用consume()才能拿到完整摘要
        # 提取服务器端执行时间（单位：毫秒）
        exec_time_ms = summary.result_available_after
        total_server_time += exec_time_ms
        print(f"第{_+1}次查询服务器耗时：{exec_time_ms}ms")
    
    avg_time_ms = total_server_time / n_repeats
    print(f"\n平均服务器执行耗时：{avg_time_ms:.2f}ms")

# 关闭驱动
driver.close()

关键说明 #

• summary.result_available_after：表示从查询发送到服务器，到第一个结果可用的时间，这就是纯服务器计算的耗时（不包含结果传输到客户端的时间）。
• 预热查询：Neo4j会缓存查询计划和部分结果，第一次查询通常会慢一些，所以先跑一次预热再统计，结果更准确。

方案2：使用APOC库的计时函数 #

如果你安装了Neo4j的APOC扩展，可以直接在Cypher里嵌入计时逻辑，返回执行时间：

代码示例 #

from neo4j import GraphDatabase

driver = GraphDatabase.driver('bolt://localhost:7687', auth=('neo4j', '1234'))
n_repeats = 3
graphname = "your_graph_name"
# 用apoc.util.measure包裹你的查询，返回执行时间和结果
cypher = """
CALL apoc.util.measure(
  "MATCH (a)-[:{}*]-(b) WHERE ID(a) < ID(b) RETURN DISTINCT a, b",
  {{}}
) YIELD value, time
RETURN value, time
""".format(graphname + '_edges')

with driver.session() as session:
    total_time = 0
    for _ in range(n_repeats):
        result = session.run(cypher).single()
        exec_time_ms = result["time"]
        total_time += exec_time_ms
        print(f"第{_+1}次查询耗时：{exec_time_ms}ms")
    
    print(f"\n平均耗时：{total_time/n_repeats:.2f}ms")

driver.close()

注意事项 #

• 需要先在Neo4j中安装APOC扩展，并且在neo4j.conf中开启dbms.security.procedures.unrestricted=apoc.*。
• 这个方法返回的time也是服务器端的执行时间，和方案1的结果基本一致。

避坑提醒 #

• 不要用Python的time.time()或time.perf_counter()直接计时，因为会包含网络往返、客户端处理结果的时间，完全不符合你的对比测试需求。
• 如果你的查询结果量很大，result.consume()会等待所有结果传输完成，但result_available_after依然是服务器端计算到第一个结果的时间，如果你需要的是整个查询的服务器总耗时（包括结果生成的全部时间），可以用summary.result_consumed_after，不过这个会包含一点结果传输的时间，需要根据你的需求选择。

内容的提问来源于stack exchange，提问作者J. Gambolputty