[Redis] Redis 분산락(Distributed Lock)을 이용한 재고관리 구현 - 1

프로젝트 개요

이번 프로젝트에서는 분산 시스템 환경에서 발생하는 다양한 문제 상황중, 서로 다른 요청이 하나의 데이터에 몰리게 되어 발생하는 레이스 컨디션(Race Condition) 문제를 분산 락을 이용해 해결하는 미니 프로젝트를 진행하고 이를 정리했습니다.

 

코드들을 모두 게시글에 작성할 수 없어, 일부 설정은 다른 게시글에 정리된 내용을 링크하였습니다.

프로젝트 전체 코드는 아래 github을 통해 확인할 수 있습니다.
https://github.com/kwj2435/spring-redis-stock-manager

---

기술 스택

Spring Boot 3.4.1

Spring Data JPA
Java 17
Redis 7.0

Kafka

H2 Database

 

구현 요구 사항

주요 기능

1. 재고 등록: 제품과 초기 재고량을 Redis에 저장.
2. 재고 감소: 주문 요청 시 재고를 감소시키고, 남은 재고가 없으면 처리 실패
3. 재고 조회: 현재 재고 상태를 조회
4. 분산락 적용: 동시 재고 감소 요청 시 일관성 유지

비기능 요구 사항

1. 동시성 제어: 분산락을 사용하여 레이스 컨디션 방지
2. 에러 처리: 재고 부족, 락 획득 실패 등의 상황 처리

---

1. 프로젝트 세팅

1.1 Redis 클러스터 구성

이번 프로젝트에서는 DB의 재고 데이터를 Redis에 캐싱 후, Redis에서 다중 요청으로 인해 발생하는 레이스 컨디션 문제를 분산락으로 해결할 예정입니다. 이를 위해 우선 Redis 클러스터를 구성해줍니다.

Redis 클러스터는 3개의 Master 노드, 3개의 Slave 노드로 총 6개 노드로 구성하였습니다.

개발 환경 구성의 편의성을 위해서 Docker-Compose를 이용하여 Redis 노드를 생성하고, 이후 구성된 노드를 기반으로 redis-cli을 통해 최종적으로 Cluster 구성을 마무리 하였습니다.

Redis 클러스터 구성하기

[Redis] Redis 클러스터(Redis Cluster) 구성하기

 

Redis 클러스터와 Spring 연동

Spring 프로젝트에 spring-boot-starter-data-redis 의존성을 추가후 구성된 redis 클러스터와 연동을 위해 yml 파일을 설정하고, RedisTemplate을 이용하기 위한 RedisConfig 파일을 만듭니다.

spring: 
  data:
    redis:
      cluster:
        nodes:
          - 127.0.0.1:6379
          - 127.0.0.1:6380
          - 127.0.0.1:6381
          - 127.0.0.1:6382
          - 127.0.0.1:6383
          - 127.0.0.1:6384

@Configuration
public class RedisConfig {

  @Bean
  public RedisTemplate<String, Object> redisTemplate(RedisClusterConfiguration clusterConfiguration) {
    RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
    redisTemplate.setConnectionFactory(redisConnectionFactory(clusterConfiguration));
    redisTemplate.setKeySerializer(new StringRedisSerializer());
    redisTemplate.setValueSerializer(new StringRedisSerializer());
    return redisTemplate;
  }

  @Bean
  public LettuceConnectionFactory redisConnectionFactory(RedisClusterConfiguration clusterConfiguration) {
    return new LettuceConnectionFactory(clusterConfiguration);
  }

  @Bean
  public RedisClusterConfiguration clusterConfiguration() {
    RedisClusterConfiguration clusterConfig = new RedisClusterConfiguration();
    clusterConfig.setClusterNodes(Arrays.asList(
        new RedisNode("127.0.0.1", 7001),
        new RedisNode("127.0.0.1", 7002),
        new RedisNode("127.0.0.1", 7003),
        new RedisNode("127.0.0.1", 7004),
        new RedisNode("127.0.0.1", 7005),
        new RedisNode("127.0.0.1", 7006)
    ));
    return clusterConfig;
  }
}

 

Spring Boot Redis에서는 기본적으로 Lettuce를 기본 클라이언트로 사용합니다. Lettuce와 비교대상으로 언급되는 Jedis에 비하여 비동기 지원, Thread Safe, 높은 동시성과 성능에 최적화되었다는 이점으로 범용적으로 사용됩니다.
이를 위해 LettuceConnectionFactory Bean을 생성하고, Redis 클러스터 구성 정보를 RedisClusterConfiguration Bean으로 등록하여 Spring에서 인식할 수 있도록 구성합니다.

1.2 Kafka 연동

Redis에 캐싱된 데이터는 최종적으로 DB 데이터와 동기화 처리가 되어야 합니다.
이를 위해 Redis에서 실시간 변경되는 데이터를 이벤트 기반으로 DB에 반영하기 위해 Kafka를 사용합니다.
Kafka 또한 Redis와 마찬가지로 가용성과 성능을 고려한다면 클러스터 구조로 구성을 고민해야합니다.
다만, 이번 프로젝트에서는 Kafka 사용이 핵심이 아니기에 단일 노드로 간단히 구성했습니다.

먼저, Docker를 이용해서 Kafka를 구성합니다. Kafka는 Zookeeper와 함께 실행되므로 Docker Compose를 사용하여 Kafka 환경을 구성하는 것이 일반적입니다.

Redis 클러스터를 구성하며 작성된 Docker Compose 파일에 Kafka와 Zookeeper 서비스를 추가합니다.

services:
  zookeeper:
    image: confluentinc/cp-zookeeper:latest
    container_name: zookeeper
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
    ports:
      - "2181:2181"

  kafka:
    image: confluentinc/cp-kafka:latest
    container_name: kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
    depends_on:
      - zookeeper

 

다음, 도커 컴포즈로 구성된 Kafka를 Spring과 연동합니다.

build.gradle에 kafka 관련 의존성을 추가하고, 설정 파일을 추가 작성합니다.

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

spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: my-group
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

 

1.3 Spring - H2 DB 연동 및 DDL 작성

마지막으로, 개발 환경에서 가볍게 사용하기 적합한 H2 DB를 프로젝트에 연동하고, JPA사용을 위한 관련 설정을 진행해줍니다.

Spring - H2 DB 연동하기

[Spring] 스프링 부트(Spring Boot) - H2 DB 연동하여 사용하기(with.JPA)

 

프로젝트에서 사용할 DDL을 작성하고, resource 하위에 schema.sql와 data.sql 파일로 만들어 애플리케이션 시작시 자동으로 테이블 구성 및 더미데이터가 생성되도록 구성합니다.

-- schema.sql
DROP TABLE IF EXISTS INVENTORY;
CREATE TABLE INVENTORY (
   INVENTORY_ID BIGINT AUTO_INCREMENT PRIMARY KEY,       -- 상품 ID
   PRODUCT_NAME VARCHAR(255) NOT NULL,       -- 상품 이름
   STOCK_QUANTITY INT NOT NULL DEFAULT 0,    -- 재고 수량
   LAST_UPDATED TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP -- 마지막 업데이트 시간
);

 

-- data.sql
INSERT INTO INVENTORY(PRODUCT_NAME, STOCK_QUANTITY, LAST_UPDATED) VALUES('바나나', 999, CURRENT_TIMESTAMP);
INSERT INTO INVENTORY(PRODUCT_NAME, STOCK_QUANTITY, LAST_UPDATED) VALUES('사과', 999, CURRENT_TIMESTAMP);
INSERT INTO INVENTORY(PRODUCT_NAME, STOCK_QUANTITY, LAST_UPDATED) VALUES('수박', 999, CURRENT_TIMESTAMP);

2. Redis - Database 동기화 처리

서버 애플리케이션이 시작된 직후에는 아직 Redis에 DB 재고 정보가 동기화 되지 않은 상태입니다.

또한 동기화 된 이후라 할지라도, Redis에 변경된 재고 정보는 실시간으로 DB에 반영하여 다른 서비스 혹은 기타 기능에서 재고 정보를 사용할 수 있도록 동기화 처리해줘야 합니다.

1. 서버 애플리케이션 시작 시점 Redis 재고 정보 동기화

Redis Warmup은 애플리케이션이 시작된 직후 Redis가 비어있는 상태에서 발생하는 부하를 줄이기 위해 Redis Warmup이 필요합니다.
서버가 실행된 직후 DB 재고 정보를 Redis에 Warmup 하는 과정은 CommandLineRunner를 이용했습니다.

Bean이 생성되는 시점 호출되는 @PostConstructor 어노테이션과 비교하여, CommandLineRunner는 Bean이 모두 생성된 후 호출되며, 동기화에 실패했을 경우 retry 로직을 추가하는등 여러개의 Command를 순차적으로 호출할 수 있다는 장점이 있습니다.

 

우선, DB에서 조회된 엔티티를 Redis에 저장하기 적합한 형태로 담아둘 DTO를 작성합니다.

@Getter
public class Inventory {
  /** 상품 명 */
  private String productName;
  /** 상품 수량 */
  private int stockQuantity;

  private Inventory(String productName, int stockQuantity) {
    this.productName = productName;
    this.stockQuantity = stockQuantity;
  }

  public static Inventory from(InventoryEntity inventory) {
    return new Inventory(inventory.getProductName(), inventory.getStockQuantity());
  }
}

 

다음, CommandLineRunner를 구현한 InventorySyncRunner 클래스 내부에서 DB에서 조회된 엔티티를 DTO로 전환 및 Redis에 반영하도록 처리합니다.

@Component
@RequiredArgsConstructor
public class InventorySyncRunner implements CommandLineRunner {

  private final InventoryEntityRepository inventoryEntityRepository;
  private final RedisTemplate<String, Object> redisTemplate;

  @Override
  public void run(String... args) throws Exception {
    List<InventoryEntity> inventories = inventoryEntityRepository.findAll();
    for (InventoryEntity inventory : inventories) {
      Inventory inventoryDTO = Inventory.from(inventory);
      redisTemplate.opsForHash().put("inventory:details", inventory.getInventoryId(), inventoryDTO);
    }
  }
}

 

동기화 로직 완성후 애플리케이션을 실행시키면 다음과 같이 DB - Redis 동기화가 정상적으로 이루어진 것을 확인할 수 있습니다.

2. 실시간 Redis 재고 정보 DB 동기화

애플리케이션이 실행되는 시점과 별개로 서비스가 운영되면서 수시로 변경되는 Redis  데이터는 DB와 동기화를 시켜줘야합니다.
재고정보는 다양한 모듈과 서비스에서 사용되는 정보이기때문에 DB에 반영하여 사용될 수 처리되어야 하는 실시간성 데이터입니다.

이를 위해서 Redis 재고가 증감되는 시점에 Kafka로 이벤트를 발행하여 DB에 반영될 수 있도록 처리하였습니다.

 

Redis의 Keyspace Notification을 이용하여 특정 키가 변경된 시점에 이벤트를 발행하여 Spring에서 받아 처리할 수 있지만 Redis에 부하가 갈 수 있고, Redis에 의존적으로 변한다는 이유로 Spring 로직에서 재고가 증감되는 시점에 Kafka로 이벤트를 발행하도록 처리하였습니다.

 

실시간 동기화 구조

1. Redis 재고 감소 요청

2. 재고 감소 요청 성공시 Kafka Topic에 변경된 재고 정보 이벤트 발행

3. 토픽 이벤트를 Consumer가 확인후 DB에 반영

 

공통적으로 사용할 수 있는 간단한 Producer와, 발행된 재고 업데이트 이벤트를 처리할 Consumer 클래스

@Service
@RequiredArgsConstructor
public class InventoryProducer {

  private final KafkaTemplate<String, String> kafkaTemplate;

  public void sendMessage(String topic, String message) {
    kafkaTemplate.send(topic, message);
  }
}

@Service
@RequiredArgsConstructor
public class InventoryConsumer {

  private final InventoryEntityRepository inventoryEntityRepository;
  private final ObjectMapper mapper = new ObjectMapper();

  @Transactional
  @KafkaListener(topics = "stock-updates")
  public void consumeStockUpdateEvent(String message) throws JsonProcessingException {
    Inventory inventory = parseMessageToInventory(message);

    InventoryEntity inventoryEntity =
            inventoryEntityRepository.findById(inventory.getInventoryId())
                    .orElseThrow(() -> ApiException.from(ApiExceptionCode.ERR_500_10001));
    inventoryEntity.updateStockQuantity(inventory.getStockQuantity());

    // todo JsonProcessingException 발생시 DLQ로 전달 로직 추가 필요
  }

  private Inventory parseMessageToInventory(String message) throws JsonProcessingException {
    return mapper.readValue(message, Inventory.class);
  }
}

 

* 실시간 동기화를 위해 재고 변경이 이루어질때마다 이벤트를 통해 DB에 Update 요청을 보내게 되는 Write-Throgh 패턴으로 설계가 되었습니다.

Redis 캐싱의 이점을 살리고, DB 부하를 줄이기 위해서 준실시간 동기화 처리, DB Read/Write 분리등을 통해 DB 부하를 줄이도록 처리해야 합니다.

---

 

이번 글에서는 프로젝트 환경 설정과, 재고 관리 관련 API 개발까지의 과정을 정리해봤습니다.
다음 글에서는 Jedisson을 이용한 Redis 분산락 처리와 테스트를 진행하는 과정을 정리해보겠습니다.

다음글
[Redis] Redis 분산락(Distributed Lock)을 이용한 재고관리 구현 - 2