Skip to content

04-커넥션과-리액티브-트랜잭션.md

Latest commit

 

History

History
139 lines (109 loc) · 10.6 KB

04-커넥션과-리액티브-트랜잭션.md

File metadata and controls

139 lines (109 loc) · 10.6 KB
Raw

04. 커넥션과 리액티브 트랜잭션

무엇을 / 왜

이 장이 모듈의 심장이다. 블로킹 JDBC에서는 "현재 스레드"에 트랜잭션 커넥션을 ThreadLocal로 묶어 같은 트랜잭션의 여러 쿼리가 한 커넥션을 공유한다. 리액티브에서는 한 흐름이 여러 스레드를 넘나들기 때문에 ThreadLocal이 무용지물이다.

spring-r2dbc의 해법은 Reactor의 구독 컨텍스트(subscriber Context)에 트랜잭션 자원을 저장하는 것이다. spring-tx의 리액티브 버전 TransactionSynchronizationManager가 그 컨텍스트를 다루고, 이 모듈의 ConnectionFactoryUtils(커넥션 획득/반납)와 R2dbcTransactionManager(트랜잭션 경계)가 그 위에서 협력한다.

핵심 타입

   R2dbcTransactionManager  (extends AbstractReactiveTransactionManager)
   ├ doBegin / doCommit / doRollback / doSuspend / doResume / doCleanup…
   └ ConnectionFactoryTransactionObject (내부) ── 보유 ─▶ ConnectionHolder
                                                            │
   ConnectionFactoryUtils (static helper)                  │ 트랜잭션에 묶인
   ├ getConnection(cf) / releaseConnection(con, cf)        │ Connection 보관
   ├ doGetConnection / doReleaseConnection  ← 컨텍스트 인식 │
   ├ convertR2dbcException(...)             ← 예외 변환     ▼
   └ ConnectionSynchronization (내부)              ConnectionHolder
        리액티브 트랜잭션 동기화 콜백             (extends ResourceHolderSupport)
              │ 등록/조회 대상                    ├ currentConnection
              ▼                                  ├ transactionActive
   TransactionSynchronizationManager            ├ savepointCounter (NESTED용)
   (spring-tx, 구독 컨텍스트 기반)               └ 참조 카운트/rollback-only (베이스)
   ├ getResource(cf) / bindResource(cf, holder)
   └ registerSynchronization(...)

자원 저장소의 키는 ConnectionFactory 객체다. 같은 팩토리로 만든 트랜잭션 매니저와 DatabaseClient가 같은 키로 컨텍스트를 조회하므로, 트랜잭션이 묶어 둔 Connection을 DatabaseClient가 그대로 집어 쓴다. 이것이 01장에서 "같은 ConnectionFactory를 공유해야 한다"고 한 이유다.

동작 흐름 1: 컨텍스트 인식 커넥션 획득

DatabaseClient가 쿼리를 실행할 때 ConnectionFactoryUtils.getConnection(cf)을 부른다. 이게 "트랜잭션 안이면 트랜잭션 커넥션을, 밖이면 새 커넥션을"이라는 판단을 한다.

 getConnection(cf)  →  doGetConnection(cf)
   │
   ▼
 TransactionSynchronizationManager.forCurrentTransaction()  ← 구독 컨텍스트 조회
   │
   ├─ NoTransactionException (컨텍스트에 트랜잭션 없음)
   │     └▶ onErrorResume → cf.create()   "그냥 새 커넥션"
   │
   └─ 트랜잭션 컨텍스트 있음
        │
        ▼
      conHolder = mgr.getResource(cf)
        │
        ├─ holder 있고 커넥션/동기화 보유  ──▶ holder.requested(); 보유 Connection 재사용
        │
        └─ holder 없음/비어 있음
             │
             ▼
           con = cf.create()
             │
             ├─ 동기화 활성?  ── 예 ─▶ ConnectionHolder에 담고
             │                         registerSynchronization(ConnectionSynchronization)
             │                         bindResource(cf, holder)  ← 컨텍스트에 등록
             │                         이후 같은 트랜잭션 쿼리는 이 커넥션 재사용
             └─ 아니오 ─▶ con 그대로 반환

핵심은 holder.requested()/released()로 참조 카운트를 관리한다는 점(ResourceHolderSupport 상속). 같은 트랜잭션 안에서 커넥션을 여러 번 빌려도 실제 물리 커넥션은 하나이며, 모든 사용자가 반납해야 실제로 닫힌다. 반납은 doReleaseConnection이 처리하는데, "들고 있는 holder의 커넥션과 같으면 닫지 않고 released()만" 한다 — 트랜잭션 커넥션을 쿼리 하나 끝났다고 닫아 버리면 안 되기 때문이다.

ConnectionSynchronization — 트랜잭션 종료 시 정리

getConnection이 등록하는 ConnectionSynchronization은 트랜잭션 생명주기 콜백이다. suspend/resume에서 자원을 풀고 다시 묶으며, beforeCompletion/afterCompletion에서 커넥션을 실제로 닫는다. holder가 더는 열려 있지 않으면(isOpen()==false) beforeCompletion에서 일찍 닫아, 엄격한 트랜잭션 구현이 "커밋 전 close"를 기대하는 경우에 대응한다. getOrder()는 DelegatingConnectionFactory 중첩 깊이만큼 감소시켜(CONNECTION_SYNCHRONIZATION_ORDER=1000에서 한 겹마다 -1) 프록시 체인의 정리 순서를 맞춘다.

동작 흐름 2: 트랜잭션 begin → commit/rollback

@Transactional(리액티브) 또는 TransactionalOperator가 R2dbcTransactionManager를 구동한다. 이 매니저는 spring-tx의 AbstractReactiveTransactionManager 템플릿 메서드를 채운다.

 doBegin(mgr, txObject, definition)
   │
   ├─ PROPAGATION_NESTED && 이미 활성 트랜잭션?
   │     └▶ txObject.createSavepoint()   ← R2DBC savepoint로 중첩 (6.0.10+)
   │
   └─ 일반 begin (Mono.defer로 지연)
        │
        ▼
      holder 없음/동기화됨?  ── 예 ─▶ cf.create()로 새 Connection
        │                            ConnectionHolder 생성 (new=true)
        │ 아니오
        ▼ (기존 holder 재사용)
      con.beginTransaction(createTransactionDefinition(def))   ← 격리수준/readOnly/timeout 전달
        │
        ▼
      prepareTransactionalConnection(con, def)   ← enforceReadOnly면 "SET TRANSACTION READ ONLY"
        │ 성공
        ▼
      holder.setTransactionActive(true); timeout 설정
      새 holder면 mgr.bindResource(cf, holder)   ← 컨텍스트에 등록
        │ 실패
        ▼
      새 holder면 releaseConnection 후 에러 전파
      → 전체를 CannotCreateTransactionException으로 onErrorMap

createTransactionDefinition이 Spring의 TransactionDefinition(격리수준 상수, readOnly, timeout)을 R2DBC SPI의 확장 가능한 io.r2dbc.spi.TransactionDefinition(내부 ExtendedTransactionDefinition 레코드)으로 변환해 드라이버에 그대로 넘긴다. 격리수준은 resolveIsolationLevel이 Spring 상수 → R2DBC IsolationLevel로 매핑한다(READ_COMMITTED 등). 이렇게 트랜잭션 속성이 드라이버까지 전달되는 게 5.3 초기보다 발전한 부분이다.

커밋/롤백은 txObject에 위임한다.

// doCommit
return txObject.commit()
    .onErrorMap(R2dbcException.class, ex -> translateException("R2DBC commit", ex));

// ConnectionFactoryTransactionObject.commit()
return (hasSavepoint() ? Mono.empty()
        : Mono.from(getConnectionHolder().getConnection().commitTransaction()));

savepoint를 가진 중첩 트랜잭션이면 커밋은 no-op이고(바깥 트랜잭션이 진짜 커밋), 롤백은 rollbackTransactionToSavepoint로 부분 롤백한다. 커밋/롤백 중 터진 R2dbcException은 translateException → convertR2dbcException으로 Spring 예외가 된다.

doCleanupAfterCompletion — 자원 원복

트랜잭션이 끝나면 doCleanupAfterCompletion이 호출된다. savepoint만 있으면 savepoint를 풀고 커넥션은 유지한다. 새로 만든 holder면 컨텍스트에서 풀고(unbindResource), autoCommit을 원복해야 하면(mustRestoreAutoCommit) setAutoCommit(true)을, 그리고 releaseConnection으로 커넥션을 닫는다. autoCommit 원복과 close 중 에러는 디버그 로그만 남기고 삼킨다(onErrorComplete) — 정리 단계 실패가 본 로직을 깨지 않게.

ConnectionHolder의 역할

ConnectionHolder는 트랜잭션에 묶인 Connection의 보관자다. ResourceHolderSupport를 상속해 참조 카운트, rollback-only 플래그, timeout을 물려받고, 여기에 transactionActive와 savepoint 카운터(SAVEPOINT_ 접두사로 이름 생성)를 더한다. SPI 클래스로 애플리케이션이 직접 쓸 일은 없고, ConnectionFactoryUtils와 R2dbcTransactionManager가 컨텍스트에 넣고 빼는 값이다.

예외 변환 (재방문)

01장에서 본 convertR2dbcException은 이 패키지에 산다. 트랜잭션 commit/rollback(doCommit/doRollback)과 DatabaseClient의 쿼리 실행(02장 inConnection/inConnectionMany의 onErrorMap) 양쪽에서 호출되어, 드라이버가 던진 R2dbcException을 스트림 위에서 org.springframework.dao 계층으로 바꾼다. 동기 try/catch가 아니라 onErrorMap/onErrorResume을 쓰므로 리액티브 체인 어디서 터지든 동일하게 잡힌다.

설계 포인트 / 확장점

  • 컨텍스트 = ThreadLocal의 리액티브 대체: 모든 자원 공유가 TransactionSynchronizationManager.forCurrentTransaction()(구독 컨텍스트)를 통한다. NoTransactionException을 "트랜잭션 없음 → 새 커넥션"으로 우아하게 처리하는 onErrorResume이 포인트.
  • 참조 카운트 기반 수명: holder의 requested/released로 같은 커넥션을 안전하게 공유·반납. 트랜잭션 커넥션은 쿼리 종료가 아니라 트랜잭션 종료에 닫힌다.
  • 확장 메서드: createTransactionDefinition(벤더 속성 추가), prepareTransactionalConnection(readOnly 외 커스텀 SQL), resolveIsolationLevel(벤더별 격리수준), translateException을 오버라이드해 동작을 조정할 수 있다.
  • NESTED 트랜잭션: 6.0.10부터 R2DBC savepoint로 지원. createSavepoint/releaseSavepoint/rollbackTransactionToSavepoint로 구현.
  • enforceReadOnly: true면 readOnly 트랜잭션에 "SET TRANSACTION READ ONLY"를 실행(Oracle/MySQL/Postgres). 동작을 바꾸려면 prepareTransactionalConnection 오버라이드.

정리

리액티브 트랜잭션의 본질은 "자원 저장소를 스레드에서 구독 컨텍스트로 옮긴 것"이다. ConnectionFactoryUtils는 그 컨텍스트를 들여다보며 트랜잭션 안이면 묶인 커넥션을, 밖이면 새 커넥션을 주고, 참조 카운트로 수명을 관리한다. R2dbcTransactionManager는 AbstractReactiveTransactionManager 템플릿을 채워 begin/commit/rollback을 Mono 체인으로 구현하고, Spring 트랜잭션 속성을 R2DBC SPI 정의로 변환해 드라이버에 전달하며, savepoint로 중첩 트랜잭션까지 지원한다. 모든 드라이버 오류는 convertR2dbcException으로 공통 예외가 된다.