현재 우리의 서비스의 TPS가 500이라고 해도 타 서비스의 TPS가 200이라면 결국 우리의 TPS는 200이 된다.

이렇게 서비스 간에 TPS의 차이가 발생할 때 어떻게 대처해야 할까?

해결

상대 서비스의 TPS에 맞춰서 우리의 TPS를 조절해서 타 서비스에 요청을 보내야 한다. 즉, 상대 측으로 가는 요청을 제한해서 보내야 한다.

TPS를 조절하지 않는다면 타 서비스에 부하가 발생해 응답 시간은 계속 지연되면서 장애가 발생할 것이고, 해당 서비스를 사용하는 우리 서비스에도 장애가 전파되기 때문이다.

이러한 현상을 방지하기 위한 대표적인 처리율 제한 방식으로 Bulk Head 방식과 Rate Limiter 방식이 있다.

Bulk Head

• 리소스에 동시에 접근할 수 있는 요청 수를 제한하는 것이 주 목표이다.
• 스레드 풀 또는 세마포어를 통해 접근할 수 있는 요청들을 제한한다.
• 제한한 요청 값 이내의 요청이 들어오면 병렬로 요청들을 처리한다.
• 설정한 요청 수를 초과하는 경우, 일정 시간 동안 대기했다가 fallback함수로 처리한다.
• ex) DB 커넥션 풀 보호 : 동시에 최대 10개의 연결만 허용

Rate Limiter

• 일정 시간 동안 리소스에 접근할 수 있는 요청 수를 제한하는 것이 주 목표이다.
• 처리율 제한 알고리즘의 대표적인 방식으로 token bucket, sliding window counter 등이 있다.
• 설정한 요청 수를 초과하는 경우, 일정 시간 동안 대기했다가 fallback함수로 처리한다.
• ex) 하루에 최대 10,000번 요청까지 무료 , 1분에 최대 1,000번 요청 가능

정리

그렇다면 각 서비스 간의 TPS간 차이가 발생한다면 어떤 방식을 사용해야할까? 우선 TPS란 초당 처리할 수 있는 트랜잭션 수이기 때문에 이것은 일정 기간에 해당한다. 따라서 Rate Limiter를 통해서 처리율을 제한하자.

그렇다면 1초당 200개의 요청을 처리하도록 설정하는 것도 좋지만, 아래 사항들을 추가로 고려하면 더 좋다.

1. 해당 서비스를 사용하는 서비스들은 얼마나 있을까?

만약 또 다른 서비스에서도 동시에 200개를 제한한다면 해당 서비스는 1초에 400개의 요청을 처리해야 하기 때문에 본인의 TPS를 초과한다.

그렇기 때문에 해당 서비스를 사용하는 또 다른 서비스들과 TPS 비율을 조율해서 200개 보다 적은 수를 설정하는 것이 안전하다.

2. 동시에 처리할 수 있는 처리량은 어떻게 될까?

1초당 200개를 처리할 수 있다고 해서 동시에 200개를 한번에 요청하면 위험할 수 있다.

따라서 Bulk Head도 추가로 적용해서 동시에 요청할 수 있는 수를 제한하는 것이 안전하다.

마이크로서비스는 각 서비스마다 데이터베이스를 할당하고 이는 다음과 같은 이점을 준다.

1. 각 서비스는 자신의 서비스에 적합한 RDBMS를 사용할 수 있다.
2. 한 서비스의 장애가 다른 서비스로 전파되지 않는다.
3. 독립적인 서비스로 인해 배포와 확장에 유연하다.

하지만 데이터베이스가 독립적이기 때문에 트랜잭션을 보장하는 것이 어렵다는 단점이 있다.

이러한 여러 개의 독립적인 데이터베이스 위에서 하나의 트랜잭션을 수행하기 위해 2PC, SAGA 패턴 등과 같은 분산 트랜잭션 방식이 있다.

2PC

모든 서비스가 트랜잭션을 수행할 준비가 됐을 때만 최종 커밋을 진행하는 방식으로 강한 일관성을 제공한다.

동작 과정

준비 단계, 커밋 단계로 구성되어 있고, 중앙 제어자인 코디네이터가 이를 제어한다.

코디네이터는 여러 서비스들에게 읽기 및 쓰기 작업을 진행하고, 이 과정에서 리소스를 점유한다.

트랜잭션을 수행한 후 커밋을 하려 할 때 아래와 같이 진행한다.

1. 준비 단계

• 코디네이터는 모든 서비스들에게 트랜잭션 준비 요청을 브로드캐스팅한다.
• 각 서비스들은 준비 완료 또는 준비 실패 응답을 코디네이터한테 응답한다.

2. 커밋 단계

• 코디네이터는 모든 서비스의 응답을 받을 때까지 대기한다.
• 모든 서비스가 준비 완료 상태라면 커밋을, 하나라도 실패하면 롤백을 결정한다.
• 결정된 작업을 모든 서비스들에게 브로드캐스팅한다.
• 서비스들은 코디네이터의 요청을 보고 커밋 또는 롤백을 수행한다.

장단점

장점

• 모든 서비스가 성공할 때만 커밋하기 때문에 강한 일관성을 보장한다.
• 상대적으로 이해하기 쉽고, 구현이 단순하며, 관리 비용이 적다.

단점

• 데이터베이스가 분산 트랜잭션을 지원해야 한다.

• 모든 서비스의 응답을 기다려야 하기 때문에 리소스 점유 시간이 길다.

• 코디네이터의 장애가 단일 지점 장애가 된다.

  모든 서비스는 코디네이터의 결정을 기다릴 때까지 블로킹 상태가 된다.

  만일 2PC 진행 중 코디네이터에 장애가 발생한다면, 모든 서비스는 코디네이터가 복구될 때까지 블로킹 상태가 된다.

발생할 수 있는 장애 상황 및 해결 방법

데이터를 모두 기록한 뒤 코디네이터에 장애가 발생한 경우

• 각 참여자들은 아직 준비 요청을 받지 않은 상태이기 때문에 타임아웃이 발생할 때까지 준비 요청을 기다린 후 트랜잭션을 자동으로 중단한다.

참여자들에게 준비 요청을 보낸 뒤 코디네이터에 장애가 발생한 경우

• 각 참여자들은 커밋이나 롤백 요청을 기다리는 블로킹 상태가 되며, 코디네이터가 복구될 때까지 대기 하게 된다.
• 복구된 코디네이터는 트랜잭션 로그를 확인해서 각 참여자들에게 결정을 다시 요청한다.

모든 참여자에게 커밋 요청을 보낸 뒤, 혹은 보내는 중에 코디네이터에서 장애가 발생한 경우

• 각 참여자들은 이미 커밋 요청을 받았기 때문에 데이터의 변경 사항을 반영한다.
• 코디네이터가 복구된다면, 멱등키와 함께 모든 참여자들에게 커밋 요청을 다시 전송한다. 이를 통해 데이터 불일치를 해결한다.

참여자가 커밋하는 도중에 장애가 발생한 경우

• 참여자는 장애를 복구한 뒤에 트랜잭션 로그를 확인해서 커밋을 마저 진행한다.

참여자가 커밋 요청을 받기 전에 장애가 발생한 경우

• 참여자는 장애를 복구한 뒤에 트랜잭션 로그를 확인해서 준비 상태인 작업에 관해서 코디네이터에게 해당 작업의 결과를 응답 받은 후 해당 응답에 맞게 마저 진행한다.

SAGA 패턴

각 서비스는 독립적인 로컬 트랜잭션을 순차적으로 수행하고, 만일 중간에 실패한다면 그동안 성공했던 트랜잭션을 취소하는 보상 트랜잭션을 역순으로 실행하는 방식이다.

구현 방식

코레오그래피 방식

중앙 제어자 없이 메시지 브로커를 이용해 이벤트 기반으로 다른 로컬 트랜잭션을 트리거하는 방식이다.

장점

• 이벤트의 책임은 각 서비스에 있기 때문에 단일 지점 장애가 없다.
• 트랜잭션 조정을 위한 별도의 서비스가 필요없다.

단점

• 트랜잭션 추적이 어렵다.

오케스트레이션 방식

별도의 중앙 제어자가 모든 트랜잭션을 처리하고 이벤트에 따라서 각 서비스들에게 어떤 작업을 수행할지 알려준다.

장점

• 중앙 제어자로 인해 트랜잭션 추적이 쉽다.
• 중앙 제어자가 트랜잭션 및 보상 트랜잭션을 트리거하기 때문에 각 서비스의 로직은 단순해진다.

단점

• 중앙 제어자가 단일 지점 장애가 된다.

장단점

장점

• 각 서비스는 자신의 트랜잭션을 처리할 때만 리소스를 점유하므로 2PC 방식보다 락 점유 시간이 짧다.
• 트랜잭션이 작은 단위로 분리되면서 서비스의 가용성이 향상된다.
• 서비스의 독립성이 향상된다.

단점

• 각 트랜잭션에 대한 보상 로직을 설계하고 관리해야 한다.
• 트랜잭션이 진행되는 동안 일시적으로 데이터 불일치가 발생할 수 있다.
• 결과적 일관성을 보장하기 위해 트랜잭션이 발생했음을 다른 서비스들에게 적어도 한번은 알려줘야 하고, 이로 인해 구현의 난이도가 매우 높다.

What is Two Phase Commit in Distributed Transaction?

가상 면접 사례로 배우는 대규모 시스템 설계 기초2

Saga distributed transactions pattern

JVM에서 제공하는 스레드를 플랫폼 스레드라고 말하며, JNI를 통해 시스템 콜을 호출하고 커널 스레드를 생성함으로써 플랫폼 스레드는 커널 스레드와 1:1로 매핑된다.

컨텍스트 스위칭을 OS 수준에 위임하기 때문에 구현이 간단하고, 멀티 코어를 활용할 수 있다는 장점이 있지만, 다음과 같은 문제점이 있다.

1. 메인 메모리의 크기는 한정적이기 때문에 생성할 수 있는 커널 스레드의 개수는 한정적이다. → 최대 처리할 수 있는 요청량은 커널 스레드의 최대 생성 개수가 된다.
2. 블로킹 I/O 발생 시, 컨텍스트 스위칭이 커널 수준에서 발생해 비교적 오버헤드가 크다.
3. 논블로킹 I/O 방식인 리액티브 스트림즈를 통해 해당 문제를 해결할 수 있지만, 이는 러닝 커브가 높고, 이벤트 루프 기반으로 동작하기 때문에 블로킹 I/O를 잘못 사용했을 때 성능이 크게 저하된다.

이러한 문제를 해결하기 위해서 가상 스레드가 생겼다.

가상 스레드

가상 스레드는 JDK 21에서 도입된 경량 스레드로, 기존 플랫폼 스레드와 달리 커널 스레드에 종속되지 않는다.

동작 방식

가상 스레드는 커널 스레드와 1:1로 매핑한 캐리어스레드 위에서 동작하고, 가상 스레드와 캐리어 스레드는 M:N으로 매핑된다.

가상 스레드에서 블로킹 I/O가 발생한다면 해당 가상 스레드는 일시 중단 상태가 되고, 캐리어 스레드는 다른 가상 스레드를 실행한다.

이렇게 캐리어 스레드가 어떤 가상 스레드를 사용할지는 JVM 내부의 스케줄러에서 하며, 스케줄러는 ForkJoinPool을 사용한다.

이러한 동작 방식은 다음과 같은 장점이 있다.

• 가상 스레드 간의 전환은 사용자 수준에서 발생하기 때문에 컨텍스트 스위칭 비용이 낮다.
• 가상 스레드의 생성은 커널 스레드를 생성하지 않기 때문에 비교적 빠르고, 저장 공간 또한 가볍다. → 많은 수의 가상 스레드를 생성할 수 있다.

이러한 장점으로 인해 기존 방식을 그대로 유지하면서 높은 처리량을 제공할 수 있다.

💡 ForkJoinPool이란?

특정 작업을 작은 단위로 여러 개 나눠서 각 스레드에서 작업한 뒤, 결과물을 합치는 방식으로 동작하는 스레드 풀이다.

작업 훔치기 기능이 있기 때문에 먼저 작업을 다 끝낸 스레드는 아직 작업이 남은 스레드의 작업 큐의 꼬리에서부터 작업을 가져와서 수행한다.

이 기능을 통해 놀고 있는 스레드 없이 효율적으로 작업을 진행할 수 있다.

가상 스레드의 컨텍스트 스위칭

자바에서 작업을 진행할 때, 실행 할 메서드를 스택 영역에 쌓아두고 순차적으로 실행한다. 가상 스레드의 컨텍스트 스위칭은 이러한 특성을 이용한다.

작업을 진행하다가 블로킹 I/O를 만나서 중지된 시점부터 앞으로 실행될 메서드들을 잘라내서 힙 영역에 보관한다.

이후 I/O 작업이 완료되면서 작업을 재개할 때, 힙에 저장된 메서드들을 스택 영역에 복구한 뒤 중지된 시점부터 다시 순차적으로 메서드를 실행한다.

즉,"스택 프레임을 자르고 붙이는" 과정을 통해서 가상 스레드 간의 전환이 이루어진다.

이러한 과정은 park, unpark 메서드를 통해 진행된다.

park()

가상 스레드에서 블로킹 I/O가 발생해 작업을 중지할 때 실행되는 메서드로 다음과 같이 진행된다.

1. 가상 스레드의 상태를 PARKING 상태로 전환한다.

2. unmount() 함수를 실행해 현재 캐리어 스레드와의 연결을 해제한다.

3. yield 함수를 실행해 현재 캐리어 스레드를 다른 가상 스레드에게 양보한다.

   3.1. 현재 실행 중인 스택 프레임을 힙 영역에 저장한다.

   3.2. 다른 가상 스레드에게 양보한다.

4. 가상 스레드의 상태를 PARKED 상태로 전환한다.

unpark()

블로킹 I/O 작업이 완료된 후 중지된 작업을 다시 실행되는 메서드로 다음과 같이 진행한다.

1. 스케줄러에게 runContinuation 메서드 실행을 요청한다.

2. Continuation.run으로 진입해서 enterSpecial 네이티브 메서드를 실행한다.

   enterSpecial 메서드 : 힙에 저장된 작업을 스택에 복원한다.

3. 중지된 시점부터 작업을 재개한다. (mount() 실행)

가상 스레드 권장 사항

ThreadLocal에 캐싱하는 것을 지양하자.

캐싱은 생성 비용이 부담될 때 사용한다. 스레드 로컬에 캐싱하게 된다면 모든 캐리어 스레드의 스레드 로컬에 캐싱이 필요하고, 이로 인해 캐리어 스레드의 크기가 커지게 된다.

스레드 로컬을 사용하기 보단 final static과 같은 불변 변수를 공유하는 것을 권장한다.

스레드 풀을 만들어서 동시성을 제어하는 것을 지양하자.

풀을 만들어서 미리 생성해두는 것은 생성 비용에 부담이 있어서 미리 만들어두고, 이를 공유하기 위함이다. 하지만 가상 스레드는 생성 비용에 부담이 있지 않고 가볍기 때문에 스레드 풀을 만들 필요가 없다.

물론, 스레드 풀을 만들어 스레드 개수를 제한함으로써 동시성을 제어할 수 있다. (ex. 스레드를 10개로 만들어서 특정 서비스의 접근을 동시에 10개만 할 수 있도록 만들 수 있다.)

하지만 위 동작은 스레드 풀의 부수 효과이며, 본질적인 목표가 아니다.

따라서 동시성을 제어하고 싶을 땐 세마포어를 사용하는 것을 권장한다.

CPU Bound 작업을 지양하자.

가상 스레드에서 CPU Bound 작업을 진행하면 캐리어 스레드를 오랫동안 점유하기 때문에 가상 스레드의 처리량이 제한된다.

이는 처리량을 높이는 목적으로 설계된 가상 스레드의 취지에 부합하지 않기 때문에 CPU Bound 작업을 진행할 땐, 별도의 스레드 풀을 사용해 플랫폼 스레드에서 작업하는 것을 권장한다.

길고 빈번한 PINNING을 지양하자.

가상 스레드에서 synchronized를 사용하면 synchronized 내부의 모니터가 캐리어 스레드를 점유하기 때문에 가상 스레드가 고정된다. (Pinned 현상)

물론 synchronized의 작업이 짧다면 크게 상관없지만, 긴 작업을 수행하면 캐리어 스레드가 처리할 수 있는 처리량이 줄어들기 때문에 지양해야 한다.

synchronized 대신 모니터를 사용하지 않는 ReentrantLock을 사용할 것을 권장한다.

Pinned 현상이 무엇이고, 왜 발생할까

가상 스레드가 블로킹 I/O를 만나 캐리어 스레드의 연결을 해제하는 과정이 필요한데, 이 때 연결이 해제가 되지 않고 캐리어 스레드와 계속 연결된 상태가 되는 현상을 말한다.

연결이 해제되지 않는 이유는 대표적으로 3가지가 있다.

1. 스택 내부에 네이티브 메서드가 있을 때

JVM은 네이티브 코드의 실행을 제어할 수 없기 때문에 힙 영역에 데이터를 백업할 수 없다.

2. 모니터 내에서 실행

JVM에서는 객체의 모니터는 캐리어 스레드가 점유하기 때문에 Pinned 이슈가 발생한다.

3. Critical Section 내에서 실행

JVM에서는 Critical Section 내에서 실행되는 작업을 보호하기 위해서 Pinned 이슈가 발생한다.

• Critical Section 내에서 실행되는 작업 종류

  • 클래스 로딩
  • GC 및 메모리 할당
  • 스레드 생성

참고

Oracle Virtual Threads

Java의 미래, Virtual Thread

Java 가상 스레드, 깊이 있는 소스 코드 분석과 작동 원리 1편 - 생성과 시작

LY Corporation Java 가상 스레드, 깊이 있는 소스 코드 분석과 작동 원리 2편 - 컨텍스트 스위칭

LY Corporation Java 가상 스레드, 깊이 있는 소스 코드 분석과 작동 원리 3편 - 고정 이슈와 한계

스프링캠프 2024 [Track 2] 1.동시성의 미래 - 코루틴과 버츄얼 스레드 (이상훈)

• 시세 플랫폼
  • 거래소의 전문을 가져와서 가공한 뒤 내부 서비스들에게 제공한다
  • 단순히 전문을 디코딩해서 전달하지 않고, 과거 데이터 누적 혹은 합성하여 제공한다.
  • 낮은 지연시간과 빠른 장애복구를 최우선시한다.
• 시세 플랫폼 구성

• 수신부:
  • 거래소가 제공하는 시세 데이터를 UDP 멀티캐스트 그룹에 접속해서 읽어온다.
  • 처리부에게 데이터를 제공할 때, header에 수신 시각을 포함해 처리부에서 총 처리 시간을 측정한다.
• 처리부
  • 비즈니스 로직이 모여있는 곳으로 처리 결과를 Redis에 저장하거나 실시간 정보를 서비스들에게 바로 전달한다.
  • 비즈니스 로직에 blockingI/O가 있기 때문에 처리 시간에 영향이 제일 크다
  • 코드 변경이 제일 빈번해서 장애 발생율이 제일 크다.
• 조회부
  • REST API를 서비스들에게 제공한다.

문제

1. 처리부가 장애가 생기면 모든 서비스에 영향을 미친다.

• 장애 시간 동안 데이터가 유실되거나 오염될 수 있기 때문에 장애를 복구해도 장애를 겪게 된다.

해결

• (처리부 + Redis)를 2개의 그룹으로 만든다.
  • 평상시에는 A그룹에만 트래픽을 할당하다가 A 그룹에 장애가 발생하면 바로 B 그룹으로 전환한다.
  • 각 그룹의 처리부를 두 개로 늘리고, 주키퍼를 통해 리더를 선출한다.
  • 중복 데이터를 막기 위해 각 그룹의 리더만 데이터를 처리한다.
  • 장점
    • 처리부 A와 B를 번갈아 배포하면서 둘 간의 차이를 비교할 수 있다.
    • 배포에 문제가 있다면 트래픽 전환 만으로 빠르게 롤백할 수 있다.

내 생각

• 피크 시간에만 그룹 A, B를 띄워 놓는 것인지, 아니면 24시간만 띄우는 것인지 궁금하다.

  그룹 B도 동시에 띄우는 것은 서버 자원을 2배로 사용하기 때문에 돈이 엄청 들 것이기 때문이다.

• 그룹 B의 Redis는 그룹 A의 Redis의 정보를 동기화하고 있는지 궁금하다.

  그룹 A에서 장애가 발생해 그룹 B로 전환됐을 때, 그룹 A의 Redis 정보가 동기화 돼있어야 서비스가 동작할 것이라 생각하기 때문이다.

  Redis Cluster에서 자동으로 동기화 해주는 것을 사용하나???

  ⇒ (5:08) 수신부는 모든 처리부에 데이터를 보내고 있다.

• 주키퍼란 뭘까 - 공식문서

  주키퍼를 처음 들어봐서 주키퍼가 무엇인지 공식문서를 보며 살짝만 알아봤다.

  주키퍼는 분산 환경에서 사용하는 중앙 집중식 서비스로, (설정 정보 관리, 분산 동기화, 그룹 서비스) 등의 다양한 서비스를 제공한다.

2. 수신부는 처리부의 개수만큼 반복해서 데이터를 보낸다.

• 현재 처리부는 4개의 pod이지만, 추후 서비스가 커져서 처리부가 많아질 수도 있다.

해결 : 메시지 브로커를 사용하자

• 수신부와 처리부를 디커플링할 수 있고, 수신부는 데이터를 한번만 전송해도 된다. 하지만 메시지 브로커로 인해 지연 시간이 더 높아질 수 있기 때문에 메시지 브로커의 선택이 중요하다.

메시지 브로커 후보

1. UDP 멀티캐스트

   라우터 설장과 쿠버네티스 배포 설정이 추가로 필요해서 빠른 개발을 위해 선택하지 않았다.

2. 카프카

   높은 처리량, 안정성, 사내에서 많이 사용 중이기 때문에 초반에는 카프카를 선택했다.

   하지만 자체 테스트 결과 Redis Pub/Sub이 지연 시간이 제일 짧아서 Redis로 변경했다.

   

3. Redis Pub/Sub

   낮은 지연시간, 사용하기 쉽고 편리한 커맨드 지원한다.

   다른 메시지 브로커들은 메시지를 받으면 Queue에 저장하지만 Redis Pub/Sub은 메시지를 받는 즉시 채널에 등록되어 있는 구독자들에게 보낸다.

   만약 구독자가 없다면 데이터를 전송하지 않기 때문에 데이터가 유실될 수 있지만, 지연 시간을 줄이는 면에서 보면 유리하다.

Redis Pub/Sub 동작 과정

• pubsub_channel이라는 Dictionary 타입의 변수에 모든 채널과 구독자 정보를 보관한다.
• 구독자가 특정 채널을 구독하면, 해당 채널의 연결 리스트에 구독자 정보를 추가한다.
• 발행자가 메시지를 보내면 Dictionary에서 채널을 찾고, 해당 연결 리스트를 모두 순회하면서 구독자에게 메시지를 전송한다.

내 생각

• UDP 멀티캐스트란 무엇일까?

  같은 데이터를 특정 그룹에게 보내줘야 할 때 사용하는 인터넷 프로토콜이다.

  UDP를 사용하기 때문에 빠르지만, 신뢰성을 보장하지 않는다.

• Redis Pub/Sub 의 단점은 없을까?

  메모리 기반으로 빠른 전송 속도를 제공하며, 실시간 메시지 전송이 필요한 애플리케이션에 적합하다.

  하지만 메시지를 메모리에만 저장하기 때문에 메시지가 한번 소비되면 해당 메시지는 사라지기 때문에 소비자가 메시지를 받지 못한다면 복구할 수 없다.

3. 처리부의 처리 속도 개선하기

처리부는 TCP 소켓을 통해 데이터 읽기와 비즈니스 처리(Blocking I/O)를 한다.

이 중, 데이터 읽는 속도가 지연 시간에 큰 영향을 준다.

TCP 흐름 제어는 송신 속도가 수신 속도 보다 빠른 경우 데이터 유실을 방지하기 위해, 수신자는 수신할 수 있는 윈도우 사이즈를 송신자에게 전달하고, 송신자는 해당 윈도우 사이즈 만큼 데이터를 보낸다. 따라서 수신부의 버퍼가 가득차면 레디스는 데이터를 전송하지 않아 지연시간이 늘어난다.

데이터 읽기 작업과 비즈니스 처리 로직을 별도의 스레드로 구분함으로써 데이터 읽기 작업을 빨리 끝낸다.

하지만 멀티 스레딩을 통해 비즈니스 처리를 한다면 데이터의 순서를 보장할 수 없기 때문에 EventLoopGroup을 사용해 이를 해결했다.

EventLoop는 Queue를 이용해 순서를 보장하고, 하나의 스레드만 사용하기 때문에 동기화가 필요없다.

하지만 어떤 EventLoop에서 처리해야 하는지 알아야 순서를 보장하기 때문에 반드시 종목 코드를 미리 알아야 했고, 이를 위해 json 파일을 객체로 변환해야 했다. 하지만 이 변환 작업이 NioEventLoop의 CPU 자원을 많이 사용해 추가적으로 지연이 발생했다.

Redis Pub/Sub에서 제공하는 채널을 사용해, 수신부에서 처리부의 EventLoop만큼 채널을 나누어 보내고, 수신부는 해당 채널명에 해당하는 EventLoop를 찾는다.

마치며

이를 통해, 수신부에서 보낸 데이터 순서를 그대로 유지하면서 객체 변환 작업을 없애면서 지연 시간을 줄였다.

Redis Pub/Sub 자료구조를 활용해서 목표했던 트래픽을 만족하는 실시간 서비스를 만든 과정을 들으면서 너무 재밌었다. 또, 여러 기술들 중 목표를 달성하기 위한 기술을 고려해서 선택하는 이러한 과정들이 흥미로웠다.

특히나 Redis를 많이 활용했는데, 이를 보고 Redis에 대해 잘 알고 있다면 문제 해결에 많이 유용하겠다는 생각했다.

지금까지 단순히 key-value 형식으로 Redis를 사용해봤는데, Redis의 여러 자료구조를 활용한 예시들을 보면서 공부해봐야겠다는 생각이 들었다.

참고

토스ㅣSLASH 23 - 실시간 시세 데이터 안전하고 빠르게 처리하기

https://zookeeper.apache.org/

https://f-lab.kr/insight/message-queue-system-comparison

사이드 프로젝트의 DDL을 살펴보다가 MySQL에 enum 타입이 있는 것을 보았다. 더 놀라운 건 문자열 형식으로 데이터가 저장되는데, 단순히 1bytes로 저장되는 것이다.

enum 타입이 무엇인지 궁금해 이를 알아보려 한다.

enum 타입이란

Java의 Enum 타입과 마찬가지로 MySQL에서 해당 column에 들어갈 수 있는 값을 제한하는 타입이다.

문자열이 자동으로 숫자로 인코딩되어 저장되기 때문에 1bytes로 저장된다. 쿼리 결과에는 숫자가 아닌 다시 문자열로 표시된다.

enum(’a’, ‘b’, ‘c’)인 경우 첫 번째 값부터 인덱스가 1로 시작한다. (’a’의 인덱스 1, ‘b’의 인덱스 2 …)

특징

• 빈 문자열과 null 값이 들어갈 수 있고, 잘못된 값을 enum 타입에 삽입하면 빈 문자열로 들어간다. (SQL 모드가 strict라면 오류 반환)

• enum 타입의 정렬은 값의 인덱스에 따라 정렬된다. null과 빈 문자열의 경우 가장 앞 순서로 정렬되며, ‘null > 빈 문자열 > enum 값’순으로 정렬된다.

• enum에 숫자를 사용하는 것은 권장하지 않는다. enum 타입이 숫자인 경우 해당 숫자가 인덱스로 해석될 수 있기 때문이다. 만약 숫자로 사용하는 경우엔 항상 리터럴(‘ ’)로 값을 추가해야 한다.

  ex) enum(’0’,’1’,’2’) 인 경우 실수로 1을 추가하면, ‘1’이 아닌 첫 번째 값인 ‘0’이 추가된다.

enum 타입의 장점

1. enum 타입은 단 1bytes만 차지하기 때문에 데이터 공간을 절약할 수 있다.
2. 쿼리 결과로는 문자열로 보이기 때문에 가독성이 좋다.

enum 타입의 단점

1. enum 타입의 값을 수정, 추가, 제거할 때, 모든 데이터를 돌면서 수정하기 때문에 데이터양이 많은 경우 시간이 오래 소모될 수 있다.

2. enum 타입이 숫자일 때, 데이터를 문자열로 추가하지 않는다면 원하지 않은 값이 들어갈 수 있다.

3. enum 타입은 일부 DBMS에만 있기 때문에 다른 DBMS로 이전할 때 마이그레이션 과정이 더욱 복잡해진다.

결론

1bytes의 장점이 있기 때문에 enum 타입의 값이 절대로 바뀌지 않을 확신이 있을 때만 사용하면 좋을 것 같다.

변경 가능성이 있거나 숫자일 경우엔 참조 테이블을 만들어서 사용하는 것이 좋을 것 같다.

13.3.5 The ENUM Type - MySQL Docs

MySQL Enum에 대한 정리

8 Reasons Why MySQL's ENUM Data Type Is Evil

• 컴퓨터에 따로 nGrinder를 설치하기 싫어서 도커를 사용하기로 결정했다.
• nGrinder문서를 참고해서 도커를 이용해서 nGrinder를 실행했다.

nGrinder 도커로 실행하기

docker pull ngrinder/controller
docker run -d -v ~/ngrinder-controller:/opt/ngrinder-controller --name controller -p 80:80 -p 16001:16001 -p 12000-12009:12000-12009 ngrinder/controller

• -d : 백그라운드로 컨테이너 실행
• -v ~/ngrinder-controller:/opt/ngrinder-controller : 호스트와 컨테이너 간의 볼륨을 마운트한다. 호스트의 ~ngrinder-controller 경로와 컨테이너 내부의 /opt/ngrinder-controller 경로를 연결한다.
• -p : 호스트의 포트와 컨테이너의 포트를 바인딩한다.
• ngrinder/controller : 해당 도커 이미지를 사용한다.
• --name controller : 컨테이너의 이름을 controller로 설정한다.(중요!!)

docker pull ngrinder/agent
docker run -d --name agent --link controller:controller ngrinder/agent

• --link controller:controller : 컨테이너 간에 네트워크를 설정하는 단계로 컨테이너 이름이 controller인 컨테이너와 네트워크를 연결한다. :controller 로 인해 agent 컨테이너 내부에서 controller라는 이름으로 controller 컨테이너에 접근할 수 있다.

*TMI : --link *

• Docker 공식 문서를 보면 --link 명령어 같은 경우 권장하지 않고, 사용자 정의 네트워크를 만들어서 컨테이너 간의 통신을 설정하는 것을 권장한다.

• 하지만, 도커 네트워크를 만드는 과정이 수반되기 때문에 귀찮아서 nGrinder 공식 문서대로 실행했다. (--link 를 사용해도 상관없다.)

문제

테스트 할 서버는 인텔리제이를 통해서 로컬에서 실행했고, 현재 상황은 아래와 같다.

도커로 nGrinder를 성공적으로 실행됐다면, localhost:80으로 nGrinder에 접속할 수 있다. 초기 ID, PW는 admin, admin으로 로그인할 수 있다.

nGrinder에 접속해 '127.0.0.1:8080/actuator/health'로 요청을 보내는 스크립트를 작성하고 실행한 결과, Connection refused 에러가 발생했다.

에러 원인을 알기 전에 nGrinder의 동작 방식은 간단하게 다음과 같다.

• 사용자가 스크립트를 작성해서 테스트 시나리오를 controller에게 전달한다.

  • controller는 여러 대의 agent를 가지고 있으며, 해당 스크립트를 agent에게 실행하도록 한다.

• agent는 타켓 서버에 요청을 보낸다.

에러 원인

원인

• agent에서 로컬 서버로 요청을 보낸다.

• agent는 컨테이너로 실행 중이다.

• agent 컨테이너를 실행할 때 8080 포트는 아무런 연결을 하지 않았다.

nGrinder의 동작 방식에서 agent에서 타켓 서버로 요청을 보낸다. 를 주목하자. 현재 agent는 컨테이너로 실행되고 있다. 그리고, 컨테이너에서 127.0.0.1을 요청하면 agent 컨테이너 내부의 localhost:8080으로 통신을 요청한다.

하지만, 서버는 agent 컨테이너 외부에 있고, agent 컨테이너 내부의 8080포트에는 서버가 실행되고 있지 않기 때문에 Connection refused 에러가 발생한다.

해결 방법1: ngrok으로 터널링하기 [비추천]

ngrok은 로컬에 실행된 서버를 터널링을 통해 외부에서 접속할 수 있도록 도와주는 플랫폼이다.

ngrok 또한 Docker image가 있기 때문에 다음과 같이 실행할 수 있다. - 공식문서 참고

docker run -it -e NGROK_AUTHTOKEN=xyz ngrok/ngrok:alpine http host.docker.internal:8080

• -e NGROK_AUTHTOKEN: 환경 변수를 설정한다. ngrok의 토큰으로 NGROK 회원 가입 이후에 받을 수 있다.

• ngrok/ngrok:alpine: alpine 버전의 ngrok 이미지를 사용한다.

• host.docker.internal:8080: 문서에 나온 것 같이 윈도우, 맥 환경으로 실행할 때, host.docker.internal:[포트번호] 형식으로 실행해야 한다. 현재 8080 포트로 서버를 실행하기 때문에 8080으로 실행한다.

  

• --log stdout: 컨테이너에서 발생하는 것을 로그로 표현하는 명령어로 이를 통해서 url을 확인한다. 하지만 ngrok 홈페이지에 들어가서 확인할 수 있어서 해당 명령어는 선택이다.

  

이제 테스트 스크립트를 127.0.0.1 대신 ngrok에서 제공해주는 url로 바꾸면 성공적으로 부하 테스트를 성공할 수 있다.

하지만 다음과 같은 이유로 ngrok을 추천하지 않는다.

• ngrok의 무료 버전은 다음과 같이 트래픽 제한이 있다.

nGrinder에서는 수많은 요청을 ngrok에게 보내면, 대부분의 요청을 429:too many request 에러로 반환해서 20초 내로 부하테스트가 종료된다.

해결법2: 127.0.0.1 대신 host.docker.internal 사용하기

문제의 원인은 agent 컨테이너가 Host의 로컬 서버를 접근하지 못하는 것이다. Docker 문서에서 host.docker.internal명령어를 통해서 Host에 접근할 수 있는 것을 알았다.

host.docker.internal: 컨테이너 내부에서 Host의 접근이 가능하도록 만드는 DNS이다.

부하 테스트 스크립트를 127.0.0.1:8080 → host.docker.internal:8080 으로 변경했고, 그 결과 agent가 Host의 서버로 요청을 잘 보내는 것을 볼 수 있다.

HTTPResponse response = request.GET("http://host.docker.internal:8080/actuator/health", params)

마치며

도커의 동작을 잘 몰랐기 때문에 많이 해맸다고 생각한다. 이 과정을 통해서 도커의 호스트와 컨테이너 관계에 대해 이해할 수 있었고, 역시 기본부터 잘 알아야 된다는 것을 다시금 깨달았다.

데이터 셋이 수천만 건일 때, 해당 데이터를 모두 전송하는 것은 서버의 부하가 생기고, 클라이언트 또한 랜더링 시간이 증가되고, 과도한 스크롤로 사용자 경험이 안좋아진다.

이를 해결하기 위해서 데이터를 일정 개수로 잘라서 제공하면서 서버와 클라이언트 모두 윈-윈할 수 있다.

페이지네이션 방식에는 무엇이 있을까?

크게 offset 방식과 cursor 방식이 있다.

Offset 방식

offset 방식은 offset과 limit를 사용하는 방식이다.

여기서 offset n(자연수)은 n만큼의 데이터를 건너뛰라는 의미이고, limit n(자연수)은 n만큼의 데이터를 가져오라는 의미이다.

따라서 select * from table limit 10 offset 10 쿼리가 있을 때, 20개의 데이터를 조회해서 앞의 10개의 데이터는 건너뛰고, 뒤의 10개의 데이터만 가져온다.

offset 방식의 장점은 다음과 같다.

• 구현하기가 매우 쉽다.

  몇 개의 데이터를 가져올 것인지 정해서 limit의 값을 설정하고, 현재 몇 페이지인지 입력받은 값을 offset의 값으로 설정하면 페이지네이션 구현이 끝난다.

offset 방식의 단점은 아래와 같다.

• 데이터 양이 많은 경우에 성능 저하가 발생한다.

  • offset 방식은 모든 데이터를 불러온 뒤, 해당 페이지 값이 아닌 데이터들은 모두 건너뛰어서 데이터를 전송한다.

  • 따라서 limit가 100이고, 현재 페이지가 200페이지라면 총 20,000개의 데이터를 조회한 후에 19,900의 데이터를 건너뛰어서 100개의 데이터를 전송한다.

• 데이터의 삽입과 삭제가 자주 일어나는 경우 데이터가 누락되거나 중복될 수 있다.

총 8개의 데이터가 있고, 데이터를 3개씩 페이징해서 보고 있다고 가정하자.

사용자A가 1페이지를 보고 있다면 A,B,C 총 3개의 데이터를 보고 있을 것이다.

사용자A가 1페이지를 보고 있는 중에 사용자B가 데이터 C를 제거한 후에 사용자 A가 2페이지로 넘어간다고 해보자.

그 결과, 6개의 데이터를 가져와서 앞의 3개의 데이터를 건너뛰기 때문에, 사용자 A는 D,E,F가 아닌 E,F,G를 보게 되어 이전 페이지로 넘어가지 않는다면 데이터 D를 볼 수 없다.

마찬가지로 A,B,C 사이에 데이터 B-2가 추가가 된다면, 2페이지에서는 C,D,E를 보게 되어 데이터 C를 중복해서 보게 된다.

Cursor 방식

특정 포인터(커서)를 사용해서 필요로 하는 데이터가 있는 위치부터 시작해서 필요한 만큼 데이터를 전송하는 방식이다.

커서로는 순차적으로 증가한 id 또는 타임 스탬프, 인코딩된 커서, 복합 커서 등이 있다.

created_at 날짜를 기준으로 10개씩 커서 기반 페이지네이션을 한다고 가정해보자.

그렇다면 첫 데이터를 다음과 같이 가져오고, 클라이언트에게 반환할 것이다.

GET /table?limit=10

select * from table order by created_at desc limt 10

{
    "tables": [...]
    "cursor": "last_created_at"
}

이후 다음 페이지부터 다음과 같이 가져올 것이다.

GET /table?limit=10&cursor="last_create_at"

select * from table where created_at < 'last_created_at' order by created_at desc limt 10

{
    "tables": [...]
    "cursor": "last_created_at2"
}

cursor 방식의 장점으로는 다음과 같다.

• 불필요한 데이터를 조회하지 않고, 필요한 데이터만 반환할 수 있다.

  • offset 방식과 달리 모든 데이터를 조회하지 않기 때문에 대규모 데이터 환경에 매우 효율적이다.

• 일관된 결과를 제공한다.

• 데이터가 변경되더라도 커서가 특정 위치를 가리키기 때문에 일관된 결과를 제공할 수 있다.

이러한 장점 덕분에 SNS와 같이 데이터가 자주 변동되는 경우에 cursor 방식이 유용하다.

cursor 방식의 단점은 다음과 같다.

• 구현이 복잡하다.

  • 커서 방식의 성능은 인덱스에 따라서 결정된다. 만약 인덱스가 없는 컬럼을 커서로 사용한다면 오히려 좋은 성능이 안 나올 수 있다.

    • 장점으로 일관된 결과를 제공한다고 했다. 이는 커서가 순차적인 id 또는 타임스탬프일 경우 쉽게 구현할 수 있지만, 커서가 복합 커서 또는 필터와 같이 사용하는 경우 일관된 결과를 제공하기 위해 개발자가 고려해야 할 점이 많아진다.

offset, cursor 언제 사용할까

offset 방식의 경우 다음과 같을 때 사용하면 좋다.

1. 데이터의 삽입과 삭제가 자주 일어나지 않거나 정확한 순서 보장이 필요하지 않을 때: offset 방식은 데이터 중복과 누락이 발생할 수 있다. 하지만 이점이 그렇게 중요하지 않다면 적용할 만하다.

2. 데이터 양이 많지 않을 때: 데이터 양이 많을 때 성능 저하가 발생하기 때문에 데이터 양이 적다면 적용할 만하다.

3. 페이지 건너뛰기가 필요할 때: 사용자가 페이지 번호를 입력해서 특정 페이지로 바로 이동하는 경우 offset을 사용하는 것이 유리하다.

cursor 방식의 경우 다음과 같을 때 사용하면 좋다.

1. 대규모 데이터일 때: offset 방식의 경우 대규모 데이터를 조회할 때 성능 저하가 발생하므로 대규모 데이터에선 커서 방식이 적합하다.

2. 정확한 순서 보장이 필요할 때: 커서 방식은 페이지 간 중복이나 누락 없이 데이터를 처리할 수 있기 때문에 데이터 순서를 유지해야 할 때 cursor 방식이 더 적합하다.

마치며

offset과 cursor 방식의 장 단점을 학습하면서 프로젝트 진행 상황, 데이터 양를 기준으로 페이징 방식을 선택할 것 같다.

참고

적절한 Pagination 방법은 뭘까?

Offset vs Cursor-Based Pagination: Which is the Right Choice for Your Project?

Cursor-based Pagination

간단한 Job을 하나 만들어 보고 잘 동작하는지 테스트 해보고, Step의 Reader, Processor, Writer 별로 단위 테스트를 하는 방식에 대해 학습해보자.

예제 설명

공공 데이터 포털에서 csv File을 하나 다운 받고, csv 파일 내에 있는 정보를 DB에 넣는 Job을 만들 것이다. 2024-9-13일 기준 파일데이터 1위 였던 경기도 의왕시_도서관 문화강좌을 사용하려 한다.

Job 전체 코드

먼저 해당 Job의 전체 코드는 다음과 같다.

• jobRepository: 해당 Job의 Execution 정보를 DB에 저장한다.
• platformTransactionManager: Spring Batch에서 커밋 및 롤백을 관리한다.
• reader를 통해서 csv 파일을 읽고, processor를 통해서 csv 데이터를 엔티티로 변환하며, writer를 통해서 item을 10개 단위로 DB에 저장한다.
• .faultTolerant().skip(FlatFileParseException::class.java).skipLimit(15): reader에서 데이터를 읽을 때 발생하는 에러를 최대 15개까지 용인하고, 초과 시에 Job을 중단한다.

@Configuration
class FlatFileBatchConfig(
    private val platformTransactionManager: PlatformTransactionManager,
    private val jobRepository: JobRepository,
    private val lectureEntityRepository: LectureEntityRepository
) {

    @Bean
    fun lectureJob(lectureStep: Step): Job =
        JobBuilder("lectureJob", jobRepository)
            .incrementer(RunIdIncrementer())
            .start(lectureStep)
            .build()

    @JobScope
    @Bean
    fun lectureStep(
        flatFileLectureReader: FlatFileItemReader<Lecture>,
        lectureToLectureEntityProcessor: ItemProcessor<Lecture, LectureEntity>,
        lectureEntityDBWriter: RepositoryItemWriter<LectureEntity>
    ) = StepBuilder("lectureStep", jobRepository)
        .chunk<Lecture, LectureEntity>(10, platformTransactionManager)
        .reader(flatFileLectureReader)
        .faultTolerant()
        .skip(FlatFileParseException::class.java)
        .skipLimit(15)
        .processor(lectureToLectureEntityProcessor)
        .writer(lectureEntityDBWriter)
        .build();

    @StepScope
    @Bean
    fun flatFileLectureReader(
        @Value("#{jobParameters['file.input']}") input: String
    ): FlatFileItemReader<Lecture> = FlatFileItemReaderBuilder<Lecture>()
        .name("flatFileLectureReader")
        .resource(FileSystemResource(input))
        .lineTokenizer(DelimitedLineTokenizer())
        .fieldSetMapper(LectureFieldSetMapper())
        .linesToSkip(1)
        .build()

    @Bean
    @StepScope
    fun lectureToLectureEntityProcessor(): ItemProcessor<Lecture, LectureEntity> {
        return ItemProcessor<Lecture, LectureEntity> { lecture ->
            LectureEntity(
                library = lecture.library,
                title = lecture.title,
                applyStartDate = lecture.applyStartDate,
                applyEndDate = lecture.applyEndDate,
                target = lecture.target,
                trainingStartDate = lecture.trainingStartDate,
                trainingEndDate = lecture.trainingEndDate,
                trainingStartTime = lecture.trainingStartTime,
                trainingEndTime = lecture.trainingEndTime,
                materialCost = lecture.materialCost,
                applyMemberCnt = lecture.applyMemberCnt,
                waitMemberCnt = lecture.waitMemberCnt
            )
        }
    }

    @StepScope
    @Bean
    fun lectureEntityDBWriter(): RepositoryItemWriter<LectureEntity> {
        return RepositoryItemWriterBuilder<LectureEntity>()
            .repository(lectureEntityRepository)
            .methodName("save")
            .build()
    }
}

Reader

FieldSet

파일에서 읽어온 데이터를 보다 효율적으로 다룰 수 있도록 도와주는 Spring Batch의 추상화 클래스이다. 파일에서 읽어온 데이터를 일관되게 처리할 수 있고, 다양한 데이터 타입을 지원한다.

File Reader

@StepScope
@Bean
fun flatFileLectureReader(
    @Value("#{jobParameters['file.input']}") input: String
): FlatFileItemReader<Lecture> = FlatFileItemReaderBuilder<Lecture>()
    .name("flatFileLectureReader")
    .resource(FileSystemResource(input))
    .lineTokenizer(DelimitedLineTokenizer())
    .fieldSetMapper(LectureFieldSetMapper())
    .linesToSkip(1)
    .build()

1. @Value("#{jobParameters['file.input']}") input: String : 배치 작업에서 사용할 csv파일의 경로를 JobParameter로 받는다.

   JobParameter: 배치 작업을 시작할 때 사용되는 파라미터의 집합을 가지고 있으며, 해당 파라미터들로 JobInstance를 식별한다.

2. name: ItemReader의 구현체가 여러 개 있을 수 있기 때문에, name을 통해서 ItemReader의 빈을 구분한다.

3. resource: .csv 파일 경로를 설정한다.

4. LineTokenizer: 각 줄을 FieldSet으로 변환하는 추강화로 다양한 파일 형식을 지원하기 위한 여러 구현체가 있다.

   • DelimitedLineTokenizer: 필드를 쉼표, 파이프 등과 같은 구분자를 통해 구분하여 파일을 처리한다. 기본 값으로는 , 로 구분한다.
   • FixedLengthTokenizer: 각 필드의 길이가 고정된 파일을 처리한다.

5. FieldSetMapper: mapFieldSet 메서드를 통해서 FieldSet을 원하는 객체로 변환한다.

    class LectureFieldSetMapper : FieldSetMapper<Lecture> {
        override fun mapFieldSet(fieldSet: FieldSet): Lecture {
            val library = fieldSet.readString(0)
              ...
            val materialCost = fieldSet.readInt(9)
                    ...
   
            return Lecture(
                ...
            )
        }
    }

6. linesToSkip(1): csv 파일의 첫번째 줄을 건너 뛴다. csv 파일을 보면 첫번째 줄은 column 명인 것을 볼 수 있다. 해당 내용은 필요없으니 건너뛴다.

Processor

@Bean
@StepScope
fun lectureToLectureEntityProcessor(): ItemProcessor<Lecture, LectureEntity> {
    return ItemProcessor<Lecture, LectureEntity> { lecture ->
        LectureEntity(
            library = lecture.library,
            title = lecture.title,
            applyStartDate = lecture.applyStartDate,
            applyEndDate = lecture.applyEndDate,
            target = lecture.target,
            trainingStartDate = lecture.trainingStartDate,
            trainingEndDate = lecture.trainingEndDate,
            trainingStartTime = lecture.trainingStartTime,
            trainingEndTime = lecture.trainingEndTime,
            materialCost = lecture.materialCost,
            applyMemberCnt = lecture.applyMemberCnt,
            waitMemberCnt = lecture.waitMemberCnt
        )
    }
}

ItemProcessor 인터페이스의 익명 객체를 만들어서 csv 파일에서 가져온 데이터를 DB에 저장하기 위한 Entity로 변환하는 과정을 진행한다.

Writer

@StepScope
@Bean
fun lectureEntityDBWriter(): RepositoryItemWriter<LectureEntity> {
    return RepositoryItemWriterBuilder<LectureEntity>()
        .repository(lectureEntityRepository)
        .methodName("save")
        .build()
}

Spring Batch에서 제공해주는 RepositoryItemWriter를 이용해서 Entity 객체를 DB에 저장한다.

Test Code

@Configuration
@EnableAutoConfiguration
@EnableBatchProcessing
class SpringBatchTestConfig {
}

@SpringBatchTest
@SpringBootTest(classes = [FlatFileBatchConfig::class, SpringBatchTestConfig::class])
class FlatFileBatchConfigTest {
    @Autowired
    lateinit var itemReader: FlatFileItemReader<Lecture>

    @Autowired
    lateinit var processor: ItemProcessor<Lecture, LectureEntity>

    @Autowired
    lateinit var itemWriter: RepositoryItemWriter<LectureEntity>

    @Autowired
    lateinit var lectureEntityRepository: LectureEntityRepository

    @Autowired
    lateinit var jobLauncherTestUtils: JobLauncherTestUtils

    val lectures: List<Lecture> = listOf(...)

    @Test
    fun `Lecture Job E2E 테스트`() {
        val execution = jobLauncherTestUtils.launchJob(defaultJobParameters())

        assertThat(execution.exitStatus).isEqualTo(ExitStatus.COMPLETED)
    }

    @Test
    fun `FlatFileItemReader 테스트`() {
        val stepExecution = MetaDataInstanceFactory.createStepExecution(defaultJobParameters())
        val expected = lectures.iterator()

        StepScopeTestUtils.doInStepScope(stepExecution) {
            var lecture: Lecture?
            itemReader.open(stepExecution.executionContext)
            while (itemReader.read().also { lecture = it } != null) {
                assertThat(lecture).isEqualTo(expected.next())
            }
            itemReader.close()
        }
    }

    @Test
    fun `LectureProcessor 테스트`() {
        val stepExecution = MetaDataInstanceFactory.createStepExecution(defaultJobParameters())
        val lectureEntities: List<LectureEntity> = listOf(...)

        StepScopeTestUtils.doInStepScope(stepExecution) {
            val result = lectures.map { processor.process(it) }
            // then
            assertThat(result).isEqualTo(lectureEntities)
        }
    }

    @Test
    fun `LectureDBWriter 테스트`() {
        val stepExecution = MetaDataInstanceFactory.createStepExecution(defaultJobParameters())
        val lectureEntities: List<LectureEntity> = listOf(...)

        StepScopeTestUtils.doInStepScope(stepExecution) {
            itemWriter.write(Chunk(lectureEntities))
        }

        val result: List<LectureEntity> = lectureEntityRepository.findAll()

        assertThat(result).isEqualTo(lectureEntities)
    }

    private fun defaultJobParameters(): JobParameters {
        val paramsBuilder = JobParametersBuilder()
        paramsBuilder.addString("file.input", "lectures-test.csv")
        return paramsBuilder.toJobParameters()
    }
}

1. 테스트 파일 생성

• @SpringBatchTest : Spring Batch에서 제공하는 테스트를 위한 어노테이션으로 JobLauncherTestUtils , JobRepositoryTestUtils 등을 통해 배치 Job을 쉽게 실행하고 검증할 수 있다.
• @EnableAutoConfiguration : Reader객체를 Bean으로 등록한다.
• @EnableBatchProcessing : 배치를 실행할 때 필요한 JobLauncher, JobRepository 등과 컴포넌트들을 자동으로 설정해준다.

@Configuration
@EnableAutoConfiguration
@EnableBatchProcessing
class SpringBatchTestConfig {
}

@SpringBatchTest
@SpringBootTest(classes = [FlatFileBatchConfig::class, SpringBatchTestConfig::class])
class FlatFileBatchConfigTest

2. JobParameter 설정: 테스트 csv 파일 경로를 JobParameter로 설정한다.

private fun defaultJobParameters(): JobParameters {
    val paramsBuilder = JobParametersBuilder()
    paramsBuilder.addString("file.input", "lectures-test.csv")
    return paramsBuilder.toJobParameters()
}

3. Job 테스트 코드 작성: JobLauncherTestUtils를 통해서 Job을 실행하고, 해당 Job이 잘 동작했는지 테스트한다.

@Autowired
lateinit var jobLauncherTestUtils: JobLauncherTestUtils

@Test
fun `Lecture Job E2E 테스트`() {
    val execution = jobLauncherTestUtils.launchJob(defaultJobParameters())

    assertThat(execution.exitStatus).isEqualTo(ExitStatus.COMPLETED)
}

4. Reader, Processor, Writer 테스트 코드 작성: MetaDataInstanceFactory를 통해서 stepExecution을 생성하고, 특정 Step의 reader, processor, writer를 테스트한다.

@Test
fun `FlatFileItemReader 테스트`() {
    val stepExecution = MetaDataInstanceFactory.createStepExecution(defaultJobParameters())
    val expected = lectures.iterator()

    StepScopeTestUtils.doInStepScope(stepExecution) {
        var lecture: Lecture?
        itemReader.open(stepExecution.executionContext)
        while (itemReader.read().also { lecture = it } != null) {
            assertThat(lecture).isEqualTo(expected.next())
        }
        itemReader.close()
    }
}

@Test
fun `LectureProcessor 테스트`() {
    val stepExecution = MetaDataInstanceFactory.createStepExecution(defaultJobParameters())
    val lectureEntities: List<LectureEntity> = listOf(...)

    StepScopeTestUtils.doInStepScope(stepExecution) {
        val result = lectures.map { processor.process(it) }
        // then
        assertThat(result).isEqualTo(lectureEntities)
    }
}

@Test
fun `LectureDBWriter 테스트`() {
    val stepExecution = MetaDataInstanceFactory.createStepExecution(defaultJobParameters())
    val lectureEntities: List<LectureEntity> = listOf(...)

    StepScopeTestUtils.doInStepScope(stepExecution) {
        itemWriter.write(Chunk(lectureEntities))
    }

    val result: List<LectureEntity> = lectureEntityRepository.findAll()

    assertThat(result).isEqualTo(lectureEntities)
}

마치며

Job을 만들면서 Job E2E 테스트 코드와 Reader, Writer, Processor 별로 단위 테스트를 작성해볼 수 있었다.

다음에는 DB에서 데이터를 읽고 쓸 때, Paging 하면서 데이터를 불러오는 방법에 대해 학습 해보려 한다.

참고

경기도 의왕시_도서관 문화강좌

https://www.baeldung.com/spring-batch-testing-job

배치 프로그램은 데이터를 실시간으로 처리하는 것이 아닌, 일괄적으로 모아서 한번에 처리하는 작업을 말한다. 대표적인 배치 작업으로 정산 시스템, (월별,분기별,연간) 보고서 자동 생성 등이 있다.

배치 프로그램은 일관되고 정확한 결과를 보장하며, 시스템 사용률을 고려해서 배치 작업을 실행하여 서버 리소스를 효율적으로 사용할 수 있다.

Spring Batch란?

Spring Batch는 이러한 배치 프로그램을 개발자가 쉽게 구현할 수 있도록 지원한다. 관심사를 명확하게 분리해서 각 관심사 별 인터페이스를 제공함으로써 배치 프로그램을 쉽게 구현할 수 있다.

Spring Batch를 스케줄러로 오해할 수도 있다. 하지만 Spring Batch는 스케줄링 프레임워크가 아니며, 스케줄러를 대체하기 보다는 스케줄러와 함께 작동하도록 설계되었다.

Spring Batch 구조

Job Launcher

Job을 실행할 때 필요한 매개변수를 받아서 Job을 실행한다.

JobRepository

JobInstance, JobExecution, StepExecution과 같은 배치 관련 객체들에 대한 CRUD 작업을 처리한다.

Job

작업의 단위를 뜻하며, 하나의 Job에는 여러 개의 Step이 있다.

Step

하나의 독립적이고 순차적인 처리 단계를 의미하는 객체이다.

• Step의 흐름을 제어할 수 있다. Job에는 여러 Step이 있을 수 있는데, 특정 Step이 실패, 성공 시 실행 할 Step을 설정할 수 있고, 특정 조건에 따라서 Step을 스킵할 수 있도록 만들 수 있다.

• 지연 바인딩을 지원한다. 이를 통해서 실행 시점에 사용할 bean을 실행 중에 주입하면서, 유연하게 배치 작업을 설정할 수 있다.

  Job이 실행될 때에만 bean이 생성되는 @JobScope , Step이 실행될 때에만 bean이 생성되는 @StepScope가 있다.

  @JobScope는 멀티 스레드에서의 사용을 권장하지 않는다. Bean같은 경우에 프록시로 동작하기 때문에 단일 인스턴스만 생성된다. 멀티 스레드 환경에서는 동일한 인스턴스를 공유하게 되고, 이로 인해 데이터 충돌이나 불일치가 발생할 수 있다. 그에 반면 @StepScope의 Bean은 스레드마다 인스턴스가 생성되기 때문에 thread-safe하다.

• 각 Step 별로 Chunk 기반으로 실행되거나, Tasklet으로 실행될 수 있다.

Chunk 기반 Step

데이터를 하나씩 읽어서 일정한 묶음으로 모아서 처리한다.

1. 데이터를 하나씩 읽어서 묶음을 만든다.

2. 해당 묶음을 Processor로 전달해, 변환하거나 필터링한다.

3. 변환된 데이터만 Writer로 전달하여 데이터를 저장한다.

4. 데이터를 쓰고 나면 트랜잭션을 커밋해서 처리된 내용을 확정한다.

Tasklet 기반 Step

데이터를 반복해서 읽고 쓰는 것이 아닌 단일 작업을 처리해야 할 때 TaskletStep을 사용할 수 있다.

Reader

File, XML, Database 등 다양한 방식으로 데이터를 하나씩 순차적으로 읽어오는 역할을 한다. 더 이상 읽을 데이터가 없을 경우 null을 반환하여 read 연산을 중단한다.

Processor(Optional)

데이터를 변환, 필터링, 검증하는 비즈니스 로직을 담당한다. 여러 Processor를 체이닝해서 만든 복잡한 변환 로직이나, 데이터 필터링 로직을 담는다. 필터링을 할 때 null을 반환하면 해당 데이터는 Writer로 전달되지 않는다.

Writer

데이터를 출력하거나 저장하는 역할을 하며, 출력 대상으로 파일, 데이터베이스, 큐 등 다양한 형식이 될 수 있다. 데이터를 한번에 한 항목씩 읽어오는 Reader와 달리, Writer는 데이터를 일정량씩 모아서 한꺼번에 처리하면서 성능을 향상시킨다.

마치며

Spring Batch에 관해 간단하게 학습할 수 있었다.

다음은 각 Step 별로 구현 및 단위 테스트를 작성하는 방식에 대해 학습하고자 한다.

참고

Spring Batch Introduction

The Domain Language of Batch

ItemReader

ItemWriter

Item processing

Configuring a Step

Redis로 동시성을 제어하는 방식은 앞서 학습한 네임드 락 방식과 똑같이 특정 Key 값이 등록되어 있는지 확인하고, 잠금을 획득하는 방식이다. 이를 위해서는 원자성을 보장해야 한다.

원자성이란 더 이상 쪼개질 수 없는 성질이란 뜻으로 Redis의 명령어는 기본적으로 원자성을 보장한다. 하지만 여러 명령어를 실행해야 할 때는 원자성을 보장할 수 없는데, Redis는 이를 Lua Script와 Transaction을 통해서 원자성을 보장한다.

스프링에서 Redis를 사용하기 위해 아래와 같이 의존성을 추가하면 Lettuce 라이브러리 의존성이 추가된다.

implementation("org.springframework.boot:spring-boot-starter-data-redis")

Lettuce로 동시성 제어하기

Redis의 set nx 명령어를 통해서 잠금을 획득하는 연산의 원자성을 보장할 수 있다.

set nx 명령어는 key가 존재하는지 확인, 존재하지 않으면 값을 설정한다. ****

Lettuce에서 setIfAbsent 명령어를 통해서 set nx 명령어를 사용할 수 있다.

// SET key value NX
setIfAbsent(key, "lock")

// SET key value NX EX timeOut
// key 값이 없다면, time out 동안 key 값을 등록한다.
// time out이후에는 자동으로 key 값이 사라진다.
setIfAbsent(key, "lock", Duration.ofSeconds(timeOut))

구현

@Component
class LettuceLock(
    private val stringRedisTemplate: StringRedisTemplate
) {

    fun lock(key: String, timeOut: Long): Boolean {

        while (true) {
            val result = stringRedisTemplate.opsForValue()
                .setIfAbsent(key, "lock", Duration.ofSeconds(timeOut)) ?: return false
            if (result) return result
        }
    }

    fun unLock(key: String): Boolean {
        return stringRedisTemplate.delete(key)
    }
}

잠금을 획득하기 위해서는 지속적으로 잠금을 요청해야 하기 때문에 스핀락 방식을 통해 구현한다.

스핀락이란 ‘임계 영역에 진입이 불가능할 때, 진입이 가능할 때까지 루프를 돌면서 재시도하는 방식’이다.

하지만 이는 Redis에 지속적으로 요청을 보내기 때문에 Redis에 부하를 준다.

이를 해결하기 위해서 Redis의 라이브러리로 Redisson을 사용할 수 있다.

Redisson 라이브러리로 대체하기

Redisson은 Redis 라이브러리로 Redis의 자료구조인 String, Hash, List, Sorted set 등을 모두 지원하고, 동시에 Lock 과 같은 다양한 구현체를 쉽게 사용할 수 있어 다른 Redis 라이브러리 보다 러닝 커브가 낮다.

Redisson은 스핀락 방식이 아닌 Pub/Sub 방식을 통해서 잠금을 획득하기 때문에 Redis에 부하를 적게 준다. 잠금이 사용 중일 때 해당 잠금을 구독하고, 잠금을 반납할 때 구독하고 있는 스레드들에게 잠금을 사용해도 된다는 것을 알린다.

또한, 아래와 같이 루아 스크립트를 통해 원자성을 보장하면서 잠금을 획득하는 것을 볼 수 있다.

// RedissonLock.java
<T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
    return this.evalWriteSyncedAsync(
            this.getRawName(), 
            LongCodec.INSTANCE, 
            command, 
        "if (redis.call('exists', KEYS[1]) == 0) then " +
            "redis.call('hset', KEYS[1], ARGV[2], 1); " +
            "redis.call('pexpire', KEYS[1], ARGV[1]); " +
            "return nil; " +
        "end; " +
        "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
            "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
            "redis.call('pexpire', KEYS[1], ARGV[1]); " +
            "return nil; " +
        "end; " +
        "return redis.call('pttl', KEYS[1]);",
        Collections.singletonList(this.getRawName()), 
        new Object[]{unit.toMillis(leaseTime), this.getLockName(threadId)}
        );
}

구현

@Component
class RedissonLock(
    private val redissonClient: RedissonClient
) {
    fun lock(key: String, timeOut: Long): Boolean {
        val lock = redissonClient.getLock(key)
        val result = lock.tryLock(timeOut, 5, TimeUnit.SECONDS)
        return result
    }

    fun unLock(key: String): Boolean {
        val lock = redissonClient.getLock(key)
        lock.unlock()
        return true
    }
}

• tryLock을 통해서 잠금을 획득할 수 있다. 매개변수로 다음 3가지를 넣는다.

  1. 몇 초 동안 락을 대기할 것인지

  2. 잠금을 획득하고, 몇 초 뒤에 자동으로 잠금을 반납할 것인지

  3. 시간 단위는 무엇인지

마무리

지금까지 Redis를 이용한 분산 락 구현 방식에 대해 학습했다. 이전에 학습한 네임드 락 방식보다 확실히 쉽고, 커넥션 풀을 따로 관리 안해도 된다는 점에서 편했다. 하지만 커넥션 풀 관리 대신 이젠 Redis를 관리해야 하는 인프라 비용이 추가된다는 단점이 있다.

이제 동시성 문제를 해결할 때, 앞서 학습한 내용들의 장단점을 잘 파악해서 현재 상황에 맞는 트레이드 오프를 할 수 있도록 노력할 것이다.

참고

Redis - SET

대규모 처리 시 Redis 연산의 Atomic을 보장하기

스핀락 - 위키 백과

8. Distributed locks and synchronizers

레디스와 분산 락(1/2) - 레디스를 활용한 분산 락과 안전하고 빠른 락의 구현

MySQL에서 제공하는 잠금 방식으로 잠금의 대상이 테이블이나 레코드와 같은 데이터베이스 객체가 아닌, 임의의 문자열에 대해 잠금을 설정할 수 있다.

네임드 락 관련 함수

• GET_LOCK(str, timeout)
  • 문자열 str에 대해 잠금을 획득한다. 이미 잠금을 사용 중이라면, timeout 동안 대기하고, 만일 timeout이 음수라면 무한정 대기한다.
  • 특정 세션에서 여러 개의 네임드 락을 중첩해서 사용할 수 있기 때문에 데드락이 발생할 수 있다.
  • 여러 클라이언트가 락을 기다리는 경우, 락을 얻는 순서는 정의되어 있지 않기 때문에 락 요청이 발행된 순서대로 락이 얻어진다고 가정해서는 안된다.
  • 네임드 락은 트랜잭션이 커밋되거나 롤백될 때 해제되지 않는다. 따라서 RELEASE_LOCK() 함수를 사용해서 명시적으로 해제해야 한다.
  • 함수의 반환 값은 다음과 같다.
    • 1: 잠금을 성공적으로 얻은 경우
    • 0: 타임아웃된 경우
    • NULL: 오류가 발생한 경우(ex. 메모리 부족, 스레드 종료 등)
• RELEASE_LOCK(str)
  • 현재 세션에서 설정된 락만 해제할 수 있으며, 다른 세션에서 설정한 락을 해제할 수 없다.
  • 함수의 반환 값은 다음과 같다.
    • 1: 락이 성공적으로 해제된 경우
    • 0: 락이 현재 스레드에 의해 설정된 것이 아닌 경우 (이 경우 락이 해제되지 않는다).
    • NULL: 지정한 이름의 락이 존재하지 않는 경우. (ex. 해당 문자열로 GET_LOCK()을 통해 락이 획득된 적이 없거나, 이전에 이미 해제된 경우일 수 있다.)
• RELEASE_ALL_LOCKS()
  • 해당 세션에서 획득한 네임드 락을 RELEASE_ALL_LOCKS() 을 통해서 한 번에 모두 해제할 수도 있다.
  • 함수의 반환 값은 다음과 같다.
    • 해제된 잠금 수를 반환한다. (없는 경우 0을 반환한다.)
• SELECT IS_FREE_LOCK(str)
  • 해당 문자열에 대해 잠금이 설정돼 있는지 확인한다.
  • 이 함수의 반환 값은 다음과 같다.
    • 1: 사용 가능한 문자열인 경우
    • 0: 다른 세션에서 사용 중인 문자열인 경우
    • NULL: 오류가 발생한 경우

스프링에서 네임드 락 사용하기

네임드 락 로직 구현

다음과 같이 JPA를 통해 네임드 락을 획득 및 반납할 수 있었지만, 이후 레디스를 통한 분산 락 방식도 구현할 것이기 때문에 분산 락 인터페이스를 만들었다.

• @Query 를 통한 잠금 획득 및 반납

    interface TicketRepository : JpaRepository<Ticket, Long> {
  
        @Query("SELECT GET_LOCK(:key, :timeout)", nativeQuery = true)
        fun lock(@Param("key") key: String, @Param("timeout") timeout: Int): Long?
  
        @Query("SELECT RELEASE_ALL_LOCKS()", nativeQuery = true)
        fun unLock()
    }

분산 락 인터페이스는 다음과 같다.

interface DistributedLock {
    fun lock(key: String, timeOut: Int): Boolean
    fun unLock(): Boolean
}

네임드 락의 구현체는 다음과 같다.

@Component
class DatabaseNamedLock(
    private val entityManager: EntityManager
) : DistributedLock {

    override fun lock(key: String, timeOut: Int): Boolean {
        val result = entityManager.createNativeQuery(GET_LOCK)
            .setParameter(1, key)
            .setParameter(2, timeOut)
            .singleResult as? Long

        if (result == null) return false
        return result == 1L
    }

    override fun unLock(): Boolean {
        val result = entityManager.createNativeQuery(RELEASE_LOCK)
            .singleResult as? Long

        if (result == null) return false
        return result >= 1
    }

    companion object {
        private const val GET_LOCK = "SELECT GET_LOCK(?, ?)"
        private const val RELEASE_LOCK = "SELECT RELEASE_ALL_LOCKS()"
    }
}

• 테스트 코드를 작성할 때, 자주 사용했던 EntityManager를 통해서 네임드 락 쿼리를 작성했다.
• 잠금 획득: key 값에 대한 잠금 획득을 timeout 동안 시도하고, 잠금 획득의 성공 여부를 반환한다.
• 잠금 반납: 해당 세션에서 얻었던 잠금을 RELEASE_ALL_LOCKS()을 통해서 모두 반납하고, 반납의 성공 여부를 반환한다. 이 때, RELEASE_ALL_LOCKS()의 반환 값은 반납한 잠금의 개수이므로 한 세션에서 여러 번 잠금을 획득했다면 1 이상의 값이 나오는 것을 주의하자.

네임드 락 적용

네임드 락을 적용한 코드는 다음과 같다.

@Service
class TicketService(
    private val ticketRepository: TicketRepository,
    private val ticketUserRepository: TicketUserRepository,
    private val distributedLock: DistributedLock,
    transactionManager: PlatformTransactionManager
) {
    private val transactionTemplate = TransactionTemplate(transactionManager)

    fun issueTicketWithNamedLock(ticketId: Long, name: String) {
        val ticket = findTicketById(ticketId)

        val lockAcquired = distributedLock.lock(ticketId.toString(), 10)
        if (!lockAcquired) {
            println("잠금을 획득하지 못했습니다.")
            return
        }

        transactionTemplate.execute {
            val ticketCount = countTicket(ticketId)
            if (ticketCount >= 100) {
                distributedLock.unLock()
                throw IllegalStateException("티켓 소진")
            }
            val ticketUser = TicketUser(ticket = ticket, name = name)
            ticketUserRepository.save(ticketUser)
        }

        distributedLock.unLock()
    }
}

• synchronized를 통해 잠금을 걸 때와 마찬가지로 잠금 획득 → 트랜잭션 시작 순으로 진행하기 위해서 프로그래밍 방식으로 트랜잭션을 관리한다.
• 존재하는 티켓인지 확인한 후, 해당 티켓 id에 대한 잠금을 획득한다.
• 잠금을 획득한 후, 트랜잭션을 시작해서 쿠폰 발급 로직을 수행한다.
• 로직이 끝난 후에 잠금을 반납한다.
• 로직 수행 중에 만약 에러가 발생하면 별도로 잠금을 반납]한다.

    @Test
    fun `티켓 요청을 300번 요청했을 때, 100개의 티켓만 생성되어야 한다`() {
        val numberOfThread = 300
        val executorService = Executors.newFixedThreadPool(numberOfThread)
        val latch = CountDownLatch(numberOfThread)

        repeat(numberOfThread) { idx ->
            executorService.submit {
                try {
                    ticketService.issueTicketWithNamedLock(1L, "USER $idx")
                } finally {
                    latch.countDown();
                }
            }
        }

        latch.await()

        val result1 = ticketService.countTicket(1L)

        assertThat(result1).isEqualTo(100)
    }

결과적으로 300개의 요청이 동시에 들어 올 때, 성공적으로 100개의 티켓만 발급하는 것을 볼 수 있다.

그렇다면 아래와 같이 여러 티켓이 동시에 요청이 들어올 때는 어떻게 될까?

@Test
fun `3종류의 티켓 요청을 동시에 900번 요청했을 때, 각 티켓 당 100개의 티켓만 생성되어야 한다`() {
    val numberOfThread = 900
    val executorService = Executors.newFixedThreadPool(numberOfThread)
    val latch = CountDownLatch(numberOfThread)

    repeat(numberOfThread) { idx ->
        executorService.submit {
            try {
                ticketService.issueTicketWithSynchronized((idx % 3) + 1L, "USER $idx")
            } finally {
                latch.countDown();
            }
        }
    }

    latch.await()

    val result1 = ticketService.countTicket(1L)
    val result2 = ticketService.countTicket(2L)
    val result3 = ticketService.countTicket(3L)

    assertThat(result1).isEqualTo(100)
    assertThat(result2).isEqualTo(100)
    assertThat(result3).isEqualTo(100)
}

이전 글의 주제였던 synchronized 보다 요청을 빨리 수행할 것이라 생각하여 테스트를 실행했지만 다음과 같은 에러가 생기면서 데드락이 발생했다.

SQL Error: 3058, SQLState: HY000
Deadlock found when trying to get user-level lock; try rolling back transaction/releasing locks and restarting lock acquisition.

왜 데드락이 생겼을까?

데드락 문제를 파악하기 전에 스프링의 커넥션 풀에 대해 간단하게 설명하고자 한다.

스프링은 시작과 동시에 데이터베이스와 연결된 커넥션을 미리 생성해서 커넥션 풀을 만들어 두고, 쿼리를 실행할 때 해당 커넥션 풀에서 커넥션을 꺼내 쓰고 있다.

데이터베이스 서버에서는 각 커넥션 별로 세션을 만들고, 이후 커넥션을 통한 모든 요청이 해당 세션을 통해서 실행된다.

스프링의 기본 커넥션의 개수는 10개이기 때문에 데이터베이스 서버에도 총 10개의 세션이 있다.

다시 데드락 문제로 넘어가자. 해당 데드락 발생 원인은 아래와 같다.

“A” 잠금을 가지고 있는 세션 1과 “B” 잠금을 가지고 있는 세션 2가 있을 때, 세션 1은 “2”를, 세션 2는 “1”의 잠금을 획득하려 해서 데드락이 발생했다.

하지만, 서비스 로직을 보면, 잠금을 딱 한번 획득한 후에 트랜잭션이 끝나고 반납을 하고 있는데 왜 데드락이 생겼을까?

MySQL 로그 파일을 보고 원인을 알아낼 수 있었다.

2486 Query    SELECT GET_LOCK('2', 10)
2490 Query    SELECT GET_LOCK('3', 10)
2484 Query    SELECT GET_LOCK('2', 10)
2487 Query    SELECT GET_LOCK('3', 10)
2489 Query    SELECT GET_LOCK('2', 10)
2488 Query    SELECT GET_LOCK('1', 10)
2485 Query    SELECT GET_LOCK('3', 10)
2482 Query    SELECT GET_LOCK('2', 10)
2491 Query    SELECT GET_LOCK('1', 10)
2483 Query    SELECT GET_LOCK('3', 10)
2487 Query    SELECT GET_LOCK('1', 10)  <- 여기서 부터 동일한 세션을 사용한다.
2488 Query    SELECT GET_LOCK('1', 10)
2482 Query    SELECT GET_LOCK('3', 10)
2488 Query    SELECT GET_LOCK('1', 10)
2488 Query    SELECT GET_LOCK('2', 10)
2488 Query    SELECT GET_LOCK('1', 10)
2488 Query    SELECT GET_LOCK('3', 10)
2488 Query    SELECT GET_LOCK('1', 10)

위 테스트 코드를 실행한 결과, 10개의 세션에서 락을 점유한 후에도 해당 세션의 작업이 끝날 때까지 대기하지 않고, 동일한 세션에서 추가로 잠금을 획득하려 시도하는 것을 알 수 있었다.

이로 인해 데드락이 발생했다.

마치며

해당 문제를 해결해보려 2일을 시도해봤지만, 해결하지 못했다.. 시간을 오래 잡아서 나중에 다시 알아보려 한다…

이렇게 한동안 머리 아팠던 네임드 락 부분이 끝이 났다. 학습을 하면서 네임드 락의 아쉬운 점이 다음과 같았다.

데이터베이스 커넥션 관리가 어렵다.

위 데드락이 발생한 것도 그렇고, 네임드 락을 트랜잭션 중간에 사용할 때, 트랜잭션을 최소 2개를 사용해야 한다. (자세한 내용은 해당 블로그를 확인하자)

또한, synchronized와 마찬가지로 분산 DB 환경에서 잠금이 어느 DB에 획득했는지 알 수 없기 때문에 잠금을 획득 및 반납이 어려워 진다.

위와 같은 이유로 네임드 락을 통한 분산 락은 지양할 것 같다.

다음에는 Redis를 이용한 분산 락 방식에 대해 정리해보려 한다.

참고

Real MySql 8.0 (1권) 05. 트랜잭션과 잠금

14.14 Locking Functions

[E-commerce] 동시성 문제 해결하기 (비관적 락, 네임드 락, 분산 락)

MySQL을 이용한 분산락으로 여러 서버에 걸친 동시성 관리

1-1. 트랜잭션 추상화

JPA, JDBC 등 데이터 접근 방식마다 같은 기능을 다르게 구현하고 있기 때문에 데이터 접근 방식 변경 시 여러 코드에 변경 사항이 전파됐다.

이를 해결하기 위해 PlatformTransactionManager 인터페이스로 추상화해서 여러 데이터 접근 기술을 쉽게 변경 가능하도록 만들었다.

1-2. 리소스 동기화

트랜잭션을 유지하려면 트랜잭션의 시작부터 끝까지 같은 데이터베이스 커넥션을 유지해야 한다.

이전에는 파라미터로 커넥션을 전달하면서 데이터베이스 커넥션을 유지했지만, 이러한 방식은 코드가 지저분해지는 등 여러 가지 단점이 많았다.

각각의 스레드마다 별도의 저장소가 부여되기 때문에 특정 스레드만 해당 데이터에 접근할 수 있어서 멀티 스레드 환경에서 안전하게 커넥션을 보관할 수 있다.

이러한 성질을 이용해 스레드 로컬을 사용해서 데이터베이스 커넥션을 동기화한다.

2. 스프링 트랜잭션 동작 방식

2-1. 트랜잭션 시작

1. 서비스 계층에서 transactionManager.getTransaction()을 호출해서 트랜잭션을 시작한다.

• transactionManager.getTransaction() 동작 과정

  2. 데이터 소스를 통해서 DB 커넥션을 획득한다.

  3. 해당 커넥션을 수동 커밋 모드로 변경해서 실제 DB 트랜잭션을 시작한다.

  4. 커넥션을 트랜잭션 동기화 매니저에게 전달한다.

  5. 트랜잭션 동기화 매니저는 전달받은 커넥션을 스레드 로컬에 보관한다.

💡 데이터 소스(데이터베이스 커넥션 풀)

데이터 소스에서 데이터베이스 커넥션을 관리한다.

데이터베이스 커넥션을 생성하는 비용이 크기 때문에 매번 커넥션을 생성하는 것은 비효율적이다. 이를 해결하기 위해서 커넥션을 미리 확보해서 스레드 풀과 같은 커넥션 풀에 보관(기본 값은 10개)한다.

애플리케이션은 매번 커넥션을 생성하지 않고, 미리 생성해둔 커넥션을 커넥션 풀에서 꺼내 쓰고 사용 후에 반환하면 된다. 이를 통해서 커넥션을 생성하는 비용을 줄일 수 있고, 무한정으로 DB에 커넥션이 연결되는 것을 막으면서 DB를 보호하는 효과도 있다.

2-2. 비즈니스 로직 실행

6. 서비스는 비즈니스 로직을 실행하면서 레포지토리 메서드를 호출한다.

7. 레포지토리 메서드는 SQL을 데이터베이스에 전달하기 위해 커넥션이 필요하다. 이때, DataSourceUtils.getConnection()을 사용해 트랜잭션 동기화 매니저에 보관된 커넥션을 꺼내서 사용한다. 이 과정으로 자연스럽게 동일한 커넥션을 사용하면서 트랜잭션도 유지된다.

8. 커넥션을 획득하고, SQL을 DB에 전달해서 실행한다.

2-3. 트랜잭션 종료

9. 비즈니스 로직이 끝나고, 서비스 계층에서 transactionManager.commit() 또는 rollback()을 실행한다.

• transactionManager.commit() 또는 rollback() 동작 과정

  10. 트랜잭션 AOP 프록시가 커밋 또는 롤백을 통해 트랜잭션을 종료하기 위해서 트랜잭션 동기화 매니저를 통해 동기화된 커넥션을 획득한다.

  11. 획득한 커넥션을 통해 트랜잭션을 커밋 또는 롤백한다.

  12. 스레드 로컬 등과 같이 사용한 모든 리소스를 정리한다.

3. 프로그래밍 방식의 트랜잭션 관리

3-1. DataSource

1번부터 12번 과정을 다음 코드와 같이 만들 수 있다.

@Service
class Service(private val dataSource: DataSource) {
    fun dataSource(ticketId: Long, name: String) {
        // 2
        val connection = dataSource.connection
        // 3
        connection.autoCommit = false
        try {
            // 6~8
            bussinessLogic()
            // 11
            connection.commit()
        } catch (e: Exception) {
            // 11
            connection.rollback()
        } finally {
            // 12
            connection.close()
        }
    }
}

1. 데이터 소스를 주입 한다.

2. 데이터 소스로부터 DB 커넥션을 획득한다.

3. autoCommit = false를 통해서 수동 커밋 모드로 변경한다.

4. try catch 구문 안에서 비즈니스 로직을 실행하고, 오류 없이 마무리 되면 트랜잭션을 커밋한다.

5. 만일 오류가 발생하면 트랜잭션을 롤백한다.

6. 커밋 또는 롤백을 한 뒤, connection.close()를 통해서 커넥션을 반납한다.

3-2. TransactionManager

DB 커넥션 획득 및 반납, 수동 커밋 모드 설정을 별도로 안 해도 돼서 DataSource 방식 보다 비교적 편하게 트랜잭션을 관리할 수 있다.

@Service
class Service(private val transactionManager: PlatformTransactionManager) {
    fun transactionManager(ticketId: Long, name: String) {
        // 1 ~ 6
        val status = transactionManager.getTransaction(DefaultTransactionDefinition())
        try {
            bussinessLogic()
            // 9 ~ 12
            transactionManager.commit(status)
        } catch (e: Exception) {
            // 9 ~ 12
            transactionManager.rollback(status)
        }
    }
}

1. transactionManager.getTransaction() 을 통해서 DB 커넥션을 획득한다.

2. 비즈니스 로직을 실행한다.

3. 트랜잭션을 커밋 또는 롤백한다.

3-3. TransactionTemplate

TransactionTemplate은 DB 커넥션 획득 및 반납, 커밋 또는 롤백을 별도로 안 해도 된다.

@Service
class Service(transactionManager: PlatformTransactionManager) {
    private val transactionTemplate = TransactionTemplate(transactionManager)

    fun transactionTemplate(ticketId: Long, name: String) {
        transactionTemplate.execute {
            bussinessLogic()
       }
    }
}

이렇게 총 3가지 프로그래밍 방식의 트랜잭션 관리 방법을 알아 보았다.

하지만 프로그래밍 방식의 트랜잭션 관리는 비즈니스 로직에 트랜잭션 관리 로직이 추가되어 순수한 비즈니스 로직만 남길 수 없다는 단점이 있다.

이를 선언적 트랜잭션 관리 방법을 통해 보완할 수 있다.

4. 선언적 트랜잭션 관리

4-1. @Transactional

선언적 트랜잭션 관리 방법은 스프링 AOP를 이용해서 트랜잭션을 관리하는 방법이다.

@Transactional 어노테이션을 사용해서 DB 커넥션을 선언적으로 연결할 수 있고, 이로 인해 비즈니스 로직에서 트랜잭션 관련 로직을 제거하고, 순수한 비즈니스 로직만 남길 수 있다.

@Service
class Service() {

    @Transactional
    fun transactionTemplate(ticketId: Long, name: String) {
        bussinessLogic()
    }
}

프록시를 이용해서 트랜잭션을 관리하기 때문에 private 함수에 적용 불가능 , 내부 함수 호출 불가능 등을 주의해서 사용해야 한다.

마치며

이번 기회에 스프링 트랜잭션 특징, 동작 과정, 프로그래밍 방식의 트랜잭션 관리, 선언적 방식의 트랜잭션 관리에 대해 학습할 수 있었다.

이후에 트랜잭션 범위를 임의로 설정하고 싶지만 선언적 방식으로 관리가 힘들 때, 프로그래밍 방식으로 트랜잭션을 관리하는 것을 시도해볼 것 같다.

참고

스프링 DB 1편 - 데이터 접근 핵심 원리

Programmatic Transaction Management

Understanding the Spring Framework’s Declarative Transaction Implementation

팀원들이 DB에 저장된 수치를 슬랙으로 편하게 확인할 수 있도록, 서버 팀원과 함께 해당 기능을 구현했다. 나는 API 설계를 담당했고, 다음과 같은 요구 사항을 만족하는 API를 설계해야 했다.

요구사항

1. 시작 시간과 종료 시간 모두 주어진 경우: 해당 기간 동안의 수치를 확인할 수 있다.

2. 시작 일자만 주어진 경우: 시작 일자부터 지금까지의 수치를 확인할 수 있다.

3. 종료 일자만 주어진 경우: 오픈 날짜부터 종료 일자까지의 수치를 확인할 수 있다.

4. 시작 일자와 종료 일자를 모두 주어지지 않은 경우: 오픈 날짜부터 현재까지의 총 수치를 확인할 수 있다.

5. 날짜 형식: yyyy-MM-dd 형식이다.

설계

1. 쿼리 파라미터(from, to)로 시작 일자와 종료 일자를 선택적으로 받을 수 있도록 한다.
2. 현재 AuditingEntityListener를 이용해서 created_at칼럼과 updated_at칼럼이 자동으로 저장되고 있기 때문에, 입력 받은 from과 to 값을 created_at칼럼과 비교해서 데이터를 필터링한다.

• Case 1 (from = null, to = null): 오픈 날짜부터 현재까지의 수치를 반환.

• Case 2 (from = 2024-08-15, to = null): created_at이 from 이후인 데이터를 반환. (예: WHERE created_at > '2024-08-15')

• Case 3 (from = null, to = 2024-08-15): created_at이 to 이전인 데이터를 반환. (예: WHERE created_at < '2024-08-15')

• Case 4 (from = 2024-08-15, to = 2024-08-17): created_at이 from과 to 사이에 있는 데이터를 반환. (예: WHERE created_at BETWEEN '2024-08-15' AND '2024-08-17')

수치를 불러오는 쿼리마다 위 케이스를 적용하는 것은 비효율적이라고 생각했고, Querydsl을 이용해서 해당 로직을 공통으로 사용할 수 있도록 만들기로 결정했다.

구현

들어가기 전에 현재 모든 엔티티는 추상 클래스인 아래 AuditingTimeEntity를 상속을 받고 있다.

@Getter
@MappedSuperclass
@EntityListeners(AuditingEntityListener.class)
public abstract class AuditingTimeEntity {
    @CreatedDate
    private LocalDateTime createdAt;

    @LastModifiedDate
    private LocalDateTime updatedAt;
}

// 상속받아 사용 중이다.
public class Meeting extends AuditingTimeEntity
public class User extends AuditingTimeEntity 

이제 구현으로 넘어가면, 해당 로직은 두 조건을 만족해야 한다.

1. from과 to 값에 따라 동적으로 조건절을 적용해야 한다.

BooleanExpression , BooleanBuilder 이 두 기술은 Querydsl에서 동적으로 조건절을 적용할 수 있는 기술이다.

BooleanBuilder는 Builder 객체를 생성해야 하고, 조건절을 builder로 감싸 가독성을 저해한다고 판단하여 BooleanExpression을 선택했다.

BooleanBuilder	BooleanExpression

2. Meeting, User 등 서로 다른 테이블에 모두 적용할 수 있어야 한다.

서로 다른 테이블에 적용하기 위해서, 모든 테이블은 AuditingTimeEntity를 상속하고 있기 때문에 처음에는 auditingTimeEntity를 이용해서 로직을 만들었다.

하지만 아래와 같은 에러 코드를 띄우며 실패했다.

org.hibernate.query.SemanticException: Could not interpret path expression 'auditingTimeEntity.createdAt'

원인은 auditingTimeEntity는 부모 추상 클래스이기 때문에 매핑된 테이블이 없기 때문에 Hibernate에서 경로를 해석할 수 없었다. 따라서 createdAt을 사용하기 위해서는 meeting.createdAt, user.createdAt과 같이 사용해야 한다.

따라서 auditingTimeEntity를 사용하지 않고, 서로 다른 엔티티 객체에 어떻게 동일한 조건절을 반영할 수 있을까 고민했다.

Querydsl에서 사용하는 Q객체를 자세히 관찰한 결과, Q객체의 변수로 Path 프로퍼티가 사용되는 것을 확인할 수 있었고, createdAt칼럼은 DateTimePath<LocalDateTime>으로 정의되고 있었다.

따라서 createdAt 칼럼을 매개변수로 받아서 적용하기로 결정했고, 수정된 로직은 아래와 같다.

private BooleanExpression generateDateFilter(
        final DateTimePath<LocalDateTime> createdAt,
        final LocalDateTime from,
        final LocalDateTime to
) {
    if (from != null && to != null) {
        return createdAt.between(from, to);
    }
    if (from != null) {
        return createdAt.after(from);
    }
    if (to != null) {
        return createdAt.before(to);
    }
    return null; // null을 반환할 시, 자동으로 조건절에서 제거된다.
}

// 아래와 같이 사용한다.
.where(generateDateFilter(meeting.createdAt, from,to))
.where(generateDateFilter(user.createdAt, from,to))

조건절 검증

BooleanExpression을 처음 사용해보기 때문에 테스트 코드를 통해 검증하고 싶었다.

하지만 위에서 말했듯이 auditingTimeEntity 는 부모 추상 클래스이기 때문에 매핑된 테이블이 없으므로 별도의 객체를 생성할 수 없다. 그로 인해 EntityManager.persist를 통해서 테스트에 사용할 데이터를 생성할 수 없었다.

그래서 INSERT문을 통해서 데이터를 생성하기로 결정했고, 다음과 같이 테스트 데이터를 생성했다.

@Transactional
@SpringBootTest
class MetricsRepositoryGenerateDateFilterTest {
        private static final String INSERT_QUERY_TEMPLATE = "INSERT INTO meeting (title, password, duration, place_type, additional_info, created_at) VALUES ('title', '1234', 'HALF','ONLINE', '', ?)";

        @Autowired
        private EntityManager em;

        @DisplayName(
                """
                    1. 2022년 5월 20일 데이터
                    2. 2023년 1월 1일 데이터
                    3. 2023년 6월 15일 데이터
                    4. 2023년 12월 31일 데이터
                """
        )
        @BeforeEach
        public void setUp() {
            em.createNativeQuery(INSERT_QUERY_TEMPLATE)
                    .setParameter(1, "2022-05-20T10:00:00")
                    .executeUpdate();

            em.createNativeQuery(INSERT_QUERY_TEMPLATE)
                    .setParameter(1, "2023-01-01T12:00:00")
                    .executeUpdate();

            em.createNativeQuery(INSERT_QUERY_TEMPLATE)
                    .setParameter(1, "2023-06-15T15:30:00")
                    .executeUpdate();

            em.createNativeQuery(INSERT_QUERY_TEMPLATE)
                    .setParameter(1, "2023-12-31T23:59:59")
                    .executeUpdate();
        }
}

• EntityManager.createNativeQuery를 통해서 데이터를 생성한다.

• 해당 테스트의 주요 관심사는 created_at이기 때문에 나머지 데이터는 템플릿화하고, BeforeEach에서 created_at 칼럼을 설정한다.

• 데이터베이스에 INSERT 쿼리를 실행하기 위해 DB 커넥션이 필요하므로, @Transactional을 선언한다.

마무리

이러한 과정으로 요구사항을 만족하는 API를 구현할 수 있었다.

BooleanExpression을 사용해 본 점과 createNativeQuery를 사용해서 쿼리문을 통해 테스트 데이터를 생성하는 방식에 대해 학습할 수 있었다.

100개 한정인 영화 티켓을 300명이 동시에 요청한 상황이다. 테스트 코드는 아래와 같다.

@Test
fun `티켓 발급을 동시에 300번 요청했을 때, 100개의 티켓만 생성되어야 한다`() {
    val numberOfThread = 300
    val executorService = Executors.newFixedThreadPool(numberOfThread)
    val latch = CountDownLatch(numberOfThread)

    repeat(numberOfThread) { idx ->
        executorService.submit {
            try {
                ticketService.issueTicket(1L, "USER $idx")
            } finally {
                latch.countDown();
            }
        }
    }

    latch.await()

    val result = ticketService.countTicket(1L)

    assertThat(result).isEqualTo(100)
}

동시성 고려하지 않은 코드

@Service
class TicketService(
    private val ticketRepository: TicketRepository,
    private val ticketUserRepository: TicketUserRepository
) {

    @Transactional
    fun issueTicket(ticketId: Long, name: String) {
        val ticket = ticketRepository.findById(ticketId).orElseThrow { IllegalStateException("Ticket not found") }
        val ticketCount = countTicket(ticketId)
        if (ticketCount >= 100) {
            throw IllegalStateException("티켓 소진")
        }
        val ticketUser = TicketUser(ticket = ticket, name = name)
        ticketUserRepository.save(ticketUser)
    }

    fun countTicket(ticketId: Long): Int {
        return ticketUserRepository.countByTicketId(ticketId)
    }
}

위 서비스 코드는 동시성을 고려하지 않았기 때문에 100개의 티켓보다 더 많이 발급된 것을 볼 수 있다.

간단하게 이를 해결하기 위해서 synchronized를 적용하는 것이 있다.

동시성 문제 해결을 위한 synchronized 적용

synchronized는 특정 스레드가 작업을 수행하는 동안 다른 스레드는 작업을 대기하도록 만들어 주는 블로킹 동기화 방식이다. 메서드 또는 특정 영역에 락을 걸 수 있고, 인스턴스를 기준으로 락을 건다는 것을 유의해야 한다.

class A {
    // 메서드에 synchronized를 붙이는 방법
    @Synchronized
    fun print1() { ... }

    // 특정 영역에 synchronized를 붙이는 방법
    fun print2() {
        ...
        synchronized(this) {
            ...
        }
        ...
    }
}

// 인스턴스가 다를 시, 2개의 함수가 동시에 실행된다.
val a1 = A()
val a2 = A()

thread { a1.print1() }
thread { a2.print1() }

코틀린에서는 @Synchronized 를 통해서 자바의 synchronized를 메서드에 적용할 수 있다.

하지만 synchronized를 적용한다 해도 정확하게 100개의 티켓이 발급되지 않는 것을 볼 수 있다. 이는 @Transactional 과 synchronized를 동시에 사용할 때 발생하는 문제이기 때문이다.

@Transactional 은 AOP를 통해 서비스 로직이 끝난 후 커밋이나 롤백을 진행한다.

하지만 synchronized는 서비스 로직이 끝나는 즉시 락을 해제하기 때문에, 트랜잭션이 커밋되기 전에 락이 풀려서 여전히 동시성 문제가 발생한다.

프로그래밍 방식으로 트랜잭션 관리하기

현재는 1. 트랜잭션 시작 → 2. 잠금 획득 → 3. 서비스 로직 시작 → 4. 서비스 로직 종료 → 5. 잠금 반납 → 6. 트랜잭션 종료 순으로 진행되고 있어 문제가 발생했다.

이를 1. 잠금 획득 → 2. 트랜잭션 시작 → 3. 서비스 로직 시작 → 4. 서비스 로직 종료 → 5. 트랜잭션 종료 → 6. 잠금 반납 순으로 변경함으로써 문제를 해결하려 한다.

하지만, @Transactional 은 AOP로 동작하기 때문에 잠금 획득 후에 트랜잭션을 실행할 수 없다.

따라서 @Transactional을 사용한 선언적 트랜잭션 관리 방식에서, transactionTemplate을 사용한 프로그래밍 방식의 트랜잭션 관리 방식으로 아래와 같이 변경했다. (스프링 트랜잭션 관리에 대한 자세한 내용은 해당 아티클을 확인하자)

@Service
class TicketService(
    private val ticketRepository: TicketRepository,
    private val ticketUserRepository: TicketUserRepository
) {

    @Synchronized
    fun issueTicketWithSynchronized(ticketId: Long, name: String) {
        transactionTemplate.execute {
            val ticket = ticketRepository.findById(ticketId).orElseThrow { IllegalStateException("Ticket not found") }
            val ticketCount = countTicket(ticketId)
            if (ticketCount >= 100) {
                throw IllegalStateException("티켓 소진")
            }

            val ticketUser = TicketUser(ticket = ticket, name = name)
            ticketUserRepository.save(ticketUser)
        }
    }

    fun countTicket(ticketId: Long): Int {
        return ticketUserRepository.countByTicketId(ticketId)
    }
}

이를 통해서 테스트를 성공한 것을 볼 수 있다.

하지만 synchronized 를 통해 동시성을 제거하는 것은 아래와 같은 아쉬운 점이 있다.

1. 모든 티켓 요청에 대해 잠금을 건다.

   예를 들어, A 티켓은 100개로 한정되어 있고, B 티켓은 모두 발급할 수 있는 상황에서 [A,A,A,B,B,A,A,A]와 같이 요청이 들어온다면, 들어온 순서대로 요청을 처리한다.

   이로 인해 동시성 관리가 필요 없는 B 티켓은 먼저 온 요청이 끝날 때까지 대기해야 하는 상황이 발생한다.

위 상황을 아래 테스트 코드처럼 재현할 수 있다.

    @Test
    fun `3종류의 티켓 요청을 동시에 900번 요청했을 때, 각 티켓 당 100개의 티켓만 생성되어야 한다`() {
        val numberOfThread = 900
        val executorService = Executors.newFixedThreadPool(numberOfThread)
        val latch = CountDownLatch(numberOfThread)

        repeat(numberOfThread) { idx ->
            executorService.submit {
                try {
                    ticketService.issueTicketWithSynchronized((idx % 3) + 1L, "USER $idx")
                } finally {
                    latch.countDown();
                }
            }
        }

        latch.await()

        val result1 = ticketService.countTicket(1L)
        val result2 = ticketService.countTicket(2L)
        val result3 = ticketService.countTicket(3L)

        assertThat(result1).isEqualTo(100)
        assertThat(result2).isEqualTo(100)
        assertThat(result3).isEqualTo(100)
    }

테스트는 성공하지만, 900개의 요청을 순차적으로 수행하기 때문에 3~4초 정도 걸리는 것을 볼 수 있다.

2. 분산 환경에서 관리할 수 없다.

   분산 환경에서는 요청이 각각 다른 서버로 전달될 수 있기 때문에, 단일 서버 내에서의 동기화만으로는 문제가 해결되지 않는다.

따라서, synchronized만으로는 동시성 문제를 완벽히 해결할 수 없다. 분산 환경에서도 문제를 해결하기 위해서는 분산락 방식을 사용하는 등의 방법이 필요하다.

다음 글에서 데이터베이스의 네임드 락을 이용한 분산락 방식에 대해 알아보겠다.

참고 [Java] 혼동되는 synchronized 동기화 정리 [10분 테코톡] 알렉스, 열음의 멀티스레드와 동기화 In Java

이벤트 구조는 3가지 구성 요소(이벤트 , 발행자 , 구독자)가 필요하고, 이는 스프링에서 쉽게 구현할 수 있다.

1. 이벤트

발행자와 구독자 간에 사용할 이벤트를 아래와 같이 정의할 수 있다. 여기서는 이메일 전송을 위한 이벤트를 예시로 들겠다.

public record SendEmailEvent(String email) {
}

2. 구독자

@EventListener 어노테이션을 통해 이벤트 발생 시, 실행할 작업을 쉽게 정의할 수 있다.

아래 예시에서는 메일 전송 이벤트가 발행될 때, listen 함수 내부에 정의된 작업들이 실행된다.

@Component
@RequiredArgsConstructor
public class SendEmailEventListener {
    private final EmailService emailService;

    @EventListener
    public void listen(SendEmailEvent sendEmailEvent) {
        emailService.send(sendEmailEvent.email());
    }
}

3. 발행자

스프링의 ApplicationEventPublisher를 사용하여 이벤트를 발행할 수 있다. 아래와 같이 ApplicationEventPublisher를 주입을 받고, publishEvent 함수를 통해서 이벤트를 발행한다.

이벤트 발행 시, 해당 이벤트를 구독 중인 리스너들이 반응하여 정의된 작업을 수행한다.

@Service
@RequiredArgsConstructor
public class CreateUserService {
    private final UserRepository userRepository;
    private final ApplicationEventPublisher publisher;

    @Transactional
    public void save() {
        userRepository.save(new User("KWY"));
        publisher.publishEvent(new SendEmailEvent("KWY@gmail.com"));
    }
}

이렇게 발행자와 구독자 간의 이벤트 기반 구조를 스프링에서 쉽게 구현할 수 있다.

또한, 스프링 이벤트는 멀티캐스팅 방식이기 때문에, 같은 이벤트를 구독하는 리스너가 여러 개 있을 수 있으며, 이벤트가 발행되면 해당 이벤트를 구독 중인 모든 리스너가 동작한다.

아쉬운 점 : 동기

도메인 간 의존성을 덜 수 있다는 장점 외에도 API 응답 시간 단축을 기대했다. 하지만 스프링 이벤트 방식은 동기 방식으로 동작하기 때문에 리스너들의 작업이 끝난 후에 API가 종료된다. 이를 해결하기 위해, @Async 어노테이션을 통해 이벤트를 비동기 방식으로 동작하도록 만들 수 있다.

Application에는 @EnableAsync 어노테이션을, 리스너 함수에는 @Async 어노테이션을 정의하면 쉽게 이벤트를 비동기 방식으로 동작하게 만들 수 있다.

@EnableAsync
public class ServerApplication {
    public static void main(String[] args) { SpringApplication.run(ServerApplication.class, args); }
}

@Component
@RequiredArgsConstructor
public class SendEmailEventListener {
    private final EmailService emailService;

    @EventListener
    @Async
    public void listen(SendEmailEvent sendEmailEvent) {
        emailService.send(sendEmailEvent.email());
    }
}

하지만 @Async 어노테이션은 호출될 때마다 스레드를 생성하기 때문에 이벤트가 발생할 때마다 스레드를 생성, 제거하는 비용이 든다. 이를 해결하기 위해 스레드 풀을 미리 정의하는 방식을 사용한다.

@Configuration
@EnableAsync
public class SpringAsyncConfig {
    /*
        평소 스레드 최대 3개의 스레드 동작
        3개 이상의 스레드 요청이 동시에 들어올 때, 크기가 50인 큐에서 대기
        대기 큐가 가득 찼을 때, 최대 10개의 스레드 생성
    */
    @Bean
    public Executor threadPoolTaskExecutor() {
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        taskExecutor.setCorePoolSize(3);
        taskExecutor.setQueueCapacity(50);
        taskExecutor.setMaxPoolSize(10);
        return taskExecutor;
    }
}

위와 같이 Application이 아닌 Config에 @EnableAsync 어노테이션을 붙이고, 스레드 풀을 정의한다. 이후 이벤트가 발생할 때마다 매번 스레드를 생성하지 않고, Config에 정의했던 스레드 풀에서 스레드를 가져다 쓰고 반납한다.

이벤트와 트랜잭션

각 이벤트마다 트랜잭션에 포함이 될 수도, 안 될 수도 있기 때문에, 이를 위해 스프링은 @TransactionEventListener를 제공하고 있다.

@EventListener 대신 사용하면 되며, 옵션은 다음과 같다.

• BEFOR_COMMIT : : 이벤트 발행 주체가 커밋이 되기 전에 이벤트 실행

• AFTER_COMPLETION : : 이벤트 발행 주체가 커밋 또는 롤백이 된 후에 이벤트 실행

• AFTER_COMMIT (default) : 이벤트 발행 주체가 커밋이 되면 이벤트 실행

• AFTER_ROLLBACK : 이벤트 발행 주체가 롤백이 되면 이벤트 실행

주의할 점 1

AFTER_ 관련 옵션은 이벤트가 실행될 때, 트랜잭션이 이미 커밋 또는 롤백되어 종료된 상태이다. 따라서 @TransactionEventListener 내부에서 데이터를 저장 또는 수정하는 작업은 DB에 반영되지 않는다.

이를 해결하기 위해서 이벤트 함수에 @Transactional(propagation = Propagation.REQUIRES_NEW) 어노테이션을 정의해서 이벤트가 발생할 때마다 트랜잭션을 새롭게 만들어 줘야 한다.

주의할 점 2

@Transactional(propagation = Propagation.REQUIRES_NEW) 어노테이션이 붙은 이벤트가 N번 발생한다면, N개의 트랜잭션이 생성되어 DB 커넥션을 점유한다.

해당 이벤트가 동기 방식이라면 스레드가 종료(API가 종료)될 때까지 DB 커넥션을 계속 점유한다. 이를 해결하기 위해서 비동기 방식으로 이벤트를 실행함으로써, 각 이벤트가 종료될 때마다 DB 커넥션이 반납되도록 할 수 있다.

참고

[Spring] 스프링에서 이벤트의 발행과 구독 방법과 주의사항, 이벤트 사용의 장/단점과 사용 예시

스프링 이벤트 적용기

원티드 BE 챌린지 강의를 듣던 중 ‘가능한 @Transactional(readOnly=true) 어노테이션을 사용하는 것이 좋아요.’라는 말을 들었다. 이에 대해 의문을 갖고 왜? 사용하는 것이 좋은지 학습하고자 한다.

스프링 트랜잭션이 무엇이고, 어떻게 동작하는지 궁금하다면 해당 아티클에 정리한 내용을 참고하자.

@Transacitonal(readOnly = true)

@Transactional 어노테이션에는 여러 속성을 제공하는데 그중 하나가 readOnly 속성이다.

트랜잭션이 읽기 전용인지 읽기/쓰기 전용인지 선언하며, 기본 값은 false(읽기/쓰기)로 설정되어 있다. 읽기 전용 트랜잭션은 성능 최적화가 가능하다.

어떻게 성능 최적화가 가능할까?

• 수행 시간 및 메모리를 절약할 수 있다.

  JPA는 변경 감지 기능을 지원해서 트랜잭션을 데이터베이스로 전달하기 전에 스냅샷과 결과 값을 비교하고, 변경 사항이 있으면 추가 쿼리를 실행한다.

  하지만 읽기 전용으로 실행한다면 변경 감지를 할 필요가 없어서 스냅샷을 생성하지 않는다. 이로 인해서 변경 감지를 안 하기 때문에 수행 시간이 줄어들고, 스냅샷을 생성하지 않기 때문에 메모리를 절약할 수 있어 성능 최적화가 가능하다.

• 데이터베이스 Replication 부하 분산

  DB의 부하를 분산하기 위해서 사용하는 Replication은 Master-Slave 구조로 있다. Master에는 쓰기 연산을, Slave에서는 읽기 연산을 수행하면서 DB의 부하를 분산한다.

  readOnly를 설정한 트랜잭션은 Slave DB에 읽기 요청을 보낸다.

  만약 readOnly를 설정하지 않으면 모든 트랜잭션이 Master로 향하게 되어 부하 분산이 제대로 이뤄지지 않는다. readOnly를 설정함으로써 Replication의 취지에 맞게 DB가 동작하게끔 만들 수 있다.

그렇다면 무조건 readOnly를 사용하는 것이 좋을까?

그렇지 않다.

읽기 전용으로 트랜잭션을 실행해서 성능을 최적화할 수 있지만, 결국 DB의 커넥션을 점유한다.

이로 인해 읽기 연산이 오래 걸린다면 커넥션을 오래 소유하게 되어 DB 커넥션 고갈로 이어질 수 있기 때문에 이 점을 주의해야 한다.

@Transactional(readOnly=true)와 붙이지 않는 것의 차이는 뭘까?

@Service
public class TestService {

    @Transactional
    public void method() throws InterruptedException {
        Thread.sleep(1000L);
    }

    @Transactional(readOnly = true)
    public void method2() throws InterruptedException {
        Thread.sleep(1000L);
    }

    public void method3() throws InterruptedException {
        Thread.sleep(1000L);
    }
}

함수가 위와 같고, DB 커넥션 풀의 수를 1로 설정한 뒤 3개의 메서드를 동시에 실행한 결과, method()과 method2()는 2초 뒤에 종료가 되고, method3()은 바로 실행이 된다.

그렇다. @Transactional 어노테이션을 붙이지 않는다면 DB 커넥션을 점유하지 않는다.

이것을 보고, @Transactional 어노테이션을 붙이지 않으면 readOnly의 이점을 가져감과 동시에 DB 커넥션을 점유하지 않기 때문에 @Transactional 어노테이션을 붙이지 않는 것이 더 좋지 않은가?’란 생각을 했다.

@Transactional 유무의 차이는 OSIV 설정을 false로 했을 때, 알 수 있다.

💡 OSIV란?

영속성 컨텍스트를 View Layer까지 유지하는 속성으로 클라이언트 요청 시점(Filter / Interceptor - Controller)부터 끝날 때까지 영속성 컨텍스트를 생성되어 유지된다. 이를 통해 View Layer에서도 Entity의 Lazy Loading이 가능해진다.

기본적으로 OSIV는 true로 설정되어 있어 @Transactional 유무의 차이를 알 수 없다.

하지만 OSIV가 false일 땐, 트랜잭션 범위를 벗어나는 순간 Entity는 영속성 컨텍스트의 관리를 받지 않는 준영속 상태가 되어서 Lazy Loading이 불가능해지는 문제점이 있다.

OSIV : true
// @Transactional(readOnly = true)
public void method2() throws InterruptedException {
    excuteLazyLoading(); <- 성공
}

OSIV : false
// @Transactional(readOnly = true)
public void method2() throws InterruptedException {
    excuteLazyLoading(); <- 에러
}

마치며

이번 학습을 통해서 @Transactional 어노테이션을 붙이는 나만의 기준이 생겼다.

• @Transactional 을 안 붙이는 상황

  데이터베이스를 Replication을 사용하지 않을 때

  조회 쿼리이면서 조회 이후에 Lazy Loading과 같이 영속성 컨텍스트의 도움이 없어도 되는 경우

• @Transactional(readOnly=true)

  데이터베이스를 Replication을 사용할 때

  조회 쿼리이면서 조회 이후에 영속성 컨텍스트의 도움이 필요한 경우

• @Transactional

  엔티티를 변경하는 로직일 때

이러한 기준을 가지면서 비즈니스 로직을 분석해 @Transactionl 을 사용하려 한다.

또한, DB 커넥션을 짧게 가져가기 위해 OSIV를 false로 바꿀 것이다.

참고

Is @Transactional(readOnly=true) a silver bullet?

[Spring] @Transactional(readOnly = true)는 왜 붙여야 할까??

사이드 프로젝트로 웹앱 서비스를 운영하고 있다. 해당 서비스에 웹 푸시 알림을 도입한다면 서비스 리텐션을 높일 수 있지 않을까 생각했고, 팀원들을 설득하기 전에 웹 푸시 알림에 대해 공부하려고 한다.

푸시 알림

알림과 푸시는 다른 의미를 가진다. 알림은 클라이언트가 사용자에게 새로운 정보를 표시하거나, 업데이트 된 내용 등을 알릴 수 있고, 푸시는 서버에서 클라이언트 방향으로 메시지를 전송하는 것이다.

푸시 기능이 가능하려면 하나의 스레드가 서버의 응답을 계속 기다려야 하는데 웹에서는 서비스 워커가 그 역할을 한다.

웹 푸시는 보안을 위해서 https 와 localhost 에서만 동작한다.

FCM을 통해서 푸시 기능을 쉽게 구현할 수 있다.

최종적으로 아래와 같은 흐름으로 푸시 알림이 진행된다.

클라이언트

vapid 키가 있어야 한다.

클라이언트는 서비스 보안을 위해서 vapid 키가 꼭 필요하고, FCM에서 이를 지원해준다.

(프로젝트 설정 → 클라우드 메시징 → 하단 웹 구성 → 웹 푸시 인증서 클릭 후 Generate key pair)

위 경로를 통해서 다음과 같이 vapid 키를 발급 받을 수 있다.

FCM으로 subscribe 요청을 보내고, FCM 토큰 값을 받는다.

Firebase로부터 토큰 값을 받기 전에 사용자에게 알림 권한 요청이 선행되어야 한다.

사용자가 알림 권한이 없는 상태에서 subscribe 요청을 보내면 에러가 발생하니 꼭 알림 권한이 있는 상태에서 subscribe 요청을 보내자.

알림 허용을 받은 후, vapid키와 함께 subscribe 요청을 FCM으로 보내서 토큰 값을 받을 수 있다. 문서의 스니펫에 나오는 getToken 메서드를 통해서 토큰을 얻을 수 있고, 이후 해당 토큰 값을 서버로 전송하면 된다.

토큰 값을 받는 자세한 과정은 Firebase 문서를 참고하면 된다.

서비스 워커

서비스 워커는 브라우저와 네트워크 사이의 가상 프록시이다. 웹 페이지와 독립된 스레드에서 실행되며 오프라인 기능, 알림 처리, 독립된 스레드에서의 복잡한 계산 등 많은 것을 할 수 있다.

FCM의 서비스 워커는 /public/firebase-messaging-sw.js 파일에서 서비스 워커 기능을 구현해야 한다. 이때, 경로와 파일 명을 준수해야 한다.

서버

클라이언트로부터 받은 토큰 값을 저장한다.

서버는 클라이언트로부터 토큰 값을 받고, 언제든 해당 클라이언트에게 푸시 이벤트를 보낼 수 있도록 이를 DB에 잘 저장해야 한다.

메시지 전송하기

서버에서의 FCM 구현 방식으로 Firebase Admin SDK 사용, FCM HTTP v1 API 사용, 기존 HTTP 프로토콜 사용 총 3가지가 있다.

이중 HTTP v1 API 방식이 가장 최신의 프로토콜로서 보다 안전한 승인과 유연한 크로스 플랫폼 메시징 기능을 제공한다. 또한, 이제 새로운 기능은 HTTP v1 API 방식에만 추가되므로 해당 방식을 사용하는 것을 권장한다.

Firebase Admin SDK 사용 방식과 HTTP v1 API 방식, 총 2개의 메시지를 전송하는 방식을 실습 해볼 것이다.

1. Firebase Admin SDK로 메시지 전송하기

애플리케이션을 시작할 때 Firebase를 초기화 한다.

메시지를 전송할 때, Builder 패턴을 이용해서 간단하게 푸시 메시지를 전송할 수 있다.

fun main(args: Array<String>) {
    val resource = ClassPathResource("firebase/fcm.json")
    val serviceAccount = FileInputStream(resource.file)

    val options = FirebaseOptions.builder()
        .setCredentials(GoogleCredentials.fromStream(serviceAccount))
        .build()

    FirebaseApp.initializeApp(options)

    runApplication<ServerApplication>(*args)
}

fun push(targetToken: String) {
  val notification: Notification = Notification.builder()
      .setTitle("타이틀")
      .setBody("바디")
      .build()

  val message: Message = Message.builder()
      .setToken(targetToken)
      .setNotification(notification)
      .build()

  FirebaseMessaging.getInstance().send(message)
}

서버에 Admin SDK를 추가하는 자세한 내용은 해당 Firebase 문서를 참고하면 된다.

Admin SDK를 통해 메시지를 전송하는 자세한 내용은 해당 Firebase 문서를 참고하면 된다.

2. FCM HTTP v1 API로 메시지 전송하기

애플리케이션을 시작할 때 따로 Firebase를 초기화 하지 않아도 된다.

JSON 형식으로 된 메시지를 통해서 사용자에게 메시지를 보낸다.

메시지를 전송할 target을 나타내는 token 또는 topic 을 무조건 설정해줘야 한다.

• 메시지 형식 예시

    data class FCMSendDto(
        @JsonProperty("validate_only")
        val validateOnly: Boolean,
        val message: MessageDto
    )

    data class MessageDto(
        val token: String,
        val notification: NotificationDto
    )

    data class NotificationDto(
        val title: String,
        val body: String
    )

JSON에 형식의 메시지는 해당 Firebase 문서를 참고하면 된다.

• FCM AccessToken 가져오기

    private fun getAccessToken(): String {
        val scopes = listOf("https://www.googleapis.com/auth/cloud-platform")
        val resource = ClassPathResource("firebase/fcm.json")
        val serviceAccount = resource.inputStream

        val googleCredentials: GoogleCredentials = GoogleCredentials
            .fromStream(serviceAccount)
            .createScoped(scopes)
        googleCredentials.refreshIfExpired()
        return googleCredentials.accessToken.tokenValue
    }

공식 문서에서 보면 refreshAccessToken() 함수를 사용하지만 accessToken을 null로 가져오기 때문에 꼭 refreshIfExpired() 함수를 사용하자

• 메시지 전송하기

    val restTemplate = RestTemplate()
    val headers = HttpHeaders().apply {
        contentType = MediaType.APPLICATION_JSON
        setBearerAuth(getAccessToken())
    };
    val objectMapper = ObjectMapper()
    val entity = HttpEntity(objectMapper.writeValueAsString(body), headers)

    val response = restTemplate.exchange(
        "https://fcm.googleapis.com/v1/projects/{projectId}/messages:send",
        HttpMethod.POST,
        entity,
        String::class.java
    )

    println(response)

HTTP 요청 주소에 있는 {parent=projects/*}는 본인의 프로젝트 id를 뜻한다. 프로젝트 id는 프로젝트 설정란에서 확인할 수 있다.

메시지 전송에 관한 자세한 내용은 해당 Firebase 문서를 참고하면 된다.

마치며

웹 푸시에 관해 학습할 때, 우선 How to make PWAs re-engageable using Notifications and Push 문서를 한번 읽는 것을 추천한다. 프로세스를 빠르게 파악할 수 있고, FCM을 도입할 때 ‘이게 이 내용이구나’를 알 수 있어서 FCM 흐름을 이해하는데 도움을 준다.

HTTP v1 API 방식을 바로 도입하는 것이 베스트긴 하지만 시간이 부족하고, 아직 FCM이 잘 이해가 되지 않는다면 우선 Admin SDK를 사용하는 방식으로 구현하는 것을 추천한다. 먼저 쉽고 간단하게 Admin SDK를 사용하는 방식으로 기능을 구현한 후에 HTTP v1 API 방식으로 마이그레이션을 하면 좋을 것 같다.

참고

How to make PWAs re-engageable using Notifications and Push

Making PWAs work offline with Service workers

Push Payload 쿡북

중간 중간 있는 Firebase 문서

특정 트랜잭션이 다른 트랜잭션에서 변경하거나 조회하는 데이터를 볼 수 있게 허용할지 말지를 결정하는 것으로 크게 READ UNCOMMITTED, READ COMMITTED, REPEATABLE READ, SERIALIZABLE 4가지 격리 수준이 있다.

READ UNCOMMITTED

특정 트랜잭션에서 처리한 작업이 완료되지 않았는데도 다른 트랜잭션에서 볼 수 있는 격리 수준으로 이를 더티 리드(DIRTY READ)라 한다.

READ COMMITTED

가장 많이 선택되는 격리 수준으로 커밋이 완료된 데이터만 다른 트랜잭션에서 조회할 수 있는 격리 수준이기 때문에 더티 리드가 발생하지 않는다.

하지만 NON-REPEATABLE READ 문제가 발생할 수 있다.

NON-REPEATABLE READ가 발생하는 과정

A 트랜잭션에서 특정 레코드를 조회한 결과와 B 트랜잭션이 해당 데이터를 변경하고 커밋한 이후, 다시 A 트랜잭션에서 동일한 레코드를 조회했을 때 결과가 달라질 수 있다.

이를 NON-REPEATABLE READ라 하며, 이는 항상 같은 결과를 가져와야 한다는 “REPEATABLE READ” 정합성에 어긋난다.

REPEATABLE READ

MySQL의 InnoDB에서 기본으로 사용되는 격리 수준으로 트랜잭션에서는 항상 동일한 데이터를 조회할 수 있는 격리 수준으로 “NON-REPEATABLE READ”가 발생하지 않는다.

하지만 다른 트랜잭션에서 수행한 작업으로 인해 레코드가 보였다 안 보였다 하는 현상인 팬텀 리드(PHANTOM READ)가 발생할 수 있다.

NON-REPEATABLE READ , 팬텀 리드 차이

• Non-repeatable reads are when your transaction reads committed UPDATES from another transaction. The same row now has different values than it did when your transaction began.*
• Phantom reads are similar but when reading from committed INSERTS and/or DELETES from another transaction. There are new rows or rows that have disappeared since you began the transaction.*

NON-REPEATABLE READ는 특정 트랜잭션에서 레코드를 변경(UPDATE)한 후 커밋한 결과로 변경된 레코드를 불러오는 현상을 말한다.

팬텀 리드는 특정 트랜잭션에서 레코드를 INSERT 또는 DELETE한 후 커밋한 결과로 레코드가 추가로 조회되거나 이전보다 덜 조회되는 현상을 말한다.

팬텀 리드가 발생하는 과정

A 트랜잭션에서 특정 레코드를 조회한 결과와 B 트랜잭션이 레코드를 추가 또는 제거하고 커밋한 이후 다시 A 트랜잭션에서 SELECT … FOR UPDATE 쿼리로 동일한 데이터를 조회했을 때 팬텀 리드가 발생한다.

SELECT … FOR UPDATE 쿼리는 언두 영역의 변경 전 데이터를 가져오는 것이 아니라 현재 레코드의 값을 가져오기 때문에 팬텀 리드가 발생한다.

하지만 InnoDB에서는 REPEATABLE READ 격리 수준에서도 팬텀 리드 현상을 배제할 수 있다.

어떻게 REPEATABLE READ 격리 수준에서 팬텀 리드 현상을 배제할 수 있을까?

InnoDB에서는 갭 락과 넥스트 락 덕분에 REPEATABLE READ 격리 수준에서도 일반적으로 팬텀 리드가 발생하지 않는다.

따라서 위 예제에서 2번 쿼리가 실행되면 3번 쿼리는 session1 트랜잭션이 끝날 때까지 기다리게 되어 팬텀 리드가 발생하는 것을 막는다.

하지만 다음과 같은 경우에는 팬텀 리드가 발생한다.

2번 과정에서 락을 걸지 않기 때문에 5번 과정에서 팬텀 리드 현상이 발견될 것이다.

레코드 락

레코드 자체를 잠그는 것으로 다른 DBMS와 마찬가지로 공유 락(Shared Lock), 배타 락(Exclusive Lock)이 있다. 차이점으로는 이전 글에서 말했듯 MySQL에서의 레코드 락은 인덱스의 레코드를 잠근다.

갭 락

레코드 자체가 아니라 레코드와 바로 인접한 레코드 사이의 간격에 있는 빈 공간을 잠그는 것을 의미하며, 레코드와 레코드 사이의 간격에 새로운 레코드가 생성되는 것을 제어한다.

예를 들어 id가 13, 17인 레코드만 있는 테이블에 아래의 쿼리를 실행할 때,

SELECT c1 FROM t WHERE c1 BETWEEN 10 AND 20 FOR UPDATE;

10과 20 사이 중 레코드가 없는 빈 공간(GAP)을 잠근다. 즉, 위 트랜잭션이 끝날 때까지 10에서 12사이, 14에서 16사이, 18에서 20사이의 빈 공간을 잠근다.

넥스트 락

레코드 락과 갭 락을 합쳐 놓은 형태의 잠금을 넥스트 키 락이라고 한다. 다음 그림을 통해 자세히 알아보자

만일 위 예제에서 SELECT * WHERE pk > 99 AND pk < 102 FOR UPDATE를 실행시켰었다면, 발견 직전의 인덱스 레코드 97과 직후 발견되는 인덱스 레코드인 103 사이인 97 < pk < 103의 범위에 대해서 Lock이 걸린다.

SERIALIZABLE

특정 트랜잭션에서 읽고 쓰는 레코드를 다른 트랜잭션에서는 절대 접근할 수 없는 격리 수준이다.

레코드를 읽기 작업을 할 때에도 공유 잠금(읽기 잠금)을 획득해야 하며, 동시에 다른 트랜잭션을 레코드를 변경하지 못하게 만든다. 이로 인해 팬텀 리드가 발생하지 않지만 동시 처리 성능은 다른 격리 수준보다 떨어진다.

마무리

각 격리 수준 별 발생할 수 있는 현상을 표로 보여주며 마무리하겠다.

https://stackoverflow.com/questions/11043712/non-repeatable-read-vs-phantom-read

https://mangkyu.tistory.com/299

https://jaeseongdev.github.io/development/2021/06/16/Lock의-종류-(Shared-Lock,-Exclusive-Lock,-Record-Lock,-Gap-Lock,-Next-key-Lock)/

멀티 모듈을 공부하기 위해 호기롭게 의존성 분리를 계획해서 모듈을 생성했지만 각 모듈 별 Gradle 관리를 어떻게 해야할지 도통 감이 오지 않았습니다.

프로젝트를 생성할 때, intellij에서 늘 자동으로 gradle을 설정해주고, dependency를 관리하는 역할로만 알았기 때문에 이번 기회에 Gradle에 대해 공부하기로 결정했습니다.

Gradle이란?

Gradle은 빌드 관리 도구 중 하나로 Groovy 기반의 빌드 스크립트인 build.gradle을 통해 의존성을 관리합니다.

빌드 관리 도구란?

우선 빌드란 소스 코드를 컴파일, 테스트, 정적분석 등을 실행하여 실행 가능한 애플리케이션으로 만들어주는 과정을 말하고, 이러한 빌드 과정을 자동화하여 관리하는 기능을 빌드 관리 도구라고 말합니다.

Gradle 프로젝트 구조

Gradle 프로젝트는 다음과 같은 구조를 가집니다.

project
├── gradle                              
│   ├── libs.versions.toml              
│   └── wrapper
│       ├── gradle-wrapper.jar
│       └── gradle-wrapper.properties
├── gradlew                             
├── gradlew.bat                         
├── settings.gradle(.kts)               
├── subproject-a
│   ├── build.gradle(.kts)              
│   └── src                             
└── subproject-b
    ├── build.gradle(.kts)              
    └── src                             

• gradle
  • wrapper 파일을 저장하는 디렉토리
• libs.versions.toml
  • 의존성 관리를 위한 Gradle 버전 카탈로그
• gradlew , gradlew.bat
  • Gradle Wrapper
• setting gradle
  • 루트 프로젝트 이름과 하위 프로젝트를 정의하는 Gradle 설정 파일
• sub project의 build gradle
  • 하위 프로젝트의 Gradle 빌드 스크립트

Gradle Wrapper

Gradle Wrapper는 Gradle을 설치하지 않아도 빌드할 수 있도록 만들어주는 파일로 Gradle 버전을 호출하는 스크립트로 Gradle 빌드를 실행하는데 권장하는 방법입니다.

Gradle Wrapper를 사용하면 서로 다른 개발 환경에서 특정 Gradle 버전으로 표준화할 수 있기 때문에 다른 개발 환경의 사용자들 모두 동일한 Gradle 버전을 제공합니다.

Gradle Wrapper로는 gradlew , gradlew.bat 있습니다.

어떻게 Gradle 버전을 표준화할 수 있을까?

Gradle Build 과정을 보면 알 수 있다.

1. Gradlew Wrapper 스크립트(gradlew,gradlew.bat)로 빌드를 실행할 때, 특정 Gradle 버전을 다운 받는다.
2. 로컬에 해당 Gradle 버전을 저장한다.
3. 해당 Gradle 버전을 사용하여 빌드를 실행한다.

이 때, 1번 과정에서 특정 Gradle 버전을 다운 받는 것을 볼 수 있는데, gradle/wrapper/gradle-wrapper.properties 파일을 보면 특정 Gradle 버전을 명시하며 Gradle 버전을 통일하는 것을 알 수 있다.

왜 Gradle을 사용할까?

빌드 관리 도구로 Maven도 있지만 왜 Gradle 사용을 추천하는 이유로는 대표적으로 성능 측면이 있습니다.

실제 Gradle Docs에서도 Gradle이 Maven보다 최소 2배 더 빠르고, 대규모 빌드의 경우 100배 더 빠르다고 나와있습니다.

Gradle이 Maven 보다 빌드 속도가 더 빠른 이유는 다음과 같습니다.

1. 점진적 빌드

빌드 실행 중 소스 코드가 변경되지 않았다면 최신 상태로 간주하고 빌드는 실행되지 않는다.

2. 빌드 캐시

두 개의 모듈이 동시에 실행되고 있을 때, 이 두 모듈이 하나의 모듈을 공유하고 있다면, 해당 모듈은 한번만 빌드되어 재사용이 가능하다.

3. 데몬 프로세스를 지원

데몬 프로세스는 사용자의 요청에 응답하기 위해 백그라운드에 실행되는 프로세스이다.

Gradle의 데몬 프로세스는 빌드 결과물을 메모리 상에 보관하여 다음 빌드 속도를 감소한다.

배운점

항상 Gradle Wrapper가 자동으로 생성되기 때문에 Gradle이 설치되어 있지 않아도 빌드가 되는 것이 당연하다고 생각했습니다. 이번 기회에 이러한 기능이 Gradle Wrapper를 통해 이루어진다는 것을 알게 되었습니다.

또한, Gradle이 Maven보다 빌드 속도가 빠르다는 것도 막연하게만 알고 있었지만, 이번에 어떤 방식으로 속도를 빠르게 만드는지 구체적으로 알 수 있었습니다.

참고

https://gradle.org/maven-vs-gradle/

https://www.youtube.com/watch?v=ntOH2bWLWQs

https://oya150.tistory.com/entry/gradle-wrapper-는-왜-필요-하나

특정 IP 주소를 명시하는 영숫자로 구성된 텍스트 문자열로, 사용자가 쉽게 웹 페이지를 접근할 수 있도록 만들어준다.

DNS (Domain Name System)

도메인 이름과 IP 주소를 매핑되어 있는 서버로 사용자가 IP 주소가 몰라도 도메인을 통해 웹 페이지를 접근할 수 있도록 만들어준다.

DNS 동작 과정

1. Local DNS에 도메인 주소에 해당하는 IP 주소가 있는지 확인하고, 만약 있다면 IP 주소를 바로 반환한다.
2. 없다면 Root DNS에 쿼리를 호출하여 해당 도메인의 TLD 네임 서버 주소를 받는다.
   • TLD 네임 서버란 .com, .org 등과 같은 도메인의 최상단에 있는 첫 번째 중단점을 나타낸다.
3. TLD 네임 서버에 쿼리를 호출하여 해당 권한 있는 네임 서버의 주소를 받는다.
   • 권한 있는 네임 서버란 IP 주소를 확인하기 위한 마지막 단계로 도메인에 매핑되어 IP 주소를 반환한다.
4. 권한 있는 네임 서버에 쿼리를 호출하여 해당 도메인의 IP 주소를 받는다.
   • TTL과 함께 반환 받으며, 해당 TTL 동안 도메인에 대한 IP 주소를 캐시한다.
5. 해당 IP주소로 이동하여 웹 페이지를 띄운다.

DNS 레코드

DNS 레코드는 DNS 서버가 어떤 식으로 처리할지 지시하는 내용을 담고 있다. DNS 레코드의 종류는 대표적으로 아래가 있다.

• A 레코드, AAAA 레코드
  • A레코드 : IPv4 주소 매핑
  • AAAA레코드 : IPv6 주소 매핑
• CNAME 레코드
  • 별칭 도메인을 정식 도메인으로 연결한다. 즉, 하위 도메인을 도메인 A 또는 AAAA 레코드에 연결하는 데 사용한다.
  • www.example.com , api.example.com 에 대해 두 개의 A 레코드를 만드는 대신 api.example.com을 CNAME 레코드에 연결한 다음 example.com의 A 레코드에 연결할 수 있다. 이를 통해 루트 도메인에 대한 IP 주소가 변경되면 A 레코드만 업데이트하면 되고 CNAME은 자동을 업데이트 된다.
• A 레코드 VS CNAME 레코드
  • A 레코드는 한번의 요청으로 IP 주소를 바로 알 수 있어 속도가 빠르다는 장점이 있는 반면, IP 주소가 자주 바뀌는 환경에서는 바뀔 때 마다 IP 주소를 변경해야 하기 때문에 번거롭다는 단점이 있다.
  • CNAME 레코드는 여러 서브 도메인들은 메인 도메인과 매핑된 CNAME 레코드로 저장하면 IP 주소가 변경될 때, 하나의 레코드만 수정하면 되는 장점이 있는 반면, A Record를 찾기 위해 여러번 요청을 해야 하기 때문에 A 레코드에 비하면 속도가 느린 단점이 있다.
• NS 레코드
  • 어떤 DNS가 해당 도메인의 IP 주소를 찾기 위해 권한 있는 네임 서버로 가는 방법을 알려준다.
  • 이를 통해 어떤 도메인에 대한 처리를 다른 도메인 네임 서버에게 위임할 수 있어서 A 업체에서 구매한 도메인을 B 업체에서 관리할 수 있다.
• MX 레코드
  • 이메일을 도메인 메일 서버로 전송하기 위해 도메인의 SMTP 이메일 서버를 지정한다.

참고

https://www.ibm.com/kr-ko/topics/dns https://www.ibm.com/kr-ko/topics/dns-records https://www.cloudflare.com/ko-kr/learning/dns/what-is-dns/ https://www.cloudflare.com/ko-kr/learning/dns/dns-server-types/ https://www.cloudflare.com/ko-kr/learning/dns/glossary/what-is-a-domain-name/ https://www.cloudflare.com/ko-kr/learning/dns/top-level-domain/ https://inpa.tistory.com/entry/WEB-🌐-DNS-개념-동작-완벽-이해-★-알기-쉽게-정리