使用Neo4j Java Driver创建节点及关系的性能优化与批量处理问询

针对你处理350k条数据时遇到的性能瓶颈，我来分享几个批量Cypher查询的优化方案，同时聊聊bolt协议+Java Driver的方案是否最优。

一、批量Cypher查询的核心优化手段 #

1. 使用UNWIND实现参数化批量导入 #

这是Neo4j官方推荐的批量处理方式，通过将多条数据打包成一个参数列表，用UNWIND遍历执行，大幅减少网络请求次数和查询解析开销，是提升批量性能最直接的手段。

示例代码：

// 构造批量数据列表，这里模拟添加350k条数据的参数Map
List<Map<String, Object>> batchData = new ArrayList<>();
for (int i = 0; i < 350000; i++) {
    Map<String, Object> params = new HashMap<>();
    params.put("personId", 1390 + i);
    params.put("obsId", 14001 + i);
    params.put("conceptId", 5978 + i);
    params.put("encounterId", 10810 + i);
    params.put("mlobsId", 2 + i);
    batchData.add(params);
}

// 批量Cypher语句，一次性处理整个数据列表
String batchCypher = """
    UNWIND $batchData AS data
    CREATE (m:MLObsTemp {
        personId: data.personId,
        obsId: data.obsId,
        conceptId: data.conceptId,
        MLObsId: data.mlobsId,
        encounterId: data.encounterId
    })
    WITH m, data
    MATCH (p:Person {personId: data.personId})
    CREATE (m)-[:PERSON]->(p)
    WITH m, data
    MATCH (e:Encounter {encounterId: data.encounterId})
    CREATE (m)-[:ENCOUNTER]->(e)
    WITH m, data
    MATCH (o:Obs {obsId: data.obsId})
    CREATE (m)-[:OBS]->(o)
    WITH m, data
    MATCH (c:Concept {conceptId: data.conceptId})
    CREATE (m)-[:CONCEPT]->(c)
""";

// 执行批量查询
try (Session session = driver.session()) {
    session.run(batchCypher, Map.of("batchData", batchData));
}

注意：单次批量大小建议控制在1000-5000条左右，根据服务器内存配置调整，避免批量过大导致内存溢出。

2. 借助APOC工具库的批量迭代功能 #

如果数据量极大（比如百万级），可以使用Neo4j的APOC扩展库中的apoc.periodic.iterate函数，它会自动将数据分批次处理，避免内存过载，还能开启并发提升速度。

首先确保已安装APOC扩展，然后通过Java Driver执行如下Cypher：

CALL apoc.periodic.iterate(
  "UNWIND $batchData AS data RETURN data",
  """
    CREATE (m:MLObsTemp {
        personId: data.personId,
        obsId: data.obsId,
        conceptId: data.conceptId,
        MLObsId: data.mlobsId,
        encounterId: data.encounterId
    })
    WITH m, data
    MATCH (p:Person {personId: data.personId})
    CREATE (m)-[:PERSON]->(p)
    WITH m, data
    MATCH (e:Encounter {encounterId: data.encounterId})
    CREATE (m)-[:ENCOUNTER]->(e)
    WITH m, data
    MATCH (o:Obs {obsId: data.obsId})
    CREATE (m)-[:OBS]->(o)
    WITH m, data
    MATCH (c:Concept {conceptId: data.conceptId})
    CREATE (m)-[:CONCEPT]->(c)
  """,
  {batchSize: 1000, parallel: true, params: {batchData: $batchData}}
)

parallel: true可开启并发处理，进一步提升效率，但要注意服务器的CPU和内存负载，避免资源耗尽。

3. 手动管理批量事务提交 #

将多条操作打包到一个事务中提交，减少事务开启/关闭的开销。Java Driver中可以通过Transaction对象手动控制事务边界：

try (Session session = driver.session()) {
    try (Transaction tx = session.beginTransaction()) {
        int batchSize = 1000;
        // 分批次提交事务
        for (int i = 0; i < batchData.size(); i += batchSize) {
            List<Map<String, Object>> subBatch = batchData.subList(i, Math.min(i + batchSize, batchData.size()));
            tx.run(batchCypher, Map.of("batchData", subBatch));
        }
        tx.commit();
    } catch (Exception e) {
        tx.rollback();
        throw e;
    }
}

二、关于Bolt协议+Java Driver的方案是否最优 #

首先可以明确：Bolt协议是Neo4j当前性能最优的连接协议，它是二进制协议，传输效率远高于HTTP协议；而官方Java Driver是兼容性最好、维护最及时的客户端，所以这个方案本身是最优选择之一，但你可以做以下优化让它更高效：

1. 升级Driver版本：你当前使用的1.6.1版本非常老旧（对应Neo4j 3.x系列），建议升级到与你的Neo4j服务器版本匹配的最新Driver（比如5.x系列），新版本修复了大量性能问题，支持更多批量优化特性。

2. 合理配置连接池：调整Driver的连接池参数，避免连接不足或资源浪费：

Driver driver = GraphDatabase.driver("bolt://localhost:7687", AuthTokens.basic("neo4j", "qwas"),
        Config.builder()
                .withMaxConnectionPoolSize(20) // 建议设置为服务器CPU核心数的2倍
                .withConnectionTimeout(Duration.ofSeconds(30))
                .withIdleTimeBeforeConnectionTest(Duration.ofMinutes(1))
                .build());

3. 用MERGE替代CREATE（如需去重）：如果你的数据可能存在重复，用MERGE代替CREATE避免重复创建节点/关系，同时MERGE会利用你已有的唯一约束，性能不会有明显损耗：

UNWIND $batchData AS data
MERGE (m:MLObsTemp {MLObsId: data.mlobsId}) // 用唯一标识作为匹配键
ON CREATE SET 
    m.personId = data.personId,
    m.obsId = data.obsId,
    m.conceptId = data.conceptId,
    m.encounterId = data.encounterId
WITH m, data
MERGE (m)-[:PERSON]->(p:Person {personId: data.personId})
// 其他关系同理用MERGE

三、额外的性能优化建议 #

• 验证索引使用情况：执行EXPLAIN你的Cypher语句，确认MATCH阶段使用了索引扫描（Index Seek）而非全表扫描（All Nodes Scan），确保你添加的唯一约束真正发挥作用。
• 减少返回数据：如果不需要返回创建的节点，去掉RETURN m，减少数据传输开销。
• 优化服务器配置：调整Neo4j的neo4j.conf，比如增大堆内存（dbms.memory.heap.max_size）、页缓存（dbms.memory.pagecache.size），提升服务器的处理能力。

内容的提问来源于stack exchange，提问作者Zakir saifi