C#中如何在同步方法内正确等待依赖的Neo4j批量加载任务

问题背景 #

• 业务需要向Neo4j批量加载两组存在强依赖的数据：第二组数据的写入依赖第一组数据创建完成的节点与关系，必须保证第一组数据完全提交落库后，才能启动第二组数据的加载
• 代码限制：Import()方法无法修改为async异步签名，不能使用await关键字
• 现有实现偶发运行时错误：

Cannot access records on this result any more as the result has already been consumed or the query runner where the result is created has already been closed.

• 直接调用任务Wait()方法的缺陷：当外层任务状态为RanToCompletion时，内部查询执行、结果消费的任务仍可能处于Faulted状态（比如Cypher语法错误场景），无法直接捕获查询执行阶段的异常，需要额外遍历检查异常集合，实现冗余且容易漏错。

现有问题实现代码如下：

public override void Import()
{
    using var session = _driver.AsyncSession(
        x => x.WithDefaultAccessMode(AccessMode.Write));

    var groupA = session.WriteTransactionAsync(async x =>
    {
        var result = await x.RunAsync(queryA);
        return result.ToListAsync();
    });
    groupA.Result.Wait();

    var groupB = session.WriteTransactionAsync(async x =>
    {
        var result = await x.RunAsync(queryB);
        return result.ToListAsync();
    });
    groupB.Result.Wait();
}

问题根因 #

现有实现的问题集中在两点：

1. 事务回调中返回了未完成的嵌套结果任务：result.ToListAsync()本身是一个独立的Task，WriteTransactionAsync虽然会自动解包一层返回值，但groupA.Result.Wait()只会等待到外层任务拿到内层Task的节点，不会等待内层结果消费任务执行完成。这就会触发竞态：事务已经提交关闭、结果游标被释放时，内层结果遍历逻辑可能还在执行，直接抛出结果已释放的错误。
2. 异常捕获链路断裂：嵌套Task抛出的异常不会被外层任务直接抛出，直接Wait外层任务时，哪怕Cypher语法错误、查询执行失败，外层任务也可能显示RanToCompletion状态，必须手动检查内层Task的异常属性才能定位问题。

推荐实现方案 #

核心原则是：在事务回调内部完成所有结果消费逻辑，使用GetAwaiter().GetResult()同步等待完整任务链执行完成，既保证执行顺序，又能完整捕获全链路异常。

public override void Import()
{
    using var session = _driver.AsyncSession(cfg => 
        cfg.WithDefaultAccessMode(AccessMode.Write));

    // 执行第一组数据写入
    var groupATask = session.WriteTransactionAsync(async tx =>
    {
        var resultCursor = await tx.RunAsync(queryA);
        // 必须在事务委托内完成所有结果操作，不要返回未完成的结果Task
        // 如果不需要读取返回记录，替换为await resultCursor.ConsumeAsync()即可，性能更高
        var records = await resultCursor.ToListAsync();
        return records;
    });
    // 同步等待第一组全链路完成：事务提交、结果消费全部执行完，异常会直接抛出
    groupATask.GetAwaiter().GetResult();

    // 第一组完全落库后再启动第二组写入
    var groupBTask = session.WriteTransactionAsync(async tx =>
    {
        var resultCursor = await tx.RunAsync(queryB);
        // 批量写入场景不需要拿记录时，优先用ConsumeAsync，减少内存占用
        await resultCursor.ConsumeAsync();
    });
    groupBTask.GetAwaiter().GetResult();
}

关键注意事项 #

• 所有结果游标操作必须在事务回调内完成：ToListAsync()/ConsumeAsync()/逐行读取等和查询结果相关的逻辑，必须在传给WriteTransactionAsync的委托内部用await执行完，不要把结果游标、未完成的结果Task返回到事务外部，从根源避免事务关闭后访问已释放资源的问题。
• 同步等待优先使用GetAwaiter().GetResult()：相比Wait()或者直接访问.Result，该方式和await的异常行为完全一致，会直接抛出任务链上的原始异常，不会包装成AggregateException，不需要手动解包遍历异常集合，能直接捕获Cypher语法错误、事务提交失败、连接异常等所有错误。
• 批量写入场景优先用ConsumeAsync()：如果不需要读取查询返回的具体记录，不要用ToListAsync()缓存所有返回结果，ConsumeAsync只会拉取查询执行的统计信息，内存占用更低、执行速度更快。
• 同一会话内的事务天然保证顺序：Neo4j驱动会保证同一个异步会话内的事务按启动顺序串行执行，只要等前一个事务的完整任务（提交+结果消费）执行完成再启动下一个事务，就不会出现第二组写入看不到第一组数据的一致性问题。

内容的提问来源于stack exchange，提问作者Dr. Strangelove