베타 출시: Neo4j용 비동기 API를 갖춘 Java Driver

• News

소개

이번 글에서는 Bolt Java 드라이버의 새로운 1.5.0-beta03 베타 버전을 소개하려고 해요. 이 버전은 용으로, 이제 비동기식 Netty 기반 인프라를 기반으로 구축되었답니다.

이전 버전의 드라이버는 차단 I/O를 사용했는데, 이는 *N*개의 동시 연결을 처리하려면 *N*개의 스레드가 필요했다는 의미였어요. 하지만 새로운 비차단 I/O를 사용하면 하나의 스레드가 여러 네트워크 연결을 비동기 방식으로 처리할 수 있어서 스레드 수를 훨씬 줄일 수 있죠. 이 기능은 쿼리와 트랜잭션을 차단 없이 실행할 수 있는 새로운 비동기식 메서드 세트를 통해 제공된답니다.

결과의 비동기 처리는 Akka 액터나 Spring Data 반응형 저장소처럼 코드가 최대한 차단되지 않아야 하는 환경에서 특히 유용할 거예요.

한 가지 중요한 점은 1.5부터 Neo4j Java 드라이버는 Java 8이 필요하다는 거예요. 기존 비동기 프로그래밍 API 및 인터페이스인 CompletionStage 와 CompletableFuture를 사용하기 위해 필요한 Java 버전을 올리기로 결정했답니다. Java 8부터만 제공되거든요. 이제 Session#runAsync(), Transaction#runAsync(), Session#readTransactionAsync() 같은 비동기 API 호출에 사용될 거예요. 이전 드라이버 버전 1.4는 여전히 Java 7이 필요하며 유지 관리되고 있어요.

비동기 API

이번 섹션에서는 1.5.0-beta03 Java 드라이버 버전에 있는 새로운 비동기 API에 대해 알아볼게요. API 대응 차단에 대해서는 다루지 않을 거예요. 자세한 내용은 Neo4j 개발자 매뉴얼을 참고해주세요. 차단 API는 비동기 API 위에 다시 구현되었기 때문에 기본 인프라를 공유한답니다.

드라이버 초기화

드라이버 API의 주요 진입점은 변경 없이 그대로 유지돼요. GraphDatabase 클래스를 사용해서 드라이버를 생성할 수 있죠.

import org.neo4j.driver.v1.AuthTokens;
import org.neo4j.driver.v1.Driver;
import org.neo4j.driver.v1.GraphDatabase;

Driver driver = GraphDatabase.driver("bolt://localhost", 
                                     AuthTokens.basic("neo4j", "test"));

Driver는 모든 Neo4j 상호 작용이 이루어지는 스레드로부터 안전한 애플리케이션 전체 객체에요.

세션

Driver 인스턴스를 사용해서 Session 인스턴스를 획득해야 쿼리를 실행하고 트랜잭션을 생성할 수 있어요.

Session은 하나 이상의 작업 단위를 논리적으로 그룹화하기 위한 클라이언트 측 추상화예요. 단일 스레드 사용을 위해 설계되었고, 요청된 작업을 실행할 때 TCP 연결을 통해 지원될 수 있어요. 라우팅 드라이버에서 bolt+routing URI를 사용하면, 단일 세션 내의 모든 트랜잭션은 책갈피를 통해 암시적으로 연결되죠.

더 자세한 내용은 Neo4j 개발자 매뉴얼의 인과 연결 섹션을 참고해주세요.

새로운 세션은 다음과 같이 생성할 수 있어요:

import org.neo4j.driver.v1.Session;

Session session = driver.session();

세션을 사용해서 첫 번째 비동기 쿼리를 실행해볼까요?

import org.neo4j.driver.v1.StatementResultCursor;

CompletionStage<StatementResultCursor> cursorStage = 
      session.runAsync("UNWIND range(1, 10) AS x RETURN x");

cursorStage.thenCompose(StatementResultCursor::listAsync)
                    .whenComplete((records, error) -> {
                           if (records != null) System.out.println( records );
                           else error.printStackTrace();
                           session.closeAsync();
                    });

Session#runAsync()를 호출하면 StatementResultCursor의 CompletionStage가 반환되는데, 쿼리 결과를 사용하기 위한 주요 추상화이며 데이터베이스에서 비동기적으로 수신돼요. 이 예제에서는 모든 결과를 나중에 출력할 목록으로 가져오고 있어요.

결과 크기에 따라 힙 메모리가 많이 필요할 수도 있어요. 규모가 큰 경우에는 #forEachAsync() 와 #nextAsync()를 사용해서 점진적으로 소비하면 좋겠죠. 생성된 세션 객체는 체인 끝에서 명시적으로 닫아줘야 해요.

Session#runAsync() 메서드는 쿼리 매개변수, 레코드 및 편의를 위한 org.neo4j.driver.v1.Statement 객체를 받아요.

단일 레코드만 반환된다고 가정하고 커서에서 단일 레코드를 안전하게 검색하는 것도 가능해요.

import org.neo4j.driver.v1.StatementResultCursor;

CompletionStage<StatementResultCursor> cursorStage = session.runAsync("MATCH (n) RETURN n LIMIT 1");

cursorStage.thenCompose(StatementResultCursor::singleAsync)
                   .thenApply(record -> record.get( 0 ).asNode())
                   .thenApply(Node::labels)
                   .exceptionally(error -> {
                       error.printStackTrace();
                       return emptyList();
                   })
                   .thenAccept(labels -> System.out.println(labels))
                   .thenCompose(ignore -> session.closeAsync());

이 코드는 가져온 Node의 모든 Label을 출력해요. 에러(데이터베이스 사용 불가, 네트워크 에러, 가져온 Node 없음 등)가 발생하면 스택 추적을 출력하는 대신 빈 Node 목록을 반환해서 명시적으로 처리하죠.

결과 레코드를 하나씩 또는 스트림으로 소비해야 할 때도 있을 텐데요. StatementResultCursor는 다음 두 가지 방법을 제공해서 이걸 가능하게 해줘요.

• CompletionStage<Record> nextAsync() – 결과 스트림의 다음 레코드로 완료된 단계를 반환해요. 스트림의 끝에 도달했을 때는 null을 반환하죠. 쿼리가 실패하면 예외적으로 스테이지가 완료될 수도 있어요.
• CompletionStage<ResultSummary> forEachAsync(Consumer<Record> action) – 요약과 함께 완료된 단계를 반환하고, 제공된 작업을 결과 스트림의 모든 레코드에 적용해요.

#forEachAsync() 메서드는 StatementResultCursor를 RxJava 1.x 라이브러리의 rx.Observable로 변환하는 데 사용할 수 있어요. rx.subject.PublishSubject를 사용한 간단한 예시는 다음과 같아요:

import rx.Observable;
import rx.subjects.PublishSubject;

Observable<Record> fetchRecords(Session session, String query) {
    PublishSubject<Record> subject = PublishSubject.create();
    session.runAsync(query)
                .thenCompose(cursor -> cursor.forEachAsync(subject::onNext))
                .whenComplete((summary, error) -> {
                    if (error != null) {
                        subject.onError( error );
                    } else {
                        System.out.println( summary );
                        subject.onCompleted();
                    }
                });
    return subject;
}

Observable<Record> recordsObservable = fetchRecords(session, "MATCH (n:Person) RETURN n");
recordsObservable.subscribe(
    record -> System.out.println(record),
    error -> error.printStackTrace(),
    () -> System.out.println("Query completed")
);

수신된 모든 레코드는 #forEachAsync()와 PublishSubject를 사용해서 소비되고, 구독자는 모든 기록에 액세스할 수 있게 되죠.

트랜잭션

Session을 사용하면 독립 실행형 쿼리를 실행할 수 있을 뿐만 아니라, 명시적인 트랜잭션 내에서 쿼리를 실행할 수도 있어요. 호출자는 트랜잭션 시작, Cypher 쿼리 실행, 커밋 또는 롤백을 제어할 수 있죠.

다음과 같이 사용하는 걸 추천해요. Neo4j 개발자 매뉴얼에 자세히 설명된 트랜잭션 기능 API를 명시적 트랜잭션을 통해 사용하는 것이죠. 이는 차단 API와 비동기 API 모두에 해당돼요.

이번 섹션에서는 비동기 트랜잭션 기능을 살펴볼게요.

비동기 트랜잭션 기능의 두 가지 주요 진입점은 다음과 같아요.

• Session#readTransactionAsync(TransactionWork<CompletionStage<T>>)
• Session#writeTransactionAsync(TransactionWork<CompletionStage<T>>)

이를 통해 주어진 읽기/쓰기 트랜잭션을 실행할 수 있는 TransactionWork 객체를 비동기 방식으로 생성할 수 있어요.

Node를 생성하는 간단한 쓰기 트랜잭션은 다음과 같아요.

session.writeTransactionAsync(tx ->
  tx.runAsync("CREATE (n:Person) RETURN n")
     .thenCompose(StatementResultCursor::singleAsync)
).whenComplete((record, error) -> {
    if (error != null) error.printStackTrace();
    else System.out.println(record);
    session.closeAsync();
});

Person Node를 생성하는 쓰기 트랜잭션을 실행하고 결과 레코드를 출력해요. 트랜잭션을 사용하면 다양한 오버로드를 통해 비동기 방식으로 쿼리를 실행할 수 있는 Transaction#runAsync()는 Session#runAsync()와 똑같이 StatementResultCursor를 반환하죠. 주어진 TransactionWork가 성공하면 트랜잭션이 자동으로 커밋되고, 실패하면 롤백돼요.

단일 문으로 구성된 읽기 트랜잭션은 다음과 같아요.

session.readTransactionAsync(tx ->
    tx.runAsync("MATCH (n:Person) RETURN n")
      .thenCompose(cursor -> cursor.forEachAsync(System.out::println))
).whenComplete((ignore, error) -> {
    if ( error != null ) error.printStackTrace();
    session.closeAsync();
});

위의 예시에서 모든 기록은 트랜잭션 본문 내에서 소비되고 인쇄돼요. 나중에 자동으로 커밋되거나 롤백될 거예요.

드라이버 종료

사용한 Driver 인스턴스는 설정된 모든 네트워크 연결을 유지하는데, 애플리케이션이 종료되거나 Neo4j와의 상호 작용이 완료되면 명시적으로 닫아줘야 해요. 그렇지 않으면 파일 디스크립터 누출이 발생하거나 애플리케이션이 종료되지 않을 수 있어요. 드라이버는 다음과 같이 닫을 수 있어요:

driver.close();

close 작업은 모든 네트워크 연결과 I/O 스레드를 종료해요. 이는 블로킹 작업이며 모든 리소스가 종료되면 반환될 거예요.

Maven과 함께 사용

새로운 Java 드라이버 릴리스를 이제 Maven Central Repository에서 사용할 수 있어요. 다음 종속성 정의를 사용해서 Maven 프로젝트에 포함할 수 있죠.

<dependency>
    <groupId>org.neo4j.driver</groupId>
    <artifactId>neo4j-java-driver</artifactId>
    <version>1.5.0-beta03</version>
</dependency>

드라이버에는 Netty에 대한 컴파일 시간 종속성이 있지만 최종 드라이버 아티팩트에 섀도 처리되므로 버전 종속성 충돌은 없을 거예요.

기타 주목할 만한 변경 사항

이번 Java 드라이버 릴리스에는 비동기 API 외에도 몇 가지 새로운 기능이 추가되었어요. 가장 눈에 띄는 것은 다음과 같아요:

• 인과 클러스터링을 위한 새로운 로드 밸런싱 전략: 라운드 로빈 대신 최소 연결 전략을 사용해요. 이는 일부 클러스터 멤버의 성능이 네트워크 또는 기타 유사한 문제로 인해 저하될 때 성능을 향상시키고 저하를 줄일 수 있어요.
• 향상된 연결 풀링: 이제 Java 드라이버를 사용하면 Config.build().withMaxConnectionPoolSize(25)를 통해 서버 주소당 풀의 연결 양에 대한 제한을 설정할 수 있고, Config.build().withConnectionAcquisitionTimeout(10, TimeUnit.SECONDS)를 통해 연결 획득 시간 초과를 설정할 수 있어요.
• 최대 연결 수명: Java 드라이버는 Config.build().withMaxConnectionLifetime(1, TimeUnit.HOURS)를 사용해서 구성할 수 있는 연결 수명을 제한할 수 있어요.

결론

Bolt Java 드라이버 1.5는 새로운 기능이 많이 포함된 대규모 릴리스예요. 새로운 비동기 API는 이번 작업과 가장 관련이 깊고, 드라이버 사용자가 다른 방식으로 상호 작용할 수 있도록 해주죠. 또한 다음과 같은 새롭고 풍부한 Java 8 API에 대한 액세스를 제공해요: CompletionStage.

이 시점에서 새로운 비동기 API에 대한 커뮤니티의 의견은 정말 큰 도움이 될 거예요. API 디자인을 미세 조정하고 비동기 코드 기반에 최대한 많은 가치를 제공할 수 있을 테니까요.

여기에 설명된 드라이버는 베타 버전이므로 프로덕션 환경에서 사용하면 안 돼요. 가장 안정적인 Neo4j Java 드라이버 버전을 보려면 여기를 클릭하세요.

• bolt
• java
• 자바 8
• Maven
• Netty
• Spring Data

---

에이치시스템즈의 LogTree는 Neo4j 기반 GraphRAG 플랫폼으로, 데이터를 자동으로 지식그래프화하고 자연어 질의로 즉시 답을 제공합니다.

👉 에이치시스템즈 홈페이지