Lost Update 문제를 락으로만 해결할 수 있을까?
🔏 앞서 살펴본 비관 락 (SELECT FOR UPDATE)의 방법도 무리 없음
@Lock PESSIMISTIC_FORCE_INCREMENT (읽기 조차 막아버림)
=> 바꾸는 테이블 데이터가 HOST 테이블 밖에 없고
=> 현재 아키텍처 구조상 트랜잭션이 길어지는 경우도 없기 때문
분산 서버에서 DB 동기화를 한다는 가정
DISK I/O 방지 용으로 REDIS 를 사용하기 때문에, REDISSON 사용 할 수 있음
# zookeeper 실행
$ bin/windows/zookeeper-server-start.bat config/zookeeper.properties
# kafka server 실행
$ bin/windows/kafka-server-start.bat config/server.properties
@EnableCaching
@Configuration
public class RedisConfig {
@Value("${spring.redis.host:localhost}")
private String host;
@Value("${spring.redis.port:6379}")
private String port;
private static final String REDISSON_HOST_PREFIX = "redis://";
// Redisson client
@Bean
public RedissonClient redissonClient(){
Config config = new Config();
config.useSingleServer().setAddress(REDISSON_HOST_PREFIX + host + ":" + port);
return Redisson.create(config);
}
// Redis template
@Bean
public RedisConnectionFactory redisConnectionFactory() {
RedisStandaloneConfiguration redisStandaloneConfiguration = new RedisStandaloneConfiguration();
redisStandaloneConfiguration.setHostName(host);
redisStandaloneConfiguration.setPort(Integer.parseInt(port));
LettuceConnectionFactory lettuceConnectionFactory = new LettuceConnectionFactory(redisStandaloneConfiguration);
return lettuceConnectionFactory;
}
@Bean
public RedisTemplate<String, String> redisTemplate() {
RedisTemplate<String, String> redisTemplate = new RedisTemplate<>();
redisTemplate.setConnectionFactory(redisConnectionFactory());
redisTemplate.setKeySerializer(new StringRedisSerializer());
redisTemplate.setValueSerializer(new StringRedisSerializer());
redisTemplate.setHashKeySerializer(new StringRedisSerializer());
redisTemplate.setHashValueSerializer(new StringRedisSerializer());
return redisTemplate;
}
}
💥Could not resolve placeholder 'spring.redis.host' in value "${spring.redis.host}"
@Value 읽는 과정에서 annotation 못 읽는 문제 발생하여, 디폴트 값으로 직접 넣어주었음
3. 비즈니스 로직
@Override
public void saveRequest(Principal principal, ResrvHistRequestDto histRequestDto) throws Exception {
String userId = principal.getName();
Optional<User> user = userRepository.findById(userId);
Optional<Host> host = hostRepository.findById(Long.valueOf(histRequestDto.getHnum()));
if(user.isPresent()){
String lockKey = "LockKey:" + userId; // Redisson Lock 시작
RLock lock = redissonClient.getLock(lockKey);
try {
boolean isLocked = lock.tryLock(3, 3, TimeUnit.SECONDS);
if (!isLocked) {
throw new NoLockException();
}
int curReqPpl = histRequestDto.getReqPpl();
long hnum = host.get().getHnum();
int curIvtPplByHost = cahceCheck(hnum);
if (curReqPpl > curIvtPplByHost) {
throw new NoAvailableExeption(String.valueOf(hnum));
} else{
redisService.deductAvailPpl(hnum, curReqPpl); // Redis 차감
host.get().updateIvtPpl(curIvtPplByHost, histRequestDto.getReqPpl());
hostRepository.save(host.get()); // DB 차감 -> Write Through
log.info("host 예약 가능인원 차감 : {}", host.get().getIvtPpl());
..... (카프카 영역).....
}
} catch (Exception e) {
log.info("예약 과정에서의 오류 : {}", e.getMessage());
}finally {
lock.unlock(); // Redisson Lock 반납
}
} else{
throw new NoUserException();
}
}
💥 redis server 가동시키고 테스트 진행해야함
4. 코드 작성시 고려한 점
예외 처리 : 커스텀 에러에선 RuntimeException을 사용 (필요 없는 검사 예외는 바깥으로 문제를 전파하기만 한다.
public class NoLockException extends RuntimeException {
public NoLockException() {
super("Redisson Lock Failure");
}
}
슬로’s Dev Log: 최종 RuntimeException이 servlet container에 도달하면 500 에러 발생
동시성 : 최대한 적은 부분만 락을 걸어야 성능 문제를 최소화 할 수 있음
메서드 : 성격에 맞게 분리하여 작성함
private int cahceCheck(long hnum) {
// Look aside
// 1) 캐시 먼저 확인
int curIvtPplByHost = -1;
if (redisService.findRedisKey(hnum)) {
curIvtPplByHost = redisService.getAvailPpl(hnum);
log.info("현재 호스트 redis 예약 재고 : {}", curIvtPplByHost);
} else{
// 2) 없다면 DB 조회
curIvtPplByHost = hostRepository.countIvtPpl(hnum);// DB 조회
redisService.setInitialPpl(hnum, curIvtPplByHost); // redis set
log.info("현재 호스트 db 예약 재고 : {}", curIvtPplByHost);
}
return curIvtPplByHost;
}
🥑결과
1. 통합 테스트
첫 조회 후 redis -> 이후 DB 차감
만들어진 스레드 하나씩 redis를 조회하고 DB 에 재고를 차감함
이후 쿠폰 적재 예약 확정은 카프카를 통해 후처리 진행하고 있음
2. 성능 테스트 (ngrinder)
localhost로 테스트가 어려워서 ngrok을 설치하여 해당 ip로 우회하여 테스트 진행
@Test
public void test() {
// JSON 데이터 구성
Map<String, Object> histRequestDto = new HashMap<>();
histRequestDto.put("hnum", 1);
histRequestDto.put("userId", "user123");
histRequestDto.put("reqPpl", 2);
Map<String, Object> requestBody = new HashMap<>();
requestBody.put("histRequestDto", histRequestDto);
// JSON 변환
String jsonBody = new groovy.json.JsonBuilder(requestBody).toString();
// JSON 데이터 params에 추가
params.put("body", jsonBody);
HTTPResponse response = request.POST("https://ngrok에서 받은 ip/", params)
if (response.statusCode == 301 || response.statusCode == 302) {
grinder.logger.warn("Warning. The response may not be correct. The response code was {}.", response.statusCode)
} else {
assertThat(response.statusCode, is(200))
}
}
user 정보를 담은 Principal 객체를 security 단에서 부터 넘기 어려워 테스트용 user 정보를 db 에 집어넣고 실행함
99명의 유저가 동시에 접근
호스트의 재고를 10개로만 설정해두었기 때문에 금방 재고가 떨어져버려 throw exception을 던지기 때문에 테스트 결과는 유의미하지 않았음
카프카 레디스 설치 : 직선으로 떨어지는 표시는 테스트가 잘 안되고 있다는 의미….
ASIS 동기성을 고려하지 못한 코드
🏁 요청에 대해 재고가 제대로 차감이 되었는지 확인하는 통합테스트만 유의미
🥑아쉬운 점
단위테스트
의존 관계를 제거하여 서비스 단위로 Mock 객체를 생성하여 테스트 코드 작성하였으나, 주입 받는 객체가 계속해서 nullPointerException이 뜸
성능테스트
재고가 빠르게 없어지기 때문에 테스트의 의미가 크게 있진 않았음
성능이슈
분산 락을 사용하거나 캐시 서버를 덧붙였다는 것은 곧 스레드가 네트워크 통신에 대한 비용이 발생할 수 있다는 것이다.
하나의 DB 로만 연결할 때에는 공유 자원에만 락을 거는 방법도 고려해볼 수 있음
실무 로직 파악 必
❓ 현재 실무에서 재고 관리는 어떻게 되고 있을까?
동시에 요청이 들어왔을 때 update 시에 xLock 얻으므로 동시에 다른 스레드가 해당 데이터를
"읽고(xLock 가지고 있기 때문에 읽는 것도 대기, xLock 반납된 이후에 읽음)
-> 업데이트"
하는 로직이라면 문제가 없을까?
만약 DTO 에서 재고를 담아와서 그 값에서 요청값을 차감하는 로직이라면
Lost Update 문제 발생가능성이 있음
