과거에 개발했었던 코드를 리팩토링 및 업데이트 하는 과정에서, 기존 코드는 동시성 이슈가 발생할 수 있다는걸 발견했다.
@Transactional
public ResponseEntity<EntityResponseDto> executeService(Long id) {
return this.repository.findById(id).stream()
.filter(Entity::isNotOccupied)
.map(entity -> {
entity.setOccupied(true);
return entity;
})
.map(entity -> this.repository.save(entity))
.map( entity -> this.kafaService.produceMessage..)
//추가적인 비즈니스 로직 실행.
}예제 코드로 재현해보면 이런 느낌의 코드였는데, 의도했던 바는 Database 에서 조회해온 Domain Entity 의 Occupied 상태값이 False 일 경우에, 딱 한번만 메시지를 발행하기를 바라는 의도였다. 그런데, 이 코드를 리팩토링 하면서 동시성 처리를 위한 기능들이 빠져 있는것을 확인할 수 있었다. 즉 동시에 여러 요청이 오게 될 경우, Database 에 Commit 되기 이전에 다른 요청이 값을 읽어가면 의도한대로 작동하지 않는, 동시성 이슈가 발생하게 된다.
실제로 동시성 문제가 발생하는지 테스트코드를 통해 확인해보자.
@Test
@DisplayName("동시에 여러 요청이 오더라도 메시지는 하나만 발행된다.")
public void executeServiceFunctionWorksSuccessfully() throws InterruptedException{
//given
CountDownLatch latch = new CountDownLatch(numberOfThreads);
ExecutorService executorService = Executors.newFixedThreadPool(numberOfThreads);
//when
for (int i = 0; i < numberOfThreads; i++) {
executorService.submit(() -> {
try{
this.service.executeService(ID);
}finally {
latch.countDown();
}
});
}
latch.await();
//then
List<History> results = this.historyService.getKafkaHistories(ID);
Assertions.assertEquals(results.size(), 1);//Only one
}결과는 정확히 1번만 실행되지 않았고, 동시성 이슈가 발생하는것을 확인할 수 있었다.
동시성 문제는 이미 여러 해법들이 알려져 있고, 대표적으로 Lock 을 이용해 이 문제를 해결할 수 있다. 우선, 적용할 수 있는 몇가지 해결책을 확인해보면
1. synchronized 함수 이용
가장 떠오르고, 가장 쉽게 생각할 수 있는 방법은 함수를 synchronized 블럭으로 만들어 단일 스레드만 접근할 수 있게 하는 것이다.
// Example 1. synchronized 함수 이용하기.
@Transactional
public synchronized ResponseEntity<EntityResponseDto> executeService(Long id) {
return this.repository.findById(id).stream()
.filter(Entity::isNotOccupied)
.map(entity -> {
entity.setOccupied(true);
return entity;
})
.map(entity -> this.repository.save(entity))
.map( entity -> this.kafaService.produceMessage..)
//추가적인 비즈니스 로직 실행.
}이 해결 방법은 생각은 들었지만 실제로 구현하기에는 무리가 있다고 생각했다.
우선 첫번째로 서비스가 Autoscaling 되면, 단일 프로세스에서만 동시성을 제어하는 synchronized 는 제역할을 수행할 수 없고,
두번째는 @Transactional 과 synchronized 키워드는 함께 사용할 수 없다. AOP 프록시 객체는 synchronized 를 상속하지 않기 때문에 올바르게 synchronized 가 동작하게끔 보장하려면 프록시 객체를 생성하지 않도록 @Transactional 을 제거해야 한다.
물론 saveAndFlush() 를 이용하거나, 기타 여러 방법을 사용하면 어찌저찌 구현이야 되겠으나 근본적으로 첫번째 문제를 해결할 방법이 없기 때문에 이 방법은 고려할 가치가 없다.
2. Database 의 Lock 이용
Database 가 제공하는 Lock 을 이용하여 구현하는 방법이 있다.
@Getter
@Entity
@NoArgsConstructor(access = AccessLevel.PROTECTED)
public class LockTestEntity{
@Id
@GeneratedValue(strategy = GenerationType.IDENTITY)
private Long id;
@Column(name = "is_occupied")
private Boolean isOccupied;
//Optimistic lock 구현.
@Version
private Long version;
@JsonSerialize(using = LocalDateTimeSerializer.class)
@JsonDeserialize(using = LocalDateTimeDeserializer.class)
@Column(name = "created_at")
private LocalDateTime createdAt;
}//Pessimistic Lock 구현
@Lock(LockModeType.PESSIMISTIC_WRITE)
@Query("SELECT l FROM locktestentity l WHERE l.id = :id")
Optional<LockTestEntity> findById(@Param("id") Long id);다음과 같이 데이터베이스가 제공하는 낙관적 락과 비관적 락을 이용해서 문제를 해결할 수 있다.
현재 문제 상황에서는 Database 의 Read & Write 에서 발생하는 동시성 이슈이므로, 이 방법을 이용하면 해결할 수 있어 보인다.
하지만 이 코드는 Lock 의 처리를 Database 에 의존하고 있기 때문에 Database 와 관련 없는 비즈니스로직의 경우 또 다른 해결책을 강구해야 한다는 것과 NoSQL 을 사용하는 경우에는 또 다른 방법으로 접근해야 한다는 문제 때문에, 이 방법으로의 구현을 꺼렸다.
3. Redis 를 이용한 Distributed lock 구현
결국 이 방법으로 구현했는데, 우선 이미 Cache 를 위해 Redis 를 운영하고 있어 추가적으로 인프라를 구축할 필요가 없고, 앞서 설명한 Autoscaling 되는 환경에서도 동일하게 운영 가능하며 Database 에 의존하는 락 구현 방식이 아니기 때문에 NoSQL 이든, SQL 이든 상관없이 동일한 로직을 사용할 수 있다는 점에서 이 방식을 선택했다.
또한 검증되고 자세한 예제가 Kurly Tech Blog 에 있어서, 이를 응용하여 구현한다면 쉽게 구현할 수 있을것 같아, 이 방식을 채택하여 개발했다.
다음 글에서는 위 테크블로그를 참고하여 실제로 어떻게 구현했는지 작성해본다.