[Redis] Redis는 언제 활용할 수 있을까? 1편 - 분산락 구현하기 (1)

🌱 들어가기 전

조금 오래되긴 했지만 이전 포스팅에서 네임드락을 통한 분산락 구현을 진행하였다. 마지막 포스팅에서 레디스를 활용한다고 말했었는데, 계속 미루다가 이번에 레디스 공부를 하면서 간단하게 구현을 진행해보았다. 분산락에 대한 개념은 이전 포스팅에서 다루었기 때문에, 레디스를 이용해서 바로 분산락을 구현해보자. 또한, 지난 포스팅에서는 Jmeter를 활용하였지만, 이번 포스팅에서는 E2E 테스트를 통해 직접 구현을 해보고자 한다.
 

---

 

🌱 구현 상황

정말 간단하게 잔액을 의미하는 'Balance'라는 값에, 2개의 요청이 동시에 들어왔을 때 레디스를 통한 분산락을 구현하여 제어해보자.
위의 결과에서 최악의 상황이라면 Request 1, 2번 모두 80이라는 값을 반환하여 정상적으로 잔액이 갱신되지 않을 수 있다.
우리는 아래와 같은 플로우를 구현할 예정이다.

1. 공유 자원을 표현할 수 있는 이름을 생성하여 락의 키를 결정하기
2. NX 옵션을 통해서 해당 키가 존재하지 않는 경우에만 락을 획득하기
3. 이미 락을 획득 한 경우 nil을 반환하여 프로세스 대기하기
4. 자원의 작업이 끝나면 DEL을 통해 락 제거하기

 여기서 NX와 DEL은 레디스에서 사용하는 명령어다.
NX는 'IF NOT EXISTS SET'과 같이, 키가 존재하지 않을 경우에만 SET을 해준다. 싱글 스레드로 동작하는 레디스에서 키가 존재하는지 판단하기 위해서는 GET으로 확인 후 SET을 해야 한다. 하지만 멀티 스레드 환경인 스프링 같은 곳에서 사용할 때 정합성 문제가 발생할 수 있기 때문에, 한 번에 중복을 체크하고 삽입까지 진행할 수 있도록 만들어진 명령어라고 생각하면 이해하기 쉽다.
DEL의 경우 쉽게 예상할 수 있듯, 'DELETE'를 의미하는 키 삭제 기능을 의미한다.

우리는 스핀 락 (Spin Lock) 방식을 통해 락을 점유하고 있는지 계속 확인하면서, 락을 잡을 수 있다면 락을 잡고 아니면 대기하는 방식으로 구현하고자 한다. 락에 대한 키를 계속 점유하지 않도록 키는 임계 영역을 빠져나온 후 제거를 해줘야 하며, 키 설정 시 TTL도 함께 걸어주는 것이 좋다.

---

 

🌱 구현하기 - 환경 구축

하단에 나온 설정들은 앞으로 redis 관련 포스팅을 할 때 계속 사용할 예정이기 때문에, 이번 포스팅에서만 환경 구축에 대해 작성하고자 한다.

 

implementation("org.springframework.boot:spring-boot-starter-data-redis")
implementation("org.springframework.boot:spring-boot-starter-data-jpa")
runtimeOnly("com.h2database:h2")
testImplementation("io.rest-assured:rest-assured")

먼저, 기본 구현을 위해 필요한 dependency들이다. (언어는 Kotlin으로 작성할 예정이다.)
스프링에서 제공하는 redis 및 jpa, 테스트 환경 구축을 위한 h2와 RestAssured를 추가해주자.

@Configuration
class RedisConfig {

    @Bean
    fun redisConnectionFactory(): RedisConnectionFactory =
        LettuceConnectionFactory()

    @Bean
    fun redisTemplate(): RedisTemplate<String, String> {
        return RedisTemplate<String, String>().apply {
            connectionFactory = redisConnectionFactory()
            keySerializer = StringRedisSerializer()
            valueSerializer = StringRedisSerializer()
        }
    }
}

이번엔 레디스 관련 설정 클래스를 만들어주자.
레디스 클라이언트 중 자바 기반 환경에서 대표적으로 뽑히는 Lettuce를 사용하였으며, 레디스에 저장될 key, value는 현재 스펙에서 모두 String을 사용하면 되기 때문에 StringRedisSerializer()를 사용해주었다. 취향에 맞게 커스텀을 해주면 될 것 같다.

 

class RedisService(
    private val redisTemplate: RedisTemplate<String, String>
) : CacheService {
    override fun delete(key: String) {
        redisTemplate.delete(key)
    }
}

또한, 캐시의 키를 제거해주는 함수를 미리 작성해두었다. (자주 사용될 예정이라 제일 상단으로 빼두었다.)

 

---

 

🌱 구현하기 - 서비스 로직

먼저, 계좌 정보를 저장하기 위해 간단한 도메인 객체를 만들어보자.

@Entity
class Account(

    @Id
    val id: Long = 0L
) {
    var balance: Int = 0

    fun addBalance(amount: Int) {
        balance += amount
    }

    fun subtractBalance(amount: Int) {
        balance -= amount
    }
}

계좌의 잔액을 추가해주는 메서드와, 제거해주는 메서드 2가지를 만들었다.

 

import org.springframework.data.jpa.repository.JpaRepository

fun AccountRepository.getAccountById(id: Long): Account {
    return findAccountById(id) ?: throw IllegalArgumentException("존재하지 않는 계좌입니다")
}

interface AccountRepository : JpaRepository<Account, Long> {
    fun findAccountById(id: Long): Account?
}

그리고 간단하게 아이디를 통해 계좌를 조회해주는 확장 함수를 만들었다.

개인적으로 확장 함수를 활용해서 위와 같이 예외를 터트리는 방식을 선호하는데, 어떤 아키텍처를 사용하는지에 따라서 예외를 발생시키는 부분도 달라질 것 같다. (요즘은 생각이 바뀌어서, 아마 다르게 구현하지 않을까 싶다)

 

interface KeyGenerator {
    fun generateAccountKey(accountId: Long): String
}

@Component
class KeyGeneratorImpl : KeyGenerator {
    override fun generateAccountKey(accountId: Long): String {
        return "$accountId:account"
    }
}

다음으로, 레디스의 키를 생성해주는 메서드를 추가해주었다.

키는 겹치지 않는 것이 제일 중요한데, 여기서는 계좌에 대한 고유 아이디 (PK) 값과 'account'를 알려주는 String을 추가해서 만들어주었다. 실제로는 어떤 기능을 개발할지에 따라서 키를 만드는 방법도 다양하기 때문에 유의해야 된다.

 

@RestController
@RequestMapping("/accounts")
class AccountController(
    private val accountService: AccountService
) {

    @PostMapping("/create")
    fun createAccount(): ResponseEntity<AccountResponse> {
        val result = accountService.create()
        return ResponseEntity.ok(result)
    }

    @PostMapping("/deposit")
    fun deposit(@RequestBody request: DepositRequest): ResponseEntity<BalanceResponse> {
        val result = accountService.deposit(request)
        return ResponseEntity.ok(result)
    }

    @GetMapping("/balance/{accountId}")
    fun getBalance(@PathVariable accountId: Long): ResponseEntity<BalanceResponse> {
        val result = accountService.getBalance(accountId)
        return ResponseEntity.ok(result)
    }

    @PostMapping("/withdraw")
    fun withdraw(@RequestBody request: WithdrawRequest): ResponseEntity<BalanceResponse> {
        val result = accountService.withdraw(request)
        return ResponseEntity.ok(result)
    }
}

data class DepositRequest(
    val accountId: Long,
    val amount: Int
)

data class WithdrawRequest(
    val accountId: Long,
    val amount: Int
)

data class AccountResponse(
    val accountId: Long
)

data class BalanceResponse(
    val accountId: Long,
    val balance: Int
)

class AccountService(
    private val accountRepository: AccountRepository,
    private val keyGenerator: KeyGenerator,
    private val cacheService: CacheService
) {

    @Transactional
    fun create(): AccountResponse {
        val account = Account()
        val savedAccount = accountRepository.save(account)
        return AccountResponse(savedAccount.id)
    }

    @Transactional
    fun deposit(request: DepositRequest): BalanceResponse {
        val accountId = request.accountId
        val account = accountRepository.getAccountById(accountId)
        account.addBalance(request.amount)
        return BalanceResponse(account.id, account.balance)
    }

    @Transactional(readOnly = true)
    fun getBalance(accountId: Long): BalanceResponse {
        val account = accountRepository.getAccountById(accountId)
        return BalanceResponse(account.id, account.balance)
    }

    @Transactional
    fun withdraw(request: WithdrawRequest): BalanceResponse {
        val accountId = request.accountId
        val account = accountRepository.getAccountById(accountId)
        account.subtractBalance(request.amount)
        return BalanceResponse(account.id, account.balance)
    }
}

그렇다면, 실제 API 호출을 위해 간단한 컨트롤러 + 서비스 코드를 만들어주었다.

계좌를 생성하고, 입금하고, 출금하고, 조회하는 기능이다. 

이 기본 코드를 바탕으로 분산락 없이 한 번 테스트 코드를 작성해보자.

 

---

 

✔ Case 1 - 아무 락도 걸지 않았을 때 

class AccountControllerTest : IntegrationTestHelper() {

    @Test
    fun 락_없이_동시_출금요청() {
        // given
        val accountId = 계좌_생성()
        초기_잔액_설정(accountId, 100)
        val requestCount = 2

        // when
        // 동시에 20씩 출금을 요청함
        val executor = Executors.newFixedThreadPool(requestCount)
        val latch = CountDownLatch(requestCount)
        repeat(requestCount) {
            executor.submit {
                출금_요청(accountId, 20)
                latch.countDown()
            }
        }

        latch.await()

        // then
        val balanceResponse = 잔액_조회(accountId)

        // 원하는 결과는 60이겠지만, 다른 값이 나옴
        assertThat(balanceResponse.balance)
            .isNotEqualTo(60)
    }

    private fun 계좌_생성(): Long {
        return RestAssured.given()
            .log().all()
            .contentType(MediaType.APPLICATION_JSON_VALUE)
            .`when`()
            .post("/accounts/create")
            .then()
            .log().all()
            .extract()
            .`as`(object : TypeRef<AccountResponse>() {})
            .accountId
    }

    private fun 초기_잔액_설정(accountId: Long, amount: Int): BalanceResponse {
        val depositRequest = DepositRequest(accountId, amount)
        return RestAssured.given()
            .log().all()
            .contentType(MediaType.APPLICATION_JSON_VALUE)
            .`when`()
            .body(depositRequest)
            .post("/accounts/deposit")
            .then()
            .log().all()
            .extract()
            .`as`(object : TypeRef<BalanceResponse>() {})
    }

    private fun 출금_요청(accountId: Long, amount: Int): BalanceResponse {
        val withdrawRequest = WithdrawRequest(accountId, amount)

        return RestAssured.given()
            .log().all()
            .contentType(MediaType.APPLICATION_JSON_VALUE)
            .`when`()
            .body(withdrawRequest)
            .post("/accounts/withdraw")
            .then()
            .log().all()
            .extract()
            .`as`(object : TypeRef<BalanceResponse>() {})
    }

    private fun 잔액_조회(accountId: Long): BalanceResponse {
        return RestAssured.given()
            .log().all()
            .`when`()
            .get("/accounts/balance/$accountId")
            .then()
            .log().all()
            .extract()
            .`as`(object : TypeRef<BalanceResponse>() {})
    }
}

우선, 아무 락도 걸지 않은 상태로 요청을 한다면 어떻게 될까? 위 테스트 시나리오는 다음과 같다.

1. 초기 잔액을 100으로 세팅한다.
2. 2개의 요청이 동시에 20씩 출금 요청을 진행한다.
3. 잔액을 조회했을 때 100 - 20 - 20 = 60이 나오기를 기대한다.

2개의 요청이 동시에 오고 있으며, 테스트 코드가 먼저 끝나는 것을 방지하기 위해 모든 요청이 종료될 때까지 await()으로 대기할 수 있도록 만들었다. 하지만, 테스트 코드를 보면 isNotEqualTo(60)으로, 60이라는 값이 나오지 않은 것을 볼 수 있다.

이는, 동시성 문제로 인해서 '20 출금하라'라는 요청이 동시에 오면서 첫 번째 요청, 두 번째 요청 모두 100 -> 80으로 출금을 진행했기 때문에 발생한 문제였다.

 

---

 

✔ Case 2 - 레디스의 get - set 활용하기

그렇다면, 레디스를 활용해 보면 어떨까?

우리는 초기에 구현 시나리오에서 '키가 존재하지 않을 경우 락을 획득하고, 존재하면 락을 획득할 때까지 대기'하도록 플로우를 구성하고자 했다. 간단하게 레디스를 활용해서 키가 존재하는지 확인하기 위해 get으로 가져오고, 존재하지 않는 경우에 set을 통해 락을 획득하는 로직을 작성해보자.

interface KeyGenerator {
    fun generateAccountKey(accountId: Long): String
}

@Component
class KeyGeneratorImpl : KeyGenerator {
    override fun generateAccountKey(accountId: Long): String {
        return "$accountId:account"
    }
}

먼저, 레디스의 키를 생성해주는 메서드를 추가해주었다.    
키는 겹치지 않는 것이 제일 중요한데, 여기서는 계좌에 대한 고유 아이디 (PK) 값과 'account'를 알려주는 String을 추가해서 만들어주었다. 실제로는 어떤 기능을 개발할지에 따라서 키를 만드는 방법도 다양하기 때문에 유의해야 된다.

@Component
class RedisService(
    private val redisTemplate: RedisTemplate<String, String>
) : CacheService {
    override fun get(key: String): String? {
        return redisTemplate.opsForValue().get(key)
    }

    override fun set(key: String, value: String, duration: Duration) {
        redisTemplate.opsForValue().set(key, value, duration)
    }
}

다음으로, redisTemplate을 활용하여 String 값을 조회하고 설정하는 함수를 추가하였다.

둘 모두 opsForValues().set()과 opsForValues().get()을 활용하면 간단하게 구현할 수 있다.

set의 경우 해당 키가 언제까지 유효하도록 만들지 TTL 값을 설정해줄 수 있는데, Duration을 사용하여 원하는 만큼 지정해줄 수 있다. 보통 오래 걸리는 로직이 아니라면 30초 ~ 1분 정도로 설정하는 경우를 많이 본 것 같다.

 

@PostMapping("/withdraw-non-atomic-lock")
fun withdrawUsingNonAtomicLock(@RequestBody request: WithdrawRequest): ResponseEntity<BalanceResponse> {
    val result = accountService.withdrawUsingNonAtomicLock(request)
    return ResponseEntity.ok(result)
}

@Transactional
fun withdrawUsingNonAtomicLock(request: WithdrawRequest): BalanceResponse? {
    val accountId = request.accountId
    val accountKey = keyGenerator.generateAccountKey(accountId)

    while (true) {
        if (cacheService.get(accountKey) == null) {
            cacheService.set(accountKey, "account-withdraw", Duration.ofSeconds(5))
            break;
        }
        Thread.sleep(1000)
    }

    val account = accountRepository.getAccountById(accountId)
    account.subtractBalance(request.amount)

    // 캐시 제거 - TTL을 짧게 잡기는 했지만 명시적으로 제거함
    cacheService.delete(accountKey)
    return BalanceResponse(account.id, account.balance)
}

그리고, 실질적인 비즈니스 코드를 작성해주었다.

먼저, get()을 통해서 키가 존재하는지 확인한 다음, 존재하지 않는다면 set을 해주고, 존재하는 경우 1초씩 대기하며 재시도를 진행한다.

참고로, set을 통해 세팅해주는 value 값은 자유롭게 설정해주어도 된다. (의미없는 String 값도 상관없다.) 

중요한 건, 다른 비즈니스 로직과 키만 겹치지 않도록 만들어 주는 것이 중요하다.

추가적으로, Facade 패턴 등을 도입해서 락을 얻고 반환하는 로직은 추상화할 수 있다.
이번에는 레디스를 중심으로 적용하다 보니 이에 대해 따로 고려하지 않았지만, 지난 포스팅에서 퍼사드 패턴을 활용했기 때문에 해당 내용을 참고하여 리팩터링 해도 좋을 것 같다는 생각이 든다.

 

우리가 원하는 로직대로 작성했기 때문에, 이제 결과값으로 60이 나오지 않을까?

@Test
fun 단순_락을_이용한_동시_출금요청_원자성을보장하지않음() {
    // given
    val accountId = 계좌_생성()
    초기_잔액_설정(accountId, 100)
    val requestCount = 2

    // when
    // 동시에 20씩 출금을 요청함
    // 캐시를 사용하긴 했지만 완전히 원자성을 보장하지 않는 로직
    val executor = Executors.newFixedThreadPool(requestCount)
    val latch = CountDownLatch(requestCount)
    repeat(requestCount) {
        executor.submit {
            락을_통한_출금_요청(accountId, 20)
            latch.countDown()
        }
    }

    latch.await()

    // then
    val balanceResponse = 잔액_조회(accountId)

    // 마찬가지로 원하는 결과는 60이겠지만, 다른 값이 나옴
    assertThat(balanceResponse.balance)
        .isNotEqualTo(60)
}

private fun 락을_통한_출금_요청(accountId: Long, amount: Int): BalanceResponse {
    val withdrawRequest = WithdrawRequest(accountId, amount)

    return RestAssured.given()
        .log().all()
        .contentType(MediaType.APPLICATION_JSON_VALUE)
        .`when`()
        .body(withdrawRequest)
        .post("/accounts/withdraw-non-atomic-lock")
        .then()
        .log().all()
        .extract()
        .`as`(object : TypeRef<BalanceResponse>() {})
}

하지만, 예상과 다르게 여전히 isNotEqualTo()를 통해 60이 나오지 않은 것을 볼 수 있다. 왜 이럴까?

사실, 앞에서 이미 답을 냈다. 

우리의 스프링 애플리케이션은 멀티 스레드 기반이기 때문에, 레디스가 싱글 스레드로 동작한다고 해도 하나의 스레드가 get - set을 하는 중간에 다른 스레드가 get을 충분히 할 수 있는 상황이 만들어졌기 때문이다.

 

그림으로 보면 다음과 같은 상황이다. 레디스 내부에서 싱글 스레드로 열심히 명령어를 처리하더라도, 애플리케이션 레벨에서 들어오는 요청들은 여러 스레드가 동시에 들어오기 때문에 20, 20씩 출금하라는 요청이 적용되지 않았던 것이다.

그렇다면, GET - SET을 동시에 하도록 = 원자성을 보장하도록 만들 수 없을까?

바로, 앞서 말했던 것처럼 NX을 사용하면 된다.

 

---

 

✔ Case 3 - 레디스의 NX 활용하기

override fun setNX(key: String, value: String, duration: Duration): Boolean {
    return redisTemplate.opsForValue().setIfAbsent(key, value, duration) ?: false
}

NX의 경우 redisTemplate에서 제공하는 setIfAbsent()를 사용하면 된다.

우리가 애플리케이션 레벨에서 GET을 통해 키가 존재하는지 확인하고, SET을 통해서 값을 세팅해주는 과정을 레디스 내에서 만든 명령어라고 생각하면 된다. 만약 키가 이미 존재해서 값을 세팅하지 못했다면 false를, 키가 존재하지 않아서 값을 세팅했다면 true를 반환한다.

 

@PostMapping("/withdraw-atomic-lock")
fun withdrawUsingAtomicLock(@RequestBody request: WithdrawRequest): ResponseEntity<BalanceResponse> {
    val result = accountService.withdrawUsingAtomicLock(request)
    return ResponseEntity.ok(result)
}

@Transactional
fun withdrawUsingAtomicLock(request: WithdrawRequest): BalanceResponse {
    val accountId = request.accountId
    val accountKey = keyGenerator.generateAccountKey(accountId)

    while (!cacheService.setNX(accountKey, "account-withdraw", Duration.ofSeconds(5))) {
        Thread.sleep(1000)
    }

    val account = accountRepository.getAccountById(accountId)
    account.subtractBalance(request.amount)

    // 캐시 제거
    cacheService.delete(accountKey)
    return BalanceResponse(account.id, account.balance)
}

실제 비즈니스 로직에 적용해보자. setNX를 통해 true, false 값을 반환하기 때문에 값을 세팅하지 못했을 경우 1초씩 대기하도록 while문을 통해 쉽게 제어할 수 있게 되었다.

 

그렇다면, 테스트 코드를 통해 결과를 확인해보자.

@Test
fun 분산락을_통한_동시_출금요청() {
    // given
    val accountId = 계좌_생성()
    초기_잔액_설정(accountId, 100)
    val requestCount = 2

    // when
    // 동시에 20씩 출금을 요청함
    val executor = Executors.newFixedThreadPool(requestCount)
    val latch = CountDownLatch(requestCount)
    repeat(requestCount) {
        executor.submit {
            분산락을_통한_출금_요청(accountId, 20)
            latch.countDown()
        }
    }

    latch.await()

    // then
    val balanceResponse = 잔액_조회(accountId)

    // 락으로 인해서 순차적으로 처리되기 때문에 정상적으로 60 반환
    assertThat(balanceResponse.balance)
        .isEqualTo(60)
}

private fun 분산락을_통한_출금_요청(accountId: Long, amount: Int): BalanceResponse {
    val withdrawRequest = WithdrawRequest(accountId, amount)

    return RestAssured.given()
        .log().all()
        .contentType(MediaType.APPLICATION_JSON_VALUE)
        .`when`()
        .body(withdrawRequest)
        .post("/accounts/withdraw-atomic-lock")
        .then()
        .log().all()
        .extract()
        .`as`(object : TypeRef<BalanceResponse>() {})
}

드디어 우리가 원하는대로 60을 반환하는 것을 확인할 수 있다!

하지만 이대로 끝내기에는 뭔가 아쉽기 때문에, 조금 더 귀찮은 방법으로 원자성을 보장하는 방법을 소개해보고자 한다.

 

---

 

글이 생각보다 너무 길어져서 1, 2부로 나누어서 작성하였습니다.

다음 포스팅: https://cl8d.tistory.com/125