[Spring] Redis 동시성 문제 Redisson으로 해결!

공부를 하다가 Redis로 동시성 문제 라는 키워드를 알게되서 궁금해서 공부를 하게 되었다.

그러면 Redisson이 뭔지? 이에 대해 알아보자.

---

Redisson이란?

Redisson은 Redis를 위한 Java 클라이언트 라이브러리로, Redis의 기능을 보다 직관적으로 사용할 수 있도록 돕는다. 분산 객체, 분산 잠금, 비동기 API 등을 제공하며, 이를 통해 개발자는 복잡한 분산 시스템을 손쉽게 구현할 수 있다.

 

Redisson의 주요 특징

분산 객체

 

• RMap: 키-값 쌍을 저장할 수 있는 객체로, Java의 Map과 유사하게 동작한다.
• RSet: 중복을 허용하지 않는 집합 구조로, Java의 Set과 유사하다.
• RQueue: 큐 구조로, FIFO 방식으로 데이터를 저장하고 처리할 수 있다.

 

분산 잠금 (RLock)

 

• 잠금 획득: 여러 프로세스가 동시에 리소스에 접근할 때, 하나의 프로세스만이 해당 리소스를 사용할 수 있도록 한다.
• 잠금 해제: 사용이 끝난 후 잠금을 해제하여 다른 프로세스가 리소스에 접근할 수 있게 한다.
• 자동 해제: 설정한 시간 내에 잠금이 해제되지 않으면 자동으로 해제하는 기능을 제공한다.

비동기 API

Redisson은 비동기 처리를 지원하여, Redis 요청을 비동기적으로 처리할 수 있다.

 

Spring 통합

 

Redisson은 Spring과의 통합을 지원하여, Spring 애플리케이션에서 쉽게 Redis를 사용할 수 있다.

 

Redisson이 사용 이유

 

• 복잡성 감소: 일반적인 Redis 클라이언트 라이브러리는 사용하기 복잡할 수 있으며, Redisson은 Java의 일반적인 데이터 구조를 사용하여 이러한 복잡성을 줄인다.
• 동시성 문제: 여러 인스턴스가 동시에 같은 자원에 접근할 때 발생할 수 있는 문제를 해결하기 위해 분산 잠금을 제공한다.
• 비동기 처리의 필요성: 현대의 애플리케이션은 비동기 처리를 요구하며, Redisson은 이를 자연스럽게 지원한다.

 

---

Spring  AOP로 적용 예

AOP로 진행한 이유

1. 분산 락의 처리 로직이 경우에 따라 여기저기 필요할때 중복된 코드로 쓰이는 경우도 있지 않을까 싶어서 AOP로 분리하여 처리 로직을 분리하기 위함

2. AOP로 사용하는 경우 재사용성이 늘어남

3. 커스텀 어노테이션을 만들어서 진행시 lockname, waitTime, leaveTime 메서드별로 지정 가능

 

dependencies 추가

implementation 'org.redisson:redisson-spring-boot-starter:3.27.0'

 

application.properties 추가

spring.data.redis.host=${REDIS_HOST}
spring.data.redis.port=${REDIS_PORT}

 

RedisConfig.java 수정

@Configuration
public class RedisConfig {
    @Value("${spring.data.redis.host}")
    private String host;

    @Value("${spring.data.redis.port}")
    private int port;
    
    // redis pub/sub 통신을 위한 메세징 큐 설정
    @Bean
    public RedisMessageListenerContainer redisMessageListener(
            RedisConnectionFactory connectionFactory) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);

        return container;
    }

    // redis host , port 설정
    @Bean
    public RedisConnectionFactory redisConnectionFactory() {
        RedisStandaloneConfiguration redisConfiguration = new RedisStandaloneConfiguration();
        redisConfiguration.setHostName(host);
        redisConfiguration.setPort(port);
        return new LettuceConnectionFactory(redisConfiguration);
    }

    // redisTemplate 설정
    @Primary
    @Bean
    public RedisTemplate<String, Object> redisTemplate() {
        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(redisConnectionFactory());
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(new StringRedisSerializer());
        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashValueSerializer(new GenericJackson2JsonRedisSerializer());
        redisTemplate.setEnableTransactionSupport(true);
        redisTemplate.afterPropertiesSet();
        return redisTemplate;
    }

    // redisson 설정!!
    @Bean
    public RedissonClient redissonClient() {
        Config config = new Config();
        config.useSingleServer()
                .setAddress("redis://" + host + ":" + port)
                .setConnectionPoolSize(32);
        return Redisson.create(config);
    }
}

 

RedissonLock.java annotation 추가

@Target({ElementType.METHOD, ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface RedissonLock {
    String value(); // Lock의 이름 (고유값)
    long waitTime() default 5000L; // Lock획득을 시도하는 최대 시간 (ms)
    long leaseTime() default 2000L; // 락을 획득한 후, 점유하는 최대 시간 (ms)
}

 

RedissonLockAspect.java aop 추가

※ AOP에서 트랜잭션 분리를 위한 Transactional의 Propagation 설정!!

   @RedissonLock가 붙은 메서드는 항상 별도의 트랜잭션으로 동작하도록 설정해야한다.

@Slf4j
@Aspect
@Component
@RequiredArgsConstructor
public class RedissonLockAspect {

    private final RedissonClient redissonClient;

    @Transactional(propagation = Propagation.REQUIRES_NEW)
    @Around("@annotation(org.sparta.newsfeed.common.annotation.RedissonLock)")
    public Object redissonLock(ProceedingJoinPoint joinPoint) throws Throwable {
        MethodSignature signature = (MethodSignature) joinPoint.getSignature();
        Method method = signature.getMethod();
        RedissonLock annotation = method.getAnnotation(RedissonLock.class);

        // aop를 통한 해당 메소드의 파라미터 값 가져오기
        Object[] args = joinPoint.getArgs();
        String productId = args[0] == null ? "" : args[0].toString();

        String lockKey = productId + annotation.value();

        log.info("lock name = {}", lockKey);

        RLock lock = redissonClient.getLock(lockKey);

        Object result = null; // 리턴 값 저장할 변수

        try {
            boolean lockable = lock.tryLock(annotation.waitTime(), annotation.leaseTime(), TimeUnit.MILLISECONDS);
            if (!lockable) {
                log.info("Lock 획득 실패={}", lockKey);
                return null;
            }
            log.info("로직 수행");
            result = joinPoint.proceed();
        } catch (InterruptedException e) {
            log.info("에러 발생");
            throw e;
        } finally {
            log.info("락 해제");
            lock.unlock();
        }
        return result;
    }
}

 

메서드 적용 예!

@RedissonLock("#test")
public void updateStockRedisLock(String id , int cnt) {
    Optional<StockVo> result = stockRedisRepository.findById(id);
    if (result.isPresent()) {
        StockVo stockVo = result.get();
        stockVo.decrease(cnt);
        stockRedisRepository.save(stockVo);
    }
}

---

테스트

@SpringBootTest
class StockRedisServiceTest {

    @Autowired
    private StockRedisRepository stockRedisRepository;

    @Autowired
    private StockRedisService stockRedisService;

    private String ID;
    private final Integer CONCURRENT_COUNT = 100;

    @BeforeEach
    public void before() {
        StockVo stockVo = new StockVo("1" , "a" , 1000);
        StockVo saved = stockRedisRepository.save(stockVo);
        ID = saved.getId();
    }

    private void stockTest(Consumer<Void> action) throws InterruptedException {
        int originQuantity = stockRedisRepository.findById(ID).orElseThrow().getCnt();

        ExecutorService executorService = Executors.newFixedThreadPool(32);
        CountDownLatch latch = new CountDownLatch(CONCURRENT_COUNT);

        for (int i = 0; i < CONCURRENT_COUNT; i++) {
            executorService.submit(() -> {
                try {
                    action.accept(null);
                } finally {
                    latch.countDown();
                }
            });
        }

        latch.await();

        StockVo stockVo = stockRedisRepository.findById(ID).orElseThrow();
        assertEquals(originQuantity - CONCURRENT_COUNT, stockVo.getCnt());
    }

    @AfterEach
    public void after() {
        stockRedisRepository.deleteAll();
    }

    @Test
    @DisplayName("동시에 100명의 티켓팅 : 동시성 이슈")
    public void badStockRedisTest() throws Exception {
        stockTest((_no) -> stockRedisService.updateStockRedis(ID , 1));
    }

    @Test
    @DisplayName("동시에 100명의 티켓팅 : 분산락")
    public void redissonStockRedisTest() throws Exception {
        stockTest((_no) -> stockRedisService.updateStockRedisLock(ID , 1));
    }

}

 

실패 케이스

 

성공 케이스

성공 케이스를 보면 lock이 잘걸리면서 제대로 동시성 제어한 것으로 확인이 된다.