本地TinkerPop Gremlin事务并发读取冲突问题咨询

问题背景 #

在本地TinkerPop Gremlin环境测试以下场景时遇到冲突错误：

• 初始添加若干顶点
• 启动事务，删除并重新添加这些顶点
• 事务执行期间，用独立Go协程持续读取顶点

错误信息：
code:244 message:Conflict: element modified in another transaction attributes

疑问：读取操作难道不应该基于快照数据吗？

问题原因 #

Gremlin的事务隔离逻辑和底层存储引擎强绑定，并非所有兼容图数据库都默认提供快照隔离：

1. 默认隔离级别限制：多数图数据库（如JanusGraph、Neo4j）默认采用**读已提交（Read Committed）**隔离级别，而非快照隔离。该级别下，读操作会获取当前最新的已提交数据，但如果另一个事务正在修改同一元素（比如你代码中用固定ID Foo/Bar重新添加顶点），就会触发写-读冲突。
2. 固定ID的风险：你在addVertices中硬编码了顶点ID，事务内删除后重新添加同一ID的顶点，本质是修改该ID对应的元素；协程的读操作在事务未提交时尝试访问这个被修改的元素，直接触发冲突。
3. 连接复用问题：Gremlin Go驱动的DriverRemoteConnection默认复用连接，读协程和事务可能共享同一连接上下文，导致读操作感知到未提交的事务修改，进而触发冲突。

解决方案 #

1. 调整隔离级别（若数据库支持）：如果使用的图数据库支持快照隔离（比如JanusGraph可通过配置storage.transaction.isolation=snapshot开启），修改配置启用快照隔离，让读操作基于事务开始时的快照，避免冲突。
2. 避免固定ID重复使用：不要在事务内删除后重建同一ID的顶点，改用数据库自动生成唯一ID的方式，或直接修改元素属性而非删除重建。
3. 为读操作创建独立连接：给读协程单独创建DriverRemoteConnection，确保读操作在独立上下文执行，不受事务连接影响：

   // 在main函数中为读协程创建独立客户端
   readClient, err := gremlingo.NewDriverRemoteConnection("ws://localhost:8182/gremlin")
   if err != nil {
       log.Fatalf("Failed to create read client: %v", err)
   }
   defer readClient.Close()
   readG := gremlingo.Traversal_().WithRemote(readClient)
   // 启动协程时传入readG
   go func() {
       defer wg.Done()
       displayVertices(readG, done)
   }()
   

4. 增加读操作重试机制：如果无法修改隔离级别或连接方式，可以在读操作遇到冲突错误时自动重试，直到事务提交或回滚完成。

优化后的测试代码 #

package main

import (
    "fmt"
    "log"
    "sync"
    "time"

    gremlingo "github.com/apache/tinkerpop/gremlin-go/v3/driver"
)

func main() {
    // 主操作客户端
    client, err := gremlingo.NewDriverRemoteConnection("ws://localhost:8182/gremlin")
    if err != nil {
        log.Fatalf("Failed to create client: %v", err)
    }
    defer client.Close()

    // 读操作独立客户端
    readClient, err := gremlingo.NewDriverRemoteConnection("ws://localhost:8182/gremlin")
    if err != nil {
        log.Fatalf("Failed to create read client: %v", err)
    }
    defer readClient.Close()

    g := gremlingo.Traversal_().WithRemote(client)
    readG := gremlingo.Traversal_().WithRemote(readClient)

    // 清理现有顶点
    errChan := g.V().HasLabel("person").Drop().Iterate()
    for err := range errChan {
        if err != nil {
            panic(fmt.Errorf("dropping all vertices: %w", err))
        }
    }

    fmt.Println("Vertices dropped")

    // 初始添加顶点
    addVertices(g)
    fmt.Println("Vertices added")

    done := make(chan struct{})
    var wg sync.WaitGroup
    wg.Add(1)
    // 用独立读客户端启动协程
    go func() {
        defer wg.Done()
        displayVertices(readG, done)
    }()

    // 启动事务
    transaction := g.Tx()
    traversal, err := transaction.Begin()
    if err != nil {
        fmt.Println(err)
        return
    }

    fmt.Println("Transaction started")
    // 事务内删除顶点
    errChan = traversal.GetGraphTraversal().V().HasLabel("person").Drop().Iterate()
    for err := range errChan {
        if err != nil {
            fmt.Printf("Error dropping vertices in transaction: %v\n", err)
            err2 := transaction.Rollback()
            if err2 != nil {
                fmt.Printf("Rollback failed: %v\n", err2)
            }
            return
        }
    }

    // 事务内添加顶点（建议改用自动生成ID）
    addVertices(traversal)

    // 提交事务
    errCommit := transaction.Commit()
    if errCommit != nil {
        fmt.Printf("Committing transaction failed: %v\n", errCommit)
    } else {
        fmt.Println("Transaction committed")
    }

    close(done)
    wg.Wait()
}

func displayVertices(g *gremlingo.GraphTraversalSource, done <-chan struct{}) {
    for {
        select {
        case <-done:
            fmt.Println("Stopping vertex display.")
            return
        default:
            query := g.V().HasLabel("person").Id()
            result, err := query.ToList()
            if err != nil {
                log.Printf("Error retrieving vertices: %v", err)
                time.Sleep(time.Second)
                continue
            }
            fmt.Println("Current vertices IDs:")
            for _, res := range result {
                fmt.Printf("- %v\n", res.Data.(string))
            }
            time.Sleep(500 * time.Millisecond) // 降低轮询频率减少冲突概率
        }
    }
}

func addVertices(g *gremlingo.GraphTraversalSource) {
    // 注意：固定ID易引发冲突，建议改为数据库自动生成ID
    traversal := g.GetGraphTraversal().AddV("person").Property(gremlingo.T.Id, "Foo").Fold().AddV("person").Property(gremlingo.T.Id, "Bar").Fold()
    fmt.Printf("Executing traversal: %s\n", Translate(traversal))
    errChan := traversal.Iterate()
    for err := range errChan {
        if err != nil {
            fmt.Printf("Error adding vertices: %v\n", err)
        }
    }
}

func Translate(traversal *gremlingo.GraphTraversal) string {
    if traversal == nil || traversal.Bytecode == nil {
        return ""
    }
    translator := gremlingo.NewTranslator("g")

    queryTranslation, err := translator.Translate(traversal.Bytecode)
    if err != nil {
        panic(err)
    }

    return queryTranslation
}

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