springboot + neo4j + gremlin 排错

起因：

1. 我们在 gremlin 的配置类中向springboot 分别注入 Neo4JGraph 和 GraphTraversalSource 两个bean

@Configuration
public class GraphDatabaseInitConfig {

    @Value("${spring.data.neo4j.password}")
    private String neo4jPassword;
    @Value("${spring.data.neo4j.username}")
    private String neo4jUsername;
    @Value("${spring.data.neo4j.uri}")
    private String neo4jUri;

    @Bean
    public Neo4JGraph createNeo4jGraph() {
        Driver driver = GraphDatabase.driver(neo4jUri, AuthTokens.basic(neo4jUsername, neo4jPassword));
        Neo4JNativeElementIdProvider vertexIdProvider = new Neo4JNativeElementIdProvider();
        Neo4JNativeElementIdProvider edgeIdProvider = new Neo4JNativeElementIdProvider();
        Neo4JGraph neo4JGraph = new Neo4JGraph(driver, null, vertexIdProvider, edgeIdProvider);

        return neo4JGraph;
    }

    @Bean
    public GraphTraversalSource createGraphTraversalSource() {
        return createNeo4jGraph().traversal();
    }

}

2. 在不修改 neo4j 数据的情况下，查询数据一直正常
3. 在修改玩 neo4j 数据之后，再次查询，刚开始是正确的数据，多查几次后发现数据又变成了修改之前的数据

pom.xml 文件依赖

 <dependency>
            <groupId>com.steelbridgelabs.oss</groupId>
            <artifactId>neo4j-gremlin-bolt</artifactId>
            <version>0.4.6</version>
        </dependency>

排查过程

1. 查看查询的源码，断点跟进，没有发现问题
2. 查看 neo4j-gremlin-bolt jar包内的类，打断点跟进发现数据不变的情况下，每次查询 session（Neo4JGraph.currentSession()） 都会重新创建，但是当数据发生变动之后，session 会从之前所有查询历史中选择某一个 session 去进行处理。
3. 上面的是假象，其实数据不变的情况下，即使获取到之前的session ，数据也完全一样，查询结果也一样。

原因分析

1. Neo4JGraph 的历史 session 存储在 ThreadLocal 的map中，但查询完成后线程没有杀掉。
2. springboot 的线程也就是一个 Bean 持续存在，当某次查询跟之前查询属于同一个线程的情况下，Neo4JGraph.currentSession() 会从 ThreadLocal 的map中读取 session 信息， 也就是不会再创建新的 session。

解决方案

1. 在每次调用查询之后主动关闭

neo4JGraph.close()

2. aop 切片，在每次请求之后，执行session 关闭

补充 Neo4JGraph 流程例如：

graphTraversalSource.V().has(LabelNameConstant.EQUIP, "name", equipTypeName).toList()

V() 和 has() 会记录成 step
toList 之后才会真正触发查询

this.fill(new ArrayList<>())

Traversal.class

while (true) {
                final Traverser<E> traverser = endStep.next();
                TraversalHelper.addToCollection(collection, traverser.get(), traverser.bulk());
            }

GraphStep.class

this.iteratorSupplier = () -> (Iterator<E>) (Vertex.class.isAssignableFrom(this.returnClass) ?
                this.getTraversal().getGraph().get().vertices(this.ids) :
                this.getTraversal().getGraph().get().edges(this.ids));

this.getTraversal().getGraph().get().vertices(this.ids) 

Neo4JGraph.class

@Override
    public Iterator<Vertex> vertices(Object... ids) {
        // get current session
        Neo4JSession session = currentSession();
        // transaction should be ready for io operations
        transaction.readWrite();
        // find vertices, 此处第一次查询时，会从数据库进行match
        return session.vertices(ids);
    }

Neo4JSession.class

public Iterator<Vertex> vertices(Object[] ids) {
        Objects.requireNonNull(ids, "ids cannot be null");
        // verify identifiers
        verifyIdentifiers(Vertex.class, ids);
        // check we have all vertices already loaded ,此处判断是否是第一次查询
        if (!verticesLoaded) {
            // check ids
            if (ids.length > 0) {
                // parameters as a stream
                Set<Object> identifiers = Arrays.stream(ids).map(id -> processIdentifier(vertexIdProvider, id)).collect(Collectors.toSet());
                // filter ids, remove ids already in memory (only ids that might exist on server)
                List<Object> filter = identifiers.stream().filter(id -> !vertices.containsKey(id) && !transientVertexIndex.containsKey(id)).collect(Collectors.toList());
                // check we need to execute statement in server
                if (!filter.isEmpty()) {
                    // vertex match predicate
                    String predicate = partition.vertexMatchPredicate("n");
                    // change operator on single id filtering (performance optimization)
                    if (filter.size() == 1) {
                        // execute statement
                        Result result = executeStatement("MATCH " + generateVertexMatchPattern("n") + " WHERE " + vertexIdProvider.matchPredicateOperand("n") + " = $id" + (predicate != null ? " AND " + predicate : "") + " RETURN n", Collections.singletonMap("id", filter.get(0)));
                        // create stream from query
                        Stream<Vertex> query = vertices(result);
                        // combine stream from memory and query result
                        Iterator<Vertex> iterator = combine(Stream.concat(identifiers.stream().filter(vertices::containsKey).map(id -> (Vertex)vertices.get(id)), identifiers.stream().filter(transientVertexIndex::containsKey).map(id -> (Vertex)transientVertexIndex.get(id))), query);
                        // process summary (query has been already consumed by combine)
                        ResultSummaryLogger.log(result.consume());
                        // return iterator
                        return iterator;
                    }
                    // execute statement
                    Result result = executeStatement("MATCH " + generateVertexMatchPattern("n") + " WHERE " + vertexIdProvider.matchPredicateOperand("n") + " IN $ids" + (predicate != null ? " AND " + predicate : "") + " RETURN n", Collections.singletonMap("ids", filter));
                    // create stream from query
                    Stream<Vertex> query = vertices(result);
                    // combine stream from memory and query result
                    Iterator<Vertex> iterator = combine(Stream.concat(identifiers.stream().filter(vertices::containsKey).map(id -> (Vertex)vertices.get(id)), identifiers.stream().filter(transientVertexIndex::containsKey).map(id -> (Vertex)transientVertexIndex.get(id))), query);
                    // process summary (query has been already consumed by combine)
                    ResultSummaryLogger.log(result.consume());
                    // return iterator
                    return iterator;
                }
                // no need to execute query, only items in memory
                return combine(identifiers.stream().filter(vertices::containsKey).map(id -> (Vertex)vertices.get(id)), identifiers.stream().filter(transientVertexIndex::containsKey).map(id -> (Vertex)transientVertexIndex.get(id)));
            }
            // vertex match predicate
            String predicate = partition.vertexMatchPredicate("n");
            // execute statement
            Result result = executeStatement("MATCH " + generateVertexMatchPattern("n") + (predicate != null ? " WHERE " + predicate : "") + " RETURN n", Collections.emptyMap());
            // create stream from query
            Stream<Vertex> query = vertices(result);
            // combine stream from memory (transient) and query result
            Iterator<Vertex> iterator = combine(transientVertices.stream().map(vertex -> (Vertex)vertex), query);
            // process summary (query has been already consumed by combine)
            ResultSummaryLogger.log(result.consume());
            // it is safe to update loaded flag at this time
            verticesLoaded = true;
            // return iterator
            return iterator;
        }
        // check ids
        if (ids.length > 0) {
            // parameters as a stream (set to remove duplicated ids)
            Set<Object> identifiers = Arrays.stream(ids).map(id -> processIdentifier(vertexIdProvider, id)).collect(Collectors.toSet());
            // no need to execute query, only items in memory
            return combine(identifiers.stream().filter(vertices::containsKey).map(id -> (Vertex)vertices.get(id)), identifiers.stream().filter(transientVertexIndex::containsKey).map(id -> (Vertex)transientVertexIndex.get(id)));
        }
        // no need to execute query, all items in memory， 后面的数据全部从内存中备份出来
        return combine(transientVertices.stream().map(vertex -> (Vertex)vertex), vertices.values().stream().map(vertex -> (Vertex)vertex));
    }