무슨 상황인가?​

새로운 프로젝트에 들어가기 앞서 잠깐 까먹었던 Jenkins를 다시 사용해보려고 공부하고 있던 중이었다. 기존에는 Maven 프로젝트를 사용했지만 Gradle 프로젝트를 배포해야 하기 때문에 새롭게 테스트를 하고 있었다.

그런데 이놈 생각보다 까다롭다.. 무려 64번째 시도 끝에 성공했지만 이 글을 보는 사람들은 한 번에 성공하길 바란다.

Jenkins 설정해 보자!​

1. 내가 사용한 테스트 프로젝트​

Jenkins 배포 테스트에 사용할 프로젝트를 만들어 주었다. 의존성도 없고 그냥 RestApi 하나 띄워둔 Gradle 프로젝트이다.

이 프로젝트는 Java 17 버전과 Gradle 8.5 버전을 사용한다.

2. Jenkins 설정​

Jenkins 대쉬보드에서 Jenkins 관리 -> Tools에 들어오면 Gradle을 설정할 수 있다. 내가 사용하는 프로젝트는 Gradle 8.5 버전을 사용하기 때문에 맞추어 설정해 주고 Save 해준다.

Freestyle 프로젝트를 하나 생성해서 설정을 진행하는데 위에 프로젝트 Github 주소를 넣어준다.

소스코드 관리 탭에서 Git을 선택하고 브랜치를 알맞게 넣어준다. 처음에 master로 설정되어 있지만 현재 Github에서 자동으로 생성되는 브렌치는 main이라 잘 보고 넣어줘야 한다.

이제 아까 위에서 대쉬보드 Tools에서 설정한 Gradle를 선택해주고 스크립트를 작성하고 저장 후 실행하면 된다.

비상, 비상! 문제 발생​

1. 뭐가 된다? 문제가 된다.​

대충 몇 번을 실행해도 이런 화면을 보게 될 것이다. 끔찍하다.

로그를 확인해 보면 이렇게 나온다. 검색 유입을 위해 굳이 코드를 직접 적어보자면 아래와 같다.

Directory '/var/jenkins_home/workspace/My-Second-Project' does not contain a Gradle build. A Gradle build should contain a 'settings.gradle' or 'settings.gradle.kts' file in its root directory. It may also contain a 'build.gradle' or 'build.gradle.kts' file.

failure: build failed with an exception.

build step 'invoke gradle script' marked build as failure

FAILURE: Build failed with an exception.

* What went wrong:
Directory '/var/jenkins_home/workspace/My-Second-Project' does not contain a Gradle build.

A Gradle build should contain a 'settings.gradle' or 'settings.gradle.kts' file in its root directory. It may also contain a 'build.gradle' or 'build.gradle.kts' file.

To create a new Gradle build in this directory run 'gradlew init'

For more information about the 'init' task, please refer to https://docs.gradle.org/8.5/userguide/build_init_plugin.html in the Gradle documentation.

For more details on creating a Gradle build, please refer to https://docs.gradle.org/8.5/userguide/tutorial_using_tasks.html in the Gradle documentation.

* Try:
> Run gradlew init to create a new Gradle build in this directory.
> Run with --stacktrace option to get the stack trace.
> Run with --info or --debug option to get more log output.
> Get more help at https://help.gradle.org.

BUILD FAILED in 1s
Build step 'Invoke Gradle script' changed build result to FAILURE
Build step 'Invoke Gradle script' marked build as failure
Finished: FAILURE

이런 에러들이 뜨는데 잘 보면 /var/jenkins_home/workspace/My-Second-Project에 build.gradle가 없어서 생기는 문제이다.

3. 상황을 파악하고 대처를 해야한다.​

Jenkins를 Docker로 설치해서 운영했는데 Jenkins 컨테이너에 접속해서 에러에서 안내하는 경로로 가보면 구조가 위와 같이 구성되어 있다.

먼저, Git을 clone 해오게 되면 보통 레포지토리명/.. 이렇게 구성되어 받아지는데 Jenkins에서는 workspace에 Item명을 Root로 삼아 프로젝트가 받아진다.

Spring 프로젝트를 먼저 생성하고 git init을 하고 remote한 게 아니라 git repository를 생성하고 clone 받아 프로젝트를 생성했기 때문에 폴더가 한 댑스 들어가 있는 게 문제가 되었다.

4. 간단한 해결책은?​

있다. 간단한 해결책은 물론 있다. 바로 프로젝트가 한 댑스 들어가 있는 것을 한 꺼풀 벗겨내어 다시 Git에 올려주는 것이다. 하지만 그건 근본적인 해결 방법은 아니다. 하나의 레포지토리에 프로젝트 폴더가 꼭 하나만 있는 게 아니기 때문에 원하는 폴더를 잡을 수 있어야 한다.

그래서 나는 build.gradle 경로를 잡아주어 해결하기로 했다.

해결 방법을 세웠으면 시도하라!​

1. 일단 뭐든 조져보자.​

먼저, 간단하게 생각해 볼 수 있는 디렉토리 이동이다. 메인 패키지에서 jenkins 폴더로 들어가면 해결되는 것이기 때문에 이렇게 시도했다. 사실 구글링을 하면서 떠올린 방법이다.

결과는 실패, 하지만 괜찮다. 다음 스텝 밟아보자.

절대 경로로 다시 잡아본다. 프로젝트에 있는 jenkins 폴더에 접근하는 게 내 목표이다. 진짜 간단해 보이는데 뭔가 방법을 몰라서 못 하는 것 같다. 검색해도 잘 나오지 않았기 때문에 내가 해볼 수 있는 것은 여러 방법으로 시도해 보는 것.

결과는 실패, 진짜 괜찮다. 다음 스텝 밟아보자.

위에서 대쉬보드 Tools에서 설정한 Gradle을 사용하지 않고 직접 경로를 잡아서 시도해 본다.

결과는 실패, 이것저것 시도하길 어느새 63번째 시도였다.

답은 있다. 내가 모르는 것일 뿐!​

1. 고생 끝 해결 완료​

Gradle 설정과 빌드 스크립트는 더 이상 건드리지 않는다. 여기서 고급 버튼을 눌러서 아래로 스크롤을 해보면

빌드 Root를 설정할 수 있는 곳이 있었다. 지금까지 어디에다 하고 있었던 것인지.. 이를 마치고 실행하면

드디어 성공했다. 생각보다 사례가 없었고.. 강의에서는 Maven을 사용했기 때문에 삽질을 한동안 해댔다.

2. 결론​

간단해도 한 번은 시도해 봐야 하는 이유이다. 방법을 모르면 간단하지 않을 수 있다. 해결책을 생각해 내어도 이를 실행할 방법을 모르면 말짱 꽝이다. 많은 오류를 만나고 많이 해결해 봐야 요령이 생긴다 했다. 64번의 시도가 아깝지 않았다.

인프런에서 실습으로 배우는 선착순 이벤트 시스템이라는 재밌는 강의가 있어서 진행해 보았다. 커리큘럼을 보니까 Redis와 Kafka를 사용해서 진행하는데 Redis는 그렇다 치고, Kafka는 따로 사용해 본 적이 없어서 간단하게 경험하기 위해 진행했다.

최상용 강사님의 강의인데 이분 강의는 슬쩍슬쩍 알음알음 맛보기 좋은 강의인 것 같다. 재고 시스템을 구현하면서 동시성을 해결하는 강의도 이전에 들었는데 다시 들어보면서 정리해 볼 예정이다.

이 글에서 전체 코드를 보여주진 않으니 궁금한 사람은 해당 강의를 들어보는 것을 추천한다. 인프런이 할인을 자주해서 대충 20퍼 할인할 때 구매하면 정가인 느낌이다.(정가가 아까운 강의는 아니다.)

Docker와 MySQL​

1. Docker 설치 및 셋팅​

brew install docker
brew link docker
docker version

맥북으로 실습을 진행했다. 패키지 관리자인 brew를 통해 docker를 설치하는 것 부터 시작한다.

docker pull mysql

실습에서 기본적으로 MySQL을 사용한다. MySQL의 Image를 가져온다. 이때 버전을 기록하지 않으면 자동으로 Latest 버전을 가져오게 된다.

docker run -d -p 3306:3306 -e MYSQL_ROOT_PASSWORD=1234 --name mysql mysql
docker ps
docker exec -it mysql

MySQL 컨테이너를 생성해 주고 실행시켜 준다.

2. MySQL 연동과 데이터베이스 셋팅​

mysql -u root -p

위에서 실행한 컨테이너에서 MySQL에 접속해 준다. 암호는 위에서 설정한 1234를 참고한다.

CREATE DATABASE coupon_example;
USE coupon_example;

coupon_example 데이터베이스를 생성하고 메인 데이터베이스로 설정해 준다.

spring:
  jpa:
    hibernate:
      ddl-auto: create
    show-sql: true
  datasource:
    driver-class-name: com.mysql.cj.jdbc.Driver
    url: jdbc:mysql://127.0.0.1:3306/coupon_example
    username: root
    password: 1234

Spring 프로젝트에 application.yml 파일에 위와 같이 접속 정보를 설정해 주면 기초 공사는 끝나게 된다.

요구사항​

선착순 100명에게 할인 쿠폰을 제공하는 이벤트 진행

1. 선착순 100명만 지급.
2. 101개 지급은 안 됨.
3. 순간적으로 몰리는 트래픽을 버텨야 한다.

초기 로직 구현 및 테스트​

1. 쿠폰 생성 로직​

public void apply(Long userId) {
    long count = couponRepository.count();

    if (count > 100) {
        return;
    }

    couponRepository.save(new Coupon(userId));
}

쿠폰의 개수를 확인하고 개수가 100개를 넘었을 때 그냥 리턴해 준다. 100개가 넘지 않았을 경우 쿠폰을 생성해서 저장하는 간단한 로직으로 테스트를 진행한다.

2. 테스트 코드 작성​

@Test
public void 한번만응모() {
    applyService.apply(1L);

    long count = couponRepository.count();

    assertThat(count).isEqualTo(1);
}

간단하게 진행하면 정상적으로 응모가 되는 것을 확인할 수 있다.

초기 로직의 문제 발굴​

초기 로직이 테스트 코드를 성공하면서 정상적으로 실행이 되었다고 오해할 수도 있다. 하지만 쿠폰 발급 이벤트는 동시에 여러번 실행이 될 것이다. 이런 상황을 가정해서 실행하면 문제가 나타난다.

1. 테스트 코드​

@Test
public void 여러명응모() throws InterruptedException {
    int threadCount = 1000;
    ExecutorService executorService = Executors.newFixedThreadPool(32);
    CountDownLatch latch = new CountDownLatch(threadCount);

    for (int i = 0; i < threadCount; i++) {
        long userId = i;
        executorService.submit(() -> {
           try {
               applyService.apply(userId);
           } finally {
               latch.countDown();
           };
        });
    }
    latch.await();

    long count = couponRepository.count();

    assertThat(count).isEqualTo(100);
}

여러 개의 스레드로 쿠폰을 발급해 보는 테스트 코드이다. 해당 코드를 실행해 보면

문제가 발생한 것을 확인할 수 있다.

2. 문제는 레이스 컨디션(Race condition)​

레이스 컨디션은 두 개 이상의 스레드가 공유 자원에 엑세스하고 활용하려고 할 때 생기는 문제로 위에서는 99개의 쿠폰이 있을 때 동시에 여러 스레드가 99개임을 인지하고 쿠폰을 발급해서 생기는 문제이다.

이런 상황을 이전에 회사에서 결제와 재고 처리로직 작성하면서 겪은 적이 있었는데 뮤텍스를 사용해서 처리했었다.

하지만 뮤텍스나 세마포어를 사용한 방법 말고도 여러 방법이 있는데 Redis를 사용해서 먼저 해결해 본다.

Redis로 레이스 컨디션 해결하기​

1. Redis setting​

docker pull redis
docker run --name myredis -d -p 6379:6379 redis

위에서 MySQL을 셋팅했을 때와 마찬가지로 Redis Image를 Latest로 가져오고 컨테이너를 생성하고 실행해 준다.

implementation 'org.springframework.boot:spring-boot-starter-data-redis'

Spring에서 의존성도 추가해 준다.

application.yml은 Redis를 생성할 때 따로 사용자 설정을 하지 않았기 때문에 위 의존성을 추가하면 알아서 디폴트로 연결해 준다.

따라서 따로 설정하지 않아도 Spring에서 local에서 실행하는 Redis에 접속할 수 있다.

2. Redis로 해결하려는 이유​

Lock을 사용하면 생성하는 로직부터 발급하는 로직까지 Lock을 걸어야 한다. 때문에 성능에 불이익이 있다. 또한 synchronized를 사용하게 되면 마찬가지로 성능 저하가 생길 수 있고 서버가 늘어났을 때 소용 없게 된다.

선착순 이벤트는 쿠폰 개수에 대한 정합성이 중요하다. Redis를 사용하는 이유는 Redis에서 지원하는 incr이라는 예약어가 있는데 이는 Key에 대한 Value를 1씩 증가키신다.

기본적으로 Redis는 Single-Thread 기반으로 동작하기 때문에 문제도 없고 incr 자체의 성능도 좋아 정합성을 지키는데 문제가 없기 때문에 사용하기 적당하다 할 수 있다.

3. incr 테스트​

docker exec -it 컨테이너ID redis-cli

Redis를 실행하고

incr coupon_count
incr coupon_count

incr 명령어를 실행하면 1씩 벨류가 증가하는 것을 확인할 수 있다.

4. Repository 코드​

public long increment() {
    return redisTemplate
            .opsForValue()
            .increment("couponCount");
}

Repository에 RedisTemplate을 사용해서 코드를 작성해주고 테스트 코드를 실행하면 통과하게 된다.

이는 Single-Thread인 Redis에서 숫자를 카운팅하기 때문에 각 스레드들이 가져가는 숫자가 겹치지 않게 되어 레이스 컨디션이 발생하지 않게 되는 것이다.

5. Redis를 사용한 레이스 컨디션 해결의 문제​

Redis 자체에 문제가 있기 보다는 Redis를 통해 발급 수량을 체크하고 결국 저장은 RDB에 된다. 때문에 쿠폰이 많을수록 RDB에 부하를 주게 된다. 이때, RDB가 쿠폰 전용이 아니라 공통적으로 사용하는 경우 더 큰 문제가 생길 수 있다.

그럼 어떻게 하나? Kafka를 사용해서 이 문제를 해결할 수 있다.

Kafka를 사용해서 Redis의 문제를 해결하자​

1. Kafka가 무엇인가?​

Kafka는 분산 이벤트 스트리밍 플랫폼으로 이벤트 스트리밍은 소스에서 목적지까지 이벤트를 실시간으로 스트리밍하는 것을 말한다.

Kafka는 Pub/Sub 모델 기반의 메시지 큐인데 이 Pub/Sub은 메시지를 생성하는 서비스를 해당 메시지를 처리하는 서비스에서 분리하는 확장 가능한 비동기 메시징 서비스이다.

간단하게 설명하면 게시자 및 구독자라는 의미를 통해 게시자가 이벤트를 생성하면 같은 곳을 바라보고 있는 구독자는 해당 이벤트가 생성 됨을 알 수 있게 되는 것이다.

여기에서는 게시자 및 구독자를 Producer와 Consumer로 표현하고 발생하는 이벤트를 바라보는 통로를 Topic이라고 보겠다. 따라서 Producer -> Topic <- Consumer로 구성되어 동작한다고 보면 된다. 이때 Topic은 Queue 같은 통로로 보면 되고 데이터 삽입은 Producer에 의해서, 데이터 소비는 Consumer에 의해서 진행된다.

그럼 이제 간단하게 Kafka를 셋팅하고 테스트 해 보자.

2. Kafka 셋팅​

version: '2'
services:
  zookeeper:
    image: wurstmeister/zookeeper
    container_name: zookeeper
    ports:
      - "2181:2181"
  kafka:
    image: wurstmeister/kafka:2.12-2.5.0
    container_name: kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_ADVERTISED_HOST_NAME: 127.0.0.1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock

Spring에 docker-compose.yml을 생성해서 위와 같이 작성해 준다.

docker-compose up -d
docker-compose down

명령어를 통해 작성한 docker-compose 파일을 실행하고 종료한다. 실행 후에는 docker ps 명령어를 통해 실행이 정상적으로 되었는지 확인한다.

3. Kafka 테스트​

docker exec -it kafka kafka-topics.sh --bootstrap-server localhost:9092 --create --topic testTopic

testTopic이라는 Topic을 생성한다. 말 그대로 테스트용이다. 생성이 완료되면 Created topic testTopic.라는 문구가 뜨게 된다.

docker exec -it kafka kafka-console-producer.sh --topic testTopic --broker-list 0.0.0.0:9092

이제 Topic을 삽입할 Producer를 실행해 준다.

docker exec -it kafka kafka-console-consumer.sh --topic testTopic --bootstrap-server 0.0.0.0:9092

마지막으로 Consumer를 생성해 줘야 하는데 Producer를 실행한 터미널을 놔두고 새로운 터미널을 열어주어 해당 명령어를 실행한다.

이후 Producer에서 hello를 입력하면 같은 Topic을 바라보던 Consumer가 데이터를 가져와 출력하게 된다. 이제 테스트는 끝났다 Kafka 구현에 들어간다.

Kafka Producer 구현하기​

1. 의존성 추가 및 config 작성​

implementation 'org.springframework.kafka:spring-kafka'

Spring에 Kafka 의존성을 추가해 준다.

@Configuration
public class KafkaProducerConfig {

    @Bean
    public ProducerFactory<String, Long> producerFactory() {
        Map<String, Object> config = new HashMap<>();

        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, LongSerializer.class);

        return new DefaultKafkaProducerFactory<>(config);
    }

    @Bean
    public KafkaTemplate<String, Long> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }

}

KafkaProducerConfig.java를 작성해 주고

@Component
public class CouponCreateProducer {

    private final KafkaTemplate<String, Long> kafkaTemplate;

    public CouponCreateProducer(KafkaTemplate<String, Long> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void create(Long userId) {
        kafkaTemplate.send("coupon_create", userId);
    }
}

Producer가 보낼 Topic을 지정해 준다. coupon_create로 설정해 주었다.

public void apply(Long userId) {

//        long count = couponRepository.count();
long count = couponCountRepository.increment();

        if (count > 100) {
            return;
        }

//        couponRepository.save(new Coupon(userId));
couponCreateProducer.create(userId);
}

Service 로직에서 RDB에 저장하지 않고 Producer를 거치도록 변경해준다. (추후에 변경할 예정이니 일단을 이렇게 한다.)

2. Topic 및 Consumer 생성​

docker exec -it kafka kafka-topics.sh --bootstrap-server localhost:9092 --create --topic coupon_create

터미널을 열어 coupon_create라는 Topic을 생성해 주고

docker exec -it kafka kafka-console-consumer.sh --topic coupon_create --bootstrap-server localhost:9092 --key-deserializer "org.apache.kafka.common.serialization.StringDeserializer" --value-deserializer "org.apache.kafka.common.serialization.LongDeserializer"

Consumer도 생성해 준다.

3. 테스트 실행​

테스트 코드는 실패했다. 왜냐하면 위에서 RDB에 저장되는 부분을 지우고 Producer를 거쳐 게시하고 있는 Topic에 데이터를 보내주었기 때문이다.

위에 터미널을 보면 무수한 값들이 표현되고 있다. 이는 Consumer를 실행한 터미널 창으로 Producer가 발생시킨 이벤트를 잘 받고 있는 것을 확인할 수 있다.

이렇게 Producer의 역할은 끝났다. 이제 Consumer를 구현해보자.

Kafka Consumer 구현하기​

1. 새로운 프로젝트​

이 프로젝트는 멀티모듈로 설계되었다. Consumer는 새로운 프로젝트로 구성한다. 새로운 프로젝트에 주입될 의존성은 아래와 같다.

implementation 'org.springframework.boot:spring-boot-starter-data-jpa'
implementation 'org.springframework.kafka:spring-kafka'
runtimeOnly 'com.mysql:mysql-connector-j'

application.yml은 이전에 생성했던 것과 동일하게 넣어준다.

2. Consumer Config​

@Configuration
public class KafkaConsumerConfig {

    @Bean
    public ConsumerFactory<String, Long> consumerFactory() {
        Map<String, Object> config = new HashMap<>();

        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        config.put(ConsumerConfig.GROUP_ID_CONFIG, "group_1");
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class);

        return new DefaultKafkaConsumerFactory<>(config);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Long> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, Long> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());

        return factory;
    }
}

마찬가지로 KafkaConsumerConfig.java를 작성한다.

@Component
public class CouponCreatedConsumer {

    @KafkaListener(topics = "coupon_create", groupId = "group_1")
    public void listener(Long userId) {
        System.out.println("쿠폰이 생성되었습니다. userId : " + userId);
    }
}

위에서 Producer(게시자)가 바라보는 토픽(coupon_create)을 구독하도록 설정해준다. 간단한 출력으로 그 결과를 확인해보자.

3. 테스트 실행​

Producer가 Topic에 쏘는 데이터를 Consumer가 잘 받아오는 것을 확인할 수 있다.

4. RDB에 저장시키기​

@Component
public class CouponCreatedConsumer {

    private final CouponRepository couponRepository;

    public CouponCreatedConsumer(CouponRepository couponRepository) {
        this.couponRepository = couponRepository;
    }

    @KafkaListener(topics = "coupon_create", groupId = "group_1")
    public void listener(Long userId) {
        couponRepository.save(new Coupon(userId));
    }
}

이제 출력이 아니라 데이터베이스에 저장하도록 한다.

5. 다시 테스트​

실패했다. 갑자기? 왜 실패했을까? 바로 데이터 처리 시간 차이에 따른 실패라고 볼 수 있다.

예를 들어 Producer가 데이터를 전송하고 완료하고 나서 테스트 케이스는 종료된다. 하지만 데이터는 Consumer가 받아서 처리한다. 그렇다. 테스트 케이스는 Consumer가 데이터를 처리하는 시간까지 기다리지 않았기 때문에 생긴 문제이다.

...
latch.await();

Thread.sleep(10000);
...

테스트 코드에 Thread sleep을 주어 충분히 Consumer가 처리할 수 있도록 기다려줘보자.

성공했다.

요구사항을 추가해보자.​

1. 추가되는 요구사항과 해결 방법​

쿠폰 발급을 1인당 1개로 제한한다.

위와 같은 요구사항을 추가한다고 가정하자. 이를 해결하는 방법 또한 여러가지가 있다. 데이터베이스에서 유니크를 활용하는 방법도 있고 범위 락을 사용하는 방법도 있다. 예를 들면 아래와 같다.

// lock start
// 쿠폰 발급 여부
// if 발급 되었다면 return
// lock end

하지만 이런 범위 락은 성능 저하 이슈가 발생할 수 있기 때문에 다른 방법을 떠올려야한다. 마침, Redis에서 set을 지원한다. 이를 통해 해결해보자.

3. Redis Set​

sadd test 1

Redis는 sadd 키워드를 통해 Key, Value를 지정할 수 있다.

4. 로직 작성​

@Repository
public class AppliedUserRepository {

    private final RedisTemplate<String, String> redisTemplate;

    public AppliedUserRepository(RedisTemplate<String, String> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    public Long add(Long userId) {
        return redisTemplate
                .opsForSet()
                .add("applied_user", userId.toString());
    }
}

Redis set을 사용하는 repository를 생성해 준다. opsForSet()을 사용하면 된다.

public void apply(Long userId) {
        Long apply = appliedUserRepository.add(userId);

        if (apply != 1) {
            return;
        }

//        long count = couponRepository.count();
        long count = couponCountRepository.increment();

        if (count > 100) {
            return;
        }

//        couponRepository.save(new Coupon(userId));
        couponCreateProducer.create(userId);
    }

Redis set에 userId를 저장하고 이미 쿠폰이 발급되어 있다면 그대로 return을 해 주어 1인 1개 쿠폰 발급을 진행한다.

5. 테스트 코드 작성 및 실행​

@Test
public void 한명당_한개의쿠폰만_발급() throws InterruptedException {
    int threadCount = 1000;
    ExecutorService executorService = Executors.*newFixedThreadPool*(32);
    CountDownLatch latch = new CountDownLatch(threadCount);

    for (int i = 0; i < threadCount; i++) {
        long userId = i;
        executorService.submit(() -> {
            try {
                applyService.apply(1L);
            } finally {
                latch.countDown();
            };
        });
    }
    latch.await();

    Thread.*sleep*(10000);

    long count = couponRepository.count();

    *assertThat*(count).isEqualTo(1);
}

생성되는 userId를 1로 고정했다. 새로 작성한 테스트 코드를 실행해서 확인하자.

Consumer 터미널에 데이터가 1이 출력되고 테스트 케이스 역시 통과된다. 참고로 이전에 실행 내역이 있기 때문에 Redis에 저장된 데이터를 날려주는 것을 잊지 말아야 한다.

발급 실패한 쿠폰은 어떻게 처리하지?​

그럼 발급에 실패한 쿠폰은 어떻게 처리할까? 따로 데이터베이스에 몰아서 저장시키고 배치 등을 사용해 다시 지급하는 방향으로 수정할 수 있다.

1. 발급 실패한 쿠폰 저장하기​

@Entity
public class FailedEvent {

    @Id @GeneratedValue(strategy = GenerationType.*IDENTITY*)
    private Long id;

    private Long userId;

    public FailedEvent() {
    }

    public FailedEvent(Long userId) {
        this.userId = userId;
    }
}

FailedEvent 테이블을 생성해서 발급 실패한 쿠폰을 정의하고

@KafkaListener(topics = "coupon_create", groupId = "group_1")
public void listener(Long userId) {
    try {
        couponRepository.save(new Coupon(userId));
    } catch (Exception e) {
        logger.error("failed to create coupon :: " + userId);
        failedEventRepository.save(new FailedEvent(userId));
    }
}

쿠폰 생성할 때 실패한 경우 실패 테이블에 데이터를 쌓아준다. 역시 이후에 배치 등을 돌려 해당 쿠폰들을 처리해주면 된다.

내가 겪언던 문제들​

이전에 회사에서 트랜잭션 격리 수준에 걸려서 고생한 적이 있다. 간단한 예를 들면 재고를 체크하는 부분에서 걸렸는데 아래와 같다.

1. 재고 조회
	2. FOR 구매반복 {
		3. 아이템 갯수 감소
		4. 아이템 네이밍 숫자 증가
		5. 아이템 생성
		6. 재고 감소
		7. 아이템 업데이트
		8. 재고 업데이트
	}

위에서 전체 재고를 가져온다. 그리고 구매를 반복하는데 구매하는 만큼 아이템 갯수를 줄여주고 아이템에 붙는 네이밍 숫자를 늘려준 다음 아이템을 생성한다. 마지막으로 아이템과 재고를 업데이트 해주고 새로운 반복을 진행한다.

새로운 반복을 진행한다.여기에서 문제가 발생한다. 가져온 재고가 만약 3개다. 첫 번째 반복문을 돌면서 재고가 1개 줄어들었을 때 과연 두 번째 반복문은 그것을 기억하고 있었나?

아니었다. 이게 뭔가 했는데 트랜잭션 격리 수준을 공부하면 알 수 있는 문제였다.

트랜잭션 격리 수준 이전에 알아야 할 것들​

트랜잭션​

그럼 트랜잭션 격리 수준은 무엇일까?를 궁금해 하기 전에 트랜잭션의 성질인 ACID를 알아야 한다. 아니 그 전에 트랜잭션은 무엇일까? 여기서부터 시작하는 게 옳겠다.

트랜잭션은 DBMS에서 수행되는 데이터베이스의 상태를 변화시키기 위해 수행되는 논리적인 작업 단위라고 말할 수 있다.

1. 재고 조회
	2. FOR 구매반복 {
		3. 아이템 갯수 감소
		4. 아이템 네이밍 숫자 증가
		5. 아이템 생성
		6. 재고 감소
		7. 아이템 업데이트
		8. 재고 업데이트
	}

이 예시를 다시 가져오자. 예를 들어서 내가 아이템 3개를 구매했다. 그럼 저 구매 반복문을 3번 돌아 쿼리가 모두 실행되어 데이터베이스에 적용됨이 마땅하다.

하지만 첫 번째 반복은 성공해서 아이템 하나가 정상적으로 생성되었으나 그 이후 두 번째 반복에서 실패해 두 번째 아이템부터 생성이 되지 않았다고 하자. 그럼 이 구매 실행은 어떻게 되어야 할까?

1. 전체가 실패해야 한다.
2. 첫 번째 아이템은 구매 성공, 이후는 실패

정답은 1번이다. 이런 트랜잭션의 성질은 이미 공부하기 좋게 정의되어 있다. ACID라 불리는 트랜잭션의 성질이다.

트랜잭션의 성질 ACID​

• Atomicity 원자성
• Consistency 일관성
• Isolation 격리, 고립성
• Durability 지속성

각각 성질은 무엇을 뜻할까?

원자성은 하나의 트랜잭션이 모두 성공하거나 모두 실패해야 하는 성질을 말한다. 즉, 위에서 예시로 실행한 구매 작업은 전체가 실패하거나 전체가 성공해야 한다는 말이다. 때문에 위에 예시의 정답은 1번이 되겠다.

일관성은 트랜잭션이 완료된 이후에도 데이터베이스의 상태가 일관되어야 한다는 말인데 데이터베이스의 무결성을 말한다. 구매를 하고 구매 장부를 기록한다 했을 때 구매에 대한 구매 장부 데이터가 빠지면 안 된다. 이런 일관성을 말하기도 하고 기본 키, 왜리 키 같은 제약 조건을 지키면서 트랜잭션을 수행하는 것을 의미하기도 한다.

고립성은 하나의 트랜잭션이 작업을 수행중일 때 다른 트랜잭션이 끼어들지 못 하도록 보장해 트랜잭션끼리의 영향이나 간섭이 없도록 하는 것이다.

지속성은 트랜잭션이 성공적으로 수행되고 난 뒤에 트랜잭션에 대한 로그가 남아 결과가 안정적으로 보존되어야 하고 영구적으로 반영되어야 한다는 말이다.

트랜잭션은 commit과 rollback을 통해 하나의 작업을 성공적으로 완료 혹은 작업 취소를 결정할 수 있으며 Auto commit 옵션이 적용이 Default인 경우가 있어 잘 체크하고 사용해야 한다.

트랜잭션 격리 수준​

트랜잭션 격리 수준의 정의들​

트랜잭션 격리 수준은 비슷하거나 똑같은 말이지만 이렇게 정의한다.

• 동시에 여러 트랜잭션이 처리될 때 트랜잭션끼리 얼마나 서로 고립되어 있는지를 나타내는 것.
• 특정 트랜잭션이 다른 트랜잭션에 변경한 데이터를 볼 수 있도록 허용할지 말지 결정하는 것.
• 여러 트랜잭션이 동시에 처리될 때 특정 트랜잭션이 다른 트랜잭션에서 변경하거나 조회하는 데이터를 볼 수 있게 허용할지 여부를 결정하는 것.
• 여러 트랜잭션이 다른 트랜잭션에서 변경하거나 조회하는 데이터를 볼 수 있도록 허용할지 말지를 결정하는 것.

요약하자면 트랜잭션이 서로 고립된 정도를 나타내며 다른 트랜잭션에서 변경한 데이터를 볼 것인지 결정하는 것.쯤 되겠다.

트랜잭션 격리 수준 구분​

그럼 그 고립된 정도를 나타낸다는 지표는 어떻게 구분되는지 보자면 아래와 같다.

트랜잭션 격리 수준은 위에서 아래로 높아진다고 보면 된다.

• Read uncommited (0)
• Read committed (1)
• Repeatable read (2)
• Serializable (3)

주의해야 하는 부분​

트랜잭션 격리 수준이 높아질 수록 격리성으로 인한 이슈는 적게 발생하겠지만 동시 처리 성능이 떨어지기 때문에 성능 요구사항과 데이터 무결성을 고려해 적절한 수준을 찾아야 한다.

간단히 격리 수준에 대해서 알아보자면 아래와 같다.

Read uncommitted​

Read uncommitted는 commit이나 rollback 상관 없이 다른 트랜잭션에서 조회할 수 있다. 이는 데이터 정합성에 문제가 있기 때문에 권장되지 않는다.

Read committed​

DBMS에서 가장 많이 사용되는 격리 수준으로 트랜잭션 변경 내용이 commit 되어야만 다른 트랜잭션에서 조회할 수 있다.

Repeatable read​

트랜잭션이 시작되기 전에 커밋된 내용에 대해서만 조회할 수 있는 격리 수준이다.

Serializable​

InnoDB에서 순수한 SELECT는 어떤 잠금도 없이 동작하는데 이때 공유 잠금을 설정하여 다른 트랜잭션에서 레코드 변경을 할 수 없게 만든다. 즉, SELECT가 사용되는 데이터들에 Shared Lock을 걸기 때문에 동시 처리 능력이 떨어지고 성능 저하가 발생된다.

트랜잭션 격리성으로 생길 수 있는 이슈들​

Dirty Read​

Read Uncommitted 격리 수준에서 다른 트랜잭션이 아직 커밋되지 않은 데이터를 읽을 수 있는 이슈이다.

Non-Repeatable Read​

Read committed 단계에서 다른 트랜잭션이 커밋한 데이터를 읽을 수 있는 것을 의미한다. 한 트랜잭션에서 쿼리로 2번 이상 조회했을 때 그 결과가 상이한 이슈를 말한다.

Phantom Read​

Repeatable read 격리 수준에서 한 트랜잭션이 동일한 쿼리를 두 번 실행할 때, 두 번째 실행에서는 다른 트랜잭션이 커밋한 데이터를 읽을 수 있는 이슈이다.

결국 해결한 방법​

결국 해결한 방법 조회 쿼리에 트랜잭션을 적용해 주었다.

func (r userRepositoryImpl) FindByNickname(ctx context.Context, req *ent.User) (*ent.User, error) {
	return r.User.Query().
		Where(user.NicknameEQ(req.Nickname)).
		Only(ctx)
}

Golnag 예시를 보면 지금 이 Repository 로직은 트랜잭션을 사용하지 않는 조회이다.

func (r userRepositoryImpl) FindByNickname(ctx context.Context, req *ent.User, tx *ent.Tx) (*ent.User, error) {
	return tx.User.Query().
		Where(user.NicknameEQ(req.Nickname)).
		Only(ctx)
}

이 Repository 로직은 트랜잭션을 사용하는 조회 로직이다. Spring에서 @Transactional이라는 어노테이션을 사용해서 트랜잭션을 관리하지만 Golang의 Ent Orm에서는 Tx라는 트랜잭션을 직접 생성하여 주입해 주어야 한다.

조회 부분에 추가되는 트랜잭션은 Repeatable read 격리 수준에서 차이가 나기 때문에 이를 적용해 주어 해결했다.