起因：
- 我們?cè)?gremlin 的配置類(lèi)中向springboot 分別注入 Neo4JGraph 和 GraphTraversalSource 兩個(gè)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();
}
}
- 在不修改 neo4j 數(shù)據(jù)的情況下，查詢(xún)數(shù)據(jù)一直正常
- 在修改玩 neo4j 數(shù)據(jù)之后铐然，再次查詢(xún)，剛開(kāi)始是正確的數(shù)據(jù)趋距，多查幾次后發(fā)現(xiàn)數(shù)據(jù)又變成了修改之前的數(shù)據(jù)
pom.xml 文件依賴(lài)
<dependency>
<groupId>com.steelbridgelabs.oss</groupId>
<artifactId>neo4j-gremlin-bolt</artifactId>
<version>0.4.6</version>
</dependency>
排查過(guò)程
- 查看查詢(xún)的源碼，斷點(diǎn)跟進(jìn)搪桂，沒(méi)有發(fā)現(xiàn)問(wèn)題
- 查看 neo4j-gremlin-bolt jar包內(nèi)的類(lèi)寒砖，打斷點(diǎn)跟進(jìn)發(fā)現(xiàn)數(shù)據(jù)不變的情況下，每次查詢(xún) session（Neo4JGraph.currentSession()） 都會(huì)重新創(chuàng)建映九，但是當(dāng)數(shù)據(jù)發(fā)生變動(dòng)之后，session 會(huì)從之前所有查詢(xún)歷史中選擇某一個(gè) session 去進(jìn)行處理瞎颗。
- 上面的是假象件甥，其實(shí)數(shù)據(jù)不變的情況下，即使獲取到之前的session 哼拔，數(shù)據(jù)也完全一樣引有，查詢(xún)結(jié)果也一樣。
原因分析
- Neo4JGraph 的歷史 session 存儲(chǔ)在 ThreadLocal 的map中倦逐，但查詢(xún)完成后線(xiàn)程沒(méi)有殺掉譬正。
- springboot 的線(xiàn)程也就是一個(gè) Bean 持續(xù)存在，當(dāng)某次查詢(xún)跟之前查詢(xún)屬于同一個(gè)線(xiàn)程的情況下，Neo4JGraph.currentSession() 會(huì)從 ThreadLocal 的map中讀取 session 信息曾我， 也就是不會(huì)再創(chuàng)建新的 session粉怕。
解決方案
- 在每次調(diào)用查詢(xún)之后主動(dòng)關(guān)閉
neo4JGraph.close()
- aop 切片，在每次請(qǐng)求之后抒巢，執(zhí)行session 關(guān)閉
補(bǔ)充 Neo4JGraph 流程例如：
graphTraversalSource.V().has(LabelNameConstant.EQUIP, "name", equipTypeName).toList()
V() 和 has() 會(huì)記錄成 step
toList 之后才會(huì)真正觸發(fā)查詢(xún)
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, 此處第一次查詢(xún)時(shí)贫贝，會(huì)從數(shù)據(jù)庫(kù)進(jìn)行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 ,此處判斷是否是第一次查詢(xún)
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， 后面的數(shù)據(jù)全部從內(nèi)存中備份出來(lái)
return combine(transientVertices.stream().map(vertex -> (Vertex)vertex), vertices.values().stream().map(vertex -> (Vertex)vertex));
}
