어떤 서비스든 판매하는 상품이 있다면 상품에 달린 재고도 관리 대상입니다. 이전에 구현했던 재고 차감 로직을 복기해보고 개선해보려고 합니다. 먼저 상품 재고와 관련된 보통의 요구사항입니다.
1. 상품은 등록된 재고량을 초과하여 판매할 수 없다.
2. 재고가 없는 상품은 주문할 수 없다.
3. 상품 재고는 미리 등록해둘 수 있어야한다.
해당 포스팅에서는 1번과 2번에 집중되어있습니다.
📍 재고 차감편
보통 재고 차감시 사용하는 라이브러리는 redis 입니다. 왜 redis가 이용되고 있는 걸까요? 아래와 같은 이유들이 있습니다.
재고 차감에 redis를 사용하는 이유
1. 재고 차감은 실시간으로 이루어져야 하며, 높은 성능이 요구되는 작업이다.
-> Redis 는 Disk 기반이 아닌 Memory 기반의 저장소로서 RDB 보다 적합하다.
2. 재고 차감 시스템에서는 재고 감소와 관련된 여러 단계가 원자적으로 실행되어야한다.
-> Redis 는 Single 스레드로 동작한다.
이와 같은 이유로 redis 를 채택했다면 실제 구현도 그런 방향으로 갔을까요?
요구사항을 얼마나 잘 만족시켰는지 자가 판단 해보겠습니다.
private const val PRODUCT_ITEM_ID_KEY = "productItem"
@Component
class StockAdapter(
private val stringRedisTemplate: StringRedisTemplate,
) : DecreaseStockPort {
override fun decrease(productItemId: String, userId: String, stock: Long) {
val list: MutableList? = stringRedisTemplate.execute {
val key = productItemId
it.watch(key.toByteArray())
it.multi()
val stockCount = stringRedisTemplate
.opsForHash<string,string>()
.get(PRODUCT_ITEM_ID_KEY, productItemId)
if (stockCount == null || stockCount.toLong() <= 0) {
throw IllegalStateException("$key 재고 소진, 현재 재고 $stockCount")
}
val decreaseStockCount: Long? =
stringRedisTemplate
.opsForHash<string,string>()
.increment(PRODUCT_ITEM_ID_KEY, productItemId, stock)
logger().info { "$key 남은 재고 $decreaseStockCount userId=$userId" }
if (decreaseStockCount == null || decreaseStockCount < 0) {
it.discard()
throw IllegalStateException("$key 재고 소진, 현재 재고 $stockCount")
}
return@execute it.exec()
}
logger().info { "세션콜백 리턴 확인 $list" }
}
}
위 코드를 해석하기 위해서는 redis 명령어에 대한 이해가 필요합니다. 그래서 준비한 쪽집게 강의 입니다.
사진 오른편에는 재고 차감 플로우에 대한 설명을, 왼편에는 레디스 트랜잭션에 관여하는 다섯가지 명령어를 정리해두었습니다. 각각의 명령어가 의미하는 바를 이해하는 것도 필요하지만 트랜잭션이라는 개념 하에 서로 연관된 맥락을 파악하는 것이 중요할 것 같습니다. 저는 아래 과정을 통해 redis 트랜잭션 원리를 이해했습니다. 코드를 토대로 설명해보겠습니다.
watch는 특정 키를 모니터링하기위한 명령어입니다. 해당 키에 대한 감시가 계속 진행되는 와중에 multi 명령어로 트랜잭션을 시작할 수 있습니다. 해당 명령어를 호출한 이후의 다른 명령어들은 트랜잭션 대기열에 저장됩니다.
트랜잭션 대기열에 있는 트랜잭션들이 실질적으로 실행되는 시점은 exec 명령어를 호출할 때입니다. 해당 명령어를 통해 트랜잭션이 실행되면 multi 이후부터 exec 이전까지의 모든 명령어가 연속적으로 실행됩니다.
주의할 점은 watch 로 감시하던 키에 변경이 감지되지 않아야만 트랜젹션이 성공적으로 완료된다는 사실입니다. 만약 watch 를 호출한 이후, productItemId4에 대한 변경이 감지되면 해당 트랜잭션은 취소됩니다. 트랜잭션이 취소될 때는 discard 가 호출되어 실행 전으로 롤백됩니다.
discard 명령어는 트랜잭션을 롤백하는 역할을 합니다.
저는 여기서 약간 혼란이었습니다... 제가 알기론 redis 에 롤백 체계가 존재하지않는데, 블로그 여기 저기 설명으로는 마치 discard 가 그런 역할을 담당하는 것처럼 보였기때문입니다. Chat GPT 에게 물어보았습니다.
마치 롤백을 제공하는 것 같은데요... 얘를 100 퍼센트 믿는 건 바보니까 redis 공식문서를 뒤져보았습니다.
공식문서 Rollback 설명
트랜잭션에 대한 롤백을 지원하지 않는다고 합니다. 그럼 discard 정체는 도대체 뭐란 말임.. discard 에 대한 설명도 공식문서에서 찾아보았습니다.
공식문서 Discard 설명
정리하자면, discard 는 트랜잭션에 의한 변경(데이터)를 롤백하는 게 아니라 트랜잭션 자체를 롤백하는 것, 그러니까 시작되었던 트랜잭션을 없던 일로 하여 취소하는 것을 뜻합니다. 롤백은 롤백이긴 하네.. 그도 그럴 것이 실질적인 트랜잭션의 시작은 exec 명령어가 호출되는 시점이므로, 데이터 변경이 아직 일어나지 않았을 때인데 변경 데이터를 롤백한다는 말 자체가 성립이 안되네요.
아무튼 discard 가 실행되면 종국에는 두 가지 일이 발생합니다.
1. 대기열에 있던 명령어들이 취소되고 해당 키에 대한 모니터링이 해제되고, 새로운 트랜잭션을 시작할 수 있는 상태로 돌아간다.
2. discard 를 통해 트랜잭션이 취소되면 exec 의 반환 값은 null 이 된다.
하지만 현재 코드에서 exec 반환 값은 항상 null 입니다.
지금까지 익혔던 이론대로 다시 코드를 살펴보면 허점이 있습니다. 구멍을 찾아보았습니다. 👀
private const val PRODUCT_ITEM_ID_KEY = "productItem"
@Component
class StockAdapter(
private val stringRedisTemplate: StringRedisTemplate,
) : DecreaseStockPort {
override fun decrease(productItemId: String, userId: String, stock: Long) {
val list: MutableList? = stringRedisTemplate.execute {
val key = productItemId
it.watch(key.toByteArray())
it.multi()
val stockCount = stringRedisTemplate
.opsForHash<string,string>()
.get(PRODUCT_ITEM_ID_KEY, productItemId)
if (stockCount == null || stockCount.toLong() <= 0) {
throw IllegalStateException("$key 재고 소진, 현재 재고 $stockCount")
}
val decreaseStockCount: Long? =
stringRedisTemplate
.opsForHash<string,string>()
.increment(PRODUCT_ITEM_ID_KEY, productItemId, stock)
logger().info { "$key 남은 재고 $decreaseStockCount userId=$userId" }
if (decreaseStockCount == null || decreaseStockCount < 0) {
it.discard()
throw IllegalStateException("$key 재고 소진, 현재 재고 $stockCount")
}
return@execute it.exec()
}
logger().info { "세션콜백 리턴 확인 $list" }
}
}
17~18 번 라인에서 트랜잭션 내부에서 해당 키에 접근하고 있습니다. 이럴 경우 stockCount는 항상 null 입니다.
트랜잭션 명령어의 실행은 exec 호출 시점에 발생하기 때문입니다.
반면에, 23~24번 라인에서의 명령어 결과값 decreaseStockCount 는 항상 null 이 아닙니다. 트랜잭션 내부의 redisConnection(`it`) 을 통한 것이 아니라, 직접 stringRedisTemplate 을 통해 명령하고 있기 때문입니다. decreaseStockCount 은 트랜잭션 실행과 별도로 돌아가는 프로세스의 값입니다.
이렇게 되면, 바로 아래 위치한 재고 검증 코드가 의미 없어지게됩니다. decreaseStockCount 값 자체가 Atomic 한 트랜잭션 안에서의 결과값이 아니니까요.
이쯤되니 저는 이런 의문이 들기 시작했습니다.
"왜 이 뭉치면 살고 흩어지면 죽는 다섯 개의 명령어를 분리해놓았을까"
보통 수동 트랜잭션 관리의 장점은 다음과 같은 이유들이 있었습니다.
* 코드에서 트랜잭션의 시작과 종료를 명확히 인식할 수 있다.
* 수동 트랜잭션 관리를 통해 명시적으로 커밋하거나 롤백할 수 있어서 특정 조건에 따라 트랜잭션을 롤백하거나 일부 조건을 검사한 후 커밋하는 등의 제어가 가능하다.
특히, Redis 의 트랜잭션은 명령어 수준에서 지정됩니다. 특정 명령어들을 트랜잭션으로 묶고, 다른 명령어들은 트랜잭션과는 독립적으로 실행할 수 있습니다. 그럼에도 불구하고, 레디스에도 @Transactional 같은 게 있다면 참 좋을 것 같다는 생각이 드는 찰나, @Transactional 같은 게 아니라 바로 그 @Transactional 을 이용할 수 있는 방법을 찾게 되었습니다.
[참고 블로그](https://sabarada.tistory.com/178)
해당 블로그에서의 설명대로라면 @Transactional를 사용했을 경우 Read-only 커맨드는 Transactional Queue에 들어가지 않고 바로 실행되며 Write 커맨드만 들어가게됩니다. 그렇다면 재고가 마이너스인 경우 바로 캐치해낼 수 있습니다. 이전 코드에서의 이슈를 @Transactional 로 해결할 수 있습니다. 이슈를 명확하게 정리해보겠습니다.
SessionCallback 을 통한 multi-exec 구현 방식은 내부에서 조회한 결과로 discard 나 예외를 처리할 수 없습니다. Read-only 커맨드가 바로 실행되지 않기 때문입니다. SessionCallback 외부에서 차감할 재고가 있는지 확인해야하는 수 밖에 없는데, 문제는 Atomic 이 SessionCallback 내부에서만 보장된다는 사실입니다. 그 세상 밖은 무수한 원자들이 요청보내고 있어서 이전의 요청이 먼저 재고를 선점해버리면 SessionCallback 내에서 실행당시의 재고는 마이너스가 될 수 있습니다. 그러니까, 그들이 외부에서 재고 검증할 당시에는 재고가 존재해서 SessionCallback 까지 도달하지만, 그 동시에 SessionCallback 내부의 트랜잭션 대기열에 차감할 재고 명령어가 트랜잭션 실행을 앞두고 있을 것입니다. 재고 경합의 여지가 있습니다. 간단하게 말하면, 동시성 이슈(concurrent issue) 입니다.
문제를 해결하기 위해 다음 세가지 방법을 살펴보겠습니다.
1. @Transactional
2. LuaScript
3. Redission
세 가지 방법에 대한 검증은 아래 테스트 코드를 통해 진행하였습니다.
@Test
fun decrease() {
val invokeAll = executorService.invokeAll(IntArray(50).map {
Callable {
sut.decrease(
productItemId = "104",
userId = "${Random(Int.MAX_VALUE).nextInt()}",
-1
)
}
})
val successQueue = ConcurrentLinkedQueue<UUID>()
val failQueue = ConcurrentLinkedQueue<UUID>()
val block1: suspend CoroutineScope.() -> Unit = {
val block: suspend CoroutineScope.() -> Unit = {
for (future in invokeAll) {
kotlin.runCatching {
future.get()
}.onSuccess {
successQueue.add(UUID.randomUUID())
}.onFailure {
failQueue.add(UUID.randomUUID())
}
}
}
val async: Deferred<Unit> = async(block = block)
async.await()
}
kotlinx.coroutines.runBlocking(block = block1)
successQueue.size shouldBe 30
failQueue.size shouldBe 20
val actual = stringRedisTemplate.opsForHash<String, String>().get("productItem", "104")
actual shouldBe "0"
}
* 코루틴을 이용한 테스트 코드 입니다.
* 50개의 쓰레드에서 동시에 sut.decrease 를 호출하고, 각 작업의 성공과 실패를 추적하여 그 수를 확인합니다.
* 모든 요청이 끝나고 상품 재고가 마이너스로 떨어지지 않았는지 검증합니다.
해당 테스트 코드에 대한 자세한 설명은 아래를 참조해주세요.
본격적인 시도에 앞서 기존의 코드도 리팩토링해보았습니다.
multi-exec 구문 내부의 실행 쿼리 결과에 대한 불필요한 검증 로직을 제거했습니다.
fun decrease_v2(productItemId: String, userId: String, stock: Long) {
val list: MutableList? = stringRedisTemplate.execute {
val stockCount = it.commands().get(productItemId.toByteArray())
logger().info { "stockCount=$stockCount" }
if (stockCount == null) {
it.discard()
throw IllegalStateException("$productItemId 재고 소진 > 체크 시 $stockCount")
}
val key = productItemId
it.watch(key.toByteArray())
it.multi()
it.commands().hIncrBy(PRODUCT_ITEM_ID_KEY.toByteArray(), key.toByteArray(), stock)
return@execute it.exec()
}
logger().info { "세션콜백 리턴 확인 $list" }
}
이제 본격적으로 세가지 방법을 트라이 해보겠습니다. 화이팅.
1. @Transactional 실행기
앞서 링크해둔 블로그에서 배운대로 redisTemplate 의 transactional 활성화를 시켰고, transactionManager 를 bean 으로 등록해두었습니다.
기존 재고 차감 메소드에 @Transactional 만 붙여 테스트 코드를 실행해보겠습니다.
재고 수량이 30개인데, 50개의 요청 중 34건이 성공합니다. 재고는 마이너스까지 떨어진 상황입니다. 재고가 추가될 수 없는 한정 수량 상품이었다면 장애에 해당합니다.. 왜 이런 결과가 나오게 되었을까요.. 스프링 @Transactional 의 기본 동작 원리 때문인 것 같습니다. @Transactional 이 내부적으로 사용하는 트랜잭션 컨텍스트는 기본적으로 ThreadLocal 을 사용합니다. 현재 스레드에 대해서만 트랜잭션을 보장한다는 의미입니다. 또 다른 의미가 있다면, ThreadLocal 은 각각의 스레드에게 독립적인 저장 공간을 제공하므로 한 스레드에서 수행되는 트랜잭션이 다른 스레드에게 영향을 미치지 않는다는 것입니다. 두 가지 의미를 짬뽕해서 생각해보면, 비동기적인 작업이 실행되는 환경에서는 트랜잭션이 공유되지않으므로 동시성 이슈를 해결할 수 없다는 결론이 나옵니다. 그럼 어떻게 해야할까요? 여기저기 발품 팔아 찾아보니, reactive-friendly 한 트랜잭션 관리 방법을 사용해야한다고 권고하고 있었습니다. 쉽게말하면.. 분산 환경에서 트랜잭션을 관리할 수 있는 다른 방법을 찾아야한다는 것입니다. 이 방법은 multi-exec 구문에서 readOnly 커맨드는 가능하지만, 분산 환경에서의 원자성 보장은 하지 않습니다.
한편, @Transactional 을 이용하려면 TransactionManager 가 필요합니다. Jpa 나 Jdbc 의 TransactionManager 에 기생하는 방법이 있지만, 만약 해당 모듈이 DBMS를 사용하고 있지않다면 무거운 의존성을 불필요하게 가져가야합니다. 이 지점 또한 ...👎 입니다.
2. LuaScript 실행기
다음은 LuaScript 를 이용하는 방법입니다.
SessionCallback 내부에서 RedisScript 를 통해 직접 명령합니다.
여러 명령어를 Redis 서버에 전달하여 실행하는 방식입니다. 이 경우는 @Transactional 처럼 조회따로 명령 따로가 가능한 구조인 것 같습니다. 테스트를 실행시켜보겠습니다.
ㅋ 성공합니다...! 재고가 30개 있을 때, 50개의 요청 중 20개가 `재고없음`처리 되었습니다. 모든 요청 처리 후, 해당 상품 재고의 수량이 0개인 것으로 보아, 경합으로 인해 재고가 마이너스로 떨어지는 케이스도 이 방법으로 커버되는 것 같습니다. 그렇지만 Lua Script 도 한계는 있었습니다. Lettuce에서는 `ReadFrom` 옵션을 `REPLICA_PREFERRED`로 설정하여 조회 트래픽을 주로 복제본으로 보내는 부하 분산 기능을 제공합니다. 그러나 Lua Script를 사용할 때, Lettuce에서는 반드시 Redis 마스터에 보내야 하는 제약이 있습니다. 대용량 트래픽을 처리하는 상황에서는 부하 분산을 유지하면서도 Lua Script를 사용해야 하는데, 이러한 상황에서는 Lettuce의 부하 분산 기능을 활용하지 못한다는 이슈가 있다고 합니다.
대용량 트래픽이 모든 서비스의 고려대상이 아니긴 하지만, 언젠가는 고려대상이 될 수 있으니까요..어쨌든 Lua Script 도 완전한 해결책이 아니라는 점은 짚고 넘어가겠습니다. 그리고 개인적으로는 스크립트 형식이 마음에 들지 않았습니다. 파악해야하는 대상 코드가 찾기 힘든 곳에 위치한다는 점이 찝찝합니다.
3. Redission 실행기
23번 라인, redissonClient 를 이용해 lock을 가져옵니다.
26번 라인, 해당 lock 으로 locking 을 시도하여 key 를 선점합니다. 여기서는 상품 아이템 아이디입니다.
하위 로직에서는 재고 수량을 체크합니다. 크게 어려운 부분은 없습니다. 테스트를 실행시켜보겠습니다.
동시에 여러 요청이 와도 원자성이 지켜지는 모습입니다. 내부적으로 해당 key 에 대한 접근도 할 수 있어서 재고가 마이너스로 떨어지는 경우도 커버할 수 있습니다. 근데 여러 블로그를 살펴보니 redisson 라이브러리에는 많은 기능이 있는데, lock 만 사용하기에는 조금 무겁다라는 평이 있습니다. 케바케에 해당하는 의견이겠지만, 제가 시도했던 모든 방법 중 제일 베스트는 redisson 라이브러리였습니다.
지금까지 재고 차감 구현과 동시성 이슈에 대한 개선을 시도해보았습니다. 복기해보니 기존 구현은 구멍난 헛간처럼 느껴집니다.. 빈틈이 줄줄 새는 듯 합니다.. 하지만 한편으로는 이런 생각도 들었습니다. 완전무결한 구현은 존재할 수 있는지, 만약 그렇다면 어떤 서비스가 그래야하는지. 회사의 성장 속도에 맞춰서, 트래픽이 많고 적음에 맞춰서, 타협이 가능한 지점을 찾아나가는 것이 차라리 덜 멍청한 게 아닐까하는 생각이요. 저 혼자 정신 승리를 하고 있는 것 일까요? 그렇다면 백기 올려..🏳️
'생각' 카테고리의 다른 글
| 서킷브레이커의 사용설명서는 없다 with.연동 라이브러리들 (3) | 2024.01.29 |
|---|---|
| 믿거나 말거나, 데이터 독이 data dog인 이유 (0) | 2024.01.28 |
| 똑똑 거기 누구 없소 (2) | 2023.11.10 |
| DDD 와 TDD 는 함께 가야할 것 같아요. (3) | 2023.01.30 |
| 이상한 나라의 객체 : 객체는 思物 이다. (2) | 2023.01.05 |
