[Hype] AOP와 Redisson을 이용한 동시성 제어 코드 리팩토링

[Hype] AOP와 Redisson을 이용한 동시성 제어 코드 리팩토링

2023. 5. 24. 15:07ㆍ카테고리 없음

💡본문에 들어가기 앞서...
기존 구현 되어있던 로직 중에 게시글에 좋아요를 누르는 기능에서 여러 사람이 동시에 좋아요를 눌렀을 때 해당 게시글의 좋아요 수 정보에서 동시성 이슈가 발생할 수 있어서 이를 해결하기 위해 Redisson을 이용한 분산 락을 적용했습니다. 해당 코드를 리팩토링 하면 좋을 것 같아서 리팩토링 해보려고 합니다.

코드에 대한 설명

@Getter
@Entity
@NoArgsConstructor(access = AccessLevel.PROTECTED)
public class Post extends BaseEntity {

	@Id
	@GeneratedValue(strategy = GenerationType.IDENTITY)
	private Long id;

	@Embedded
	private Music music;

	@Lob
	private String content;

	private boolean isPossibleBattle;

	@Min(value = 0)
	private int likeCount;

	@ManyToOne(fetch = FetchType.LAZY)
	@JoinColumn(name = "member_id")
	private Member member;

	@OneToMany(mappedBy = "post")
	private final List<Like> likes = new ArrayList<>();

	@OneToMany(mappedBy = "challengedPost.post")
	private final List<Battle> challengedBattles = new ArrayList<>();

	@OneToMany(mappedBy = "challengingPost.post")
	private final List<Battle> challengingBattles = new ArrayList<>();

	public Post(Music music, String content, boolean isPossibleBattle, int likeCount, Member member) {
		checkArgument(likeCount >= 0, "좋아요 개수가 음수일 수 없습니다.", likeCount);
		this.music = music;
		this.content = content;
		this.isPossibleBattle = isPossibleBattle;
		this.likeCount = likeCount;
		this.member = member;
	}
}

먼저 Post 엔티티에서 likeCount를 넣었는데 좋아요 수를 증감시키는 로직은 많이 발생할 것 같다고 생각해서 이 부분을 필드로 추가해 DB와의 커넥션을 줄이고자 했습니다.

그래서 저 likeCount 필드에서 여러 접근이 동시에 일어날 수 있다고 생각했고 동시성 테스트를 수행해보면 아래와 같이 실패하는 것을 확인할 수있습니다.

	@BeforeEach
	void setUp() {
		member = memberRepository.save(createMember());
		post = postRepository.save(createPost(member));
	}

	@Test
	void 게시글_동시_좋아요100개_수행시_횟수만큼_증가한다() throws InterruptedException {
		// given
		int numberOfThreads = 100;
		List<Principal> principalList = new ArrayList<>();
		IntStream.rangeClosed(1, numberOfThreads)
			.forEach(i -> principalList.add(new TestAuthentication(
				new MemberDetails(String.valueOf(memberRepository.save(createMember()).getId())))));
		ExecutorService executorService = Executors.newFixedThreadPool(numberOfThreads);
		CountDownLatch countDownLatch = new CountDownLatch(numberOfThreads);

		// when
		for (int i = 0; i < numberOfThreads; i++) {
			final int I = i;
			executorService.submit(() -> {
				try {
					postService.likePost(principalList.get(I), post.getId());
				} finally {
					countDownLatch.countDown();
				}
			});
		}

		// then
		countDownLatch.await();
		Post findPost = postRepository.findById(post.getId())
			.orElseThrow();

		assertThat(findPost.getLikeCount()).isEqualTo(numberOfThreads);
	}

그래서 이것을 해결하기 위해서 Redisson을 이용하여 분산 락을 적용했었습니다.

기존 분산 락을 적용한 코드

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

	@Value("${spring.redis.port}")
	private int redisPort;

	private static final String REDISSON_HOST_PREFIX = "redis://";

	@Bean
	public RedissonClient redissonClient() {
		RedissonClient redisson;
		Config config = new Config();
		config.useSingleServer().setAddress(REDISSON_HOST_PREFIX + redisHost + ":" + redisPort);
		redisson = Redisson.create(config);
		return redisson;
	}
}

@Service
@RequiredArgsConstructor
public class PostLockFacade {

	private final PostService postService;
	private final PrincipalService principalService;
	private final RedissonClient redissonClient;

	public PostLikeResponseDto likePost(Principal principal, Long postId) {
		RLock lock = redissonClient.getLock(String.format("like:post:%d", postId));

		try {
			boolean available = lock.tryLock(10, 1, TimeUnit.SECONDS);
			if (!available) {
				throw new RuntimeException("Lock을 획득하지 못했습니다.");
			}
			Member member = principalService.getMemberByPrincipal(principal);
			return likePost(member, postId);
		} catch (InterruptedException error) {
			throw new RuntimeException(error);
		} finally {
			lock.unlock();
		}
	}

	public PostLikeResponseDto likePost(Member member, Long postId) {
		RLock lock = redissonClient.getLock(String.format("like:post:%d", postId));

		try {
			boolean available = lock.tryLock(10, 1, TimeUnit.SECONDS);
			if (!available) {
				throw new RuntimeException("Lock을 획득하지 못했습니다.");
			}
			return postService.likePost(member, postId);
		} catch (InterruptedException error) {
			throw new RuntimeException(error);
		} finally {
			lock.unlock();
		}
	}

}

Redis Configuration 파일을 만들어서 RedisClient를 빈으로 등록하고 이를 서비스 계층에서 적용하고자 했는데 try-catch 구문을 서비스 계층에서 처리하지 않기 위해서 Fasade라는 계층을 하나 더 넣어서 처리하고자 했습니다.

하지만 지금 보이는 코드와 같이 중복되는 코드가 많고 만약 다른 기능에서 분산 락이 필요하면 위와 같은 Fasade 계층을 추가하면서 중복된 코드가 똑같이 필요합니다. 그래서 이런 코드 중복을 없애기 위해서 리팩토링이 필요하다고 생각했고 이를 해결하기 위해서 AOP를 사용하자고 생각했습니다.

AOP를 사용하는 이유는 위 코드에서 중복되는 코드들은 비즈니스 로직에서의 관점이 아니라 DB 커넥션에 관련된 그런 로직이기에 관점을 분리하여서 좀 더 쉽게 분산 락을 적용하고자 하여 AOP를 사용해야겠다고 생각했습니다.

AOP를 이용한 Redisson 분산 락 구현 과정

Aspect를 적용할 포인트컷을 지정할 때 어노테이션을 이용할 예정이기에 먼저 어노테이션을 만들어 줄 것입니다. 기존 코드에서 lock을 생성할 때 waitTime, leaseTime, unit을 설정할 때 그냥 매직 넘버를 사용했기에 이를 어노테이션에서 관리하도록 구현하겠습니다.

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface DistributedLock {

	long waitTime() default 100L;

	long leaseTime() default 3L;

	TimeUnit unit() default TimeUnit.SECONDS;
}

그 후 Aspect를 구현하겠습니다

@Slf4j
@Aspect
@Component
@RequiredArgsConstructor
public class DistributedLockAspect {
	private final RedissonClient redissonClient;
	private final AopForTransaction aopForTransaction;

	@Around("@annotation(com.example.demo.common.aop.DistributedLock)")
	public Object lock(ProceedingJoinPoint joinPoint) throws Throwable {
		MethodSignature signature = (MethodSignature)joinPoint.getSignature();
		Method method = signature.getMethod();
		DistributedLock annotation = method.getAnnotation(DistributedLock.class);

		String postId = Stream.of(joinPoint.getArgs())
			.filter(arg -> arg instanceof Long)
			.map(Object::toString)
			.findFirst()
			.orElse(null);
		String lockName = String.format("%s:%s", method.getName(), postId);
		RLock rLock = redissonClient.getLock(lockName);

		try {
			boolean available = rLock.tryLock(annotation.waitTime(), annotation.leaseTime(), annotation.unit());
			if (!available) {
				return false;
			}

			return aopForTransaction.proceed(joinPoint);
		} catch (InterruptedException e) {
			log.error("Generate Interrupt!!");
			throw new InterruptedException();
		} finally {
			try {
				log.info("Redisson Lock Unlock Complete {} {}",
					method.getName(),
					lockName);
				rLock.unlock();
			} catch (IllegalMonitorStateException e) {
				log.info("Redisson Lock Already Unlock {} {}",
					method.getName(),
					lockName);
			}
		}
	}
}

위에서 구현한 어노테이션이 붙어있는 메서드에 대해서 해당 Aspect를 적용하게 됩니다. Redisson의 장점이 락을 얻는 인터페이스를 구현해놓아서 따로 retry하는 로직을 구현하지 않아도 됩니다. 내부적으로 처리해줍니다. 위에서 언급했던 waitTime은 총 몇 초까지 재시도를 해볼 것인가에 대한 시간이고 leaseTime은 락을 가지고 있을 수 있는 최대 시간을 의미합니다. 마지막으로 unit은 시간의 단위를 뜻합니다.

만약 lockName을 통해서 락을 얻으려고 시도를 했는데 실패한다면 available = false 값이 나옵니다. 그래서 내부적으로 재시도를 하게 됩니다. 락을 얻었다면 joinPoint의 proceed를 실행하게 됩니다.

finally에서 lock을 해제하게 되는데 이미 lock을 해제 한 경우에 해당 catch 문에 잡힙니다. 이는 시간이 초과되어서 재시도를 안하고 그냥 빠져나가는 상황을 의미합니다.

위에서 설명하지 않은 AopForTransaction 클래스에 대해서 설명하겠습니다. 일반적으로 Aspect에서 Around 어드바이스를 이용할 때는 joinPoint.proceed() 이런식으로 타겟 객체의 로직을 바로 수행하는데 여기선 AopForTransaction 클래스를 거쳤습니다.

@Slf4j
@Component
public class AopForTransaction {

	@Transactional(propagation = Propagation.REQUIRES_NEW)
	public Object proceed(ProceedingJoinPoint joinPoint) throws Throwable {
		return joinPoint.proceed();
	}
}

그 이유는 락과 트랜잭션의 할당 및 해제 타이밍이 중요해서 입니다.

락을 건다는 것은 특정 로우로의 접근을 막는다라고 생각하면 될 것입니다. 그럼 하나의 트랜잭션만이 접근할 수 있게 됩니다. 근데 트랜잭션을 커밋하기 전에 만약 락을 해제한다고 생각하면 업데이트된 정보를 DB에 반영하기 전에 다른 트랜잭션이 해당 정보를 조회할 수 있다는 의미이고 그럼 정합성이 어긋나게 됩니다.

그래서 여기서는 락이 생성되고 트랜잭션을 시작하고 트랜잭션에서의 업데이트 정보가 반영되고 락을 해제 하기 위해서 AopForTransaction이라는 클래스를 사용합니다.

@Service
@RequiredArgsConstructor
@Transactional(readOnly = true)
public class PostService {

	private final PostRepository postRepository;
	private final LikeRepository likeRepository;
	private final PrincipalService principalService;

	@DistributedLock
	public PostLikeResponseDto likePost(Principal principal, Long postId) {
		Member member = principalService.getMemberByPrincipal(principal);
		Post post = postRepository.findById(postId)
			.orElseThrow(() -> new EntityNotFoundException(NOT_FOUND_POST.getMessage()));

		boolean isExist = likeRepository.existsByMemberAndPost(member, post);

		if (isExist) {
			likeRepository.deleteByMemberAndPost(member, post);
			post.minusLike();
			return PostLikeResponseDto.of(false);
		} else {
			likeRepository.save(new Like(post, member));
			post.plusLike();
			return PostLikeResponseDto.of(true);
		}
	}
}

그 후 해당 어노테이션을 사용하고자 하는 메서드에 달아줌으로써 분산 락을 적용할 수 있습니다.

트러블 슈팅

Deadlock의 발생

위에서 보여준 서비스 계층 코드를 다시 살펴봅시다.

@Service
@RequiredArgsConstructor
@Transactional(readOnly = true)
public class PostService {

	private final PostRepository postRepository;
	private final LikeRepository likeRepository;
	private final PrincipalService principalService;

	@DistributedLock
	public PostLikeResponseDto likePost(Principal principal, Long postId) {
		Member member = principalService.getMemberByPrincipal(principal);
		Post post = postRepository.findById(postId)
			.orElseThrow(() -> new EntityNotFoundException(NOT_FOUND_POST.getMessage()));

		boolean isExist = likeRepository.existsByMemberAndPost(member, post);

		if (isExist) {
			likeRepository.deleteByMemberAndPost(member, post);
			post.minusLike();
			return PostLikeResponseDto.of(false);
		} else {
			likeRepository.save(new Like(post, member));
			post.plusLike();
			return PostLikeResponseDto.of(true);
		}
	}
}

여기엔 클래스자체에 Transactional(readOnly = true)가 달려있습니다. 개인적으로 조회 시의 성능을 향상 준다고 알고 있어서클래스에 붙여주는 편입니다.. 근데 이것 때문에 DeadLock이 발생했습니다.

	@BeforeEach
	void setUp() {
		member = memberRepository.save(createMember());
		post = postRepository.save(createPost(member));
	}

	@Test
	void 게시글_동시_좋아요100개_수행시_횟수만큼_증가한다() throws InterruptedException {
		// given
		int numberOfThreads = 100;
		List<Principal> principalList = new ArrayList<>();
		IntStream.rangeClosed(1, numberOfThreads)
			.forEach(i -> principalList.add(new TestAuthentication(
				new MemberDetails(String.valueOf(memberRepository.save(createMember()).getId())))));
		ExecutorService executorService = Executors.newFixedThreadPool(numberOfThreads);
		CountDownLatch countDownLatch = new CountDownLatch(numberOfThreads);

		// when
		for (int i = 0; i < numberOfThreads; i++) {
			final int I = i;
			executorService.submit(() -> {
				try {
					postService.likePost(principalList.get(I), post.getId());
				} finally {
					countDownLatch.countDown();
				}
			});
		}

		// then
		countDownLatch.await();
		Post findPost = postRepository.findById(post.getId())
			.orElseThrow();

		assertThat(findPost.getLikeCount()).isEqualTo(numberOfThreads);
	}

20230524 07:13:51.188 [pool-4-thread-3] WARN  [SqlExceptionHelper:137] : SQL Error: 1213, SQLState: 40001 
20230524 07:13:51.189 [pool-4-thread-3] ERROR [SqlExceptionHelper:142] : Deadlock found when trying to get lock; try restarting transaction 
20230524 07:13:51.192 [pool-4-thread-3] INFO  [AbstractBatchImpl:213] : HHH000010: On release of batch it still contained JDBC statements

어떻게 동작하냐면 클래스 수준에 달린 @Transactional(readOnly = true)에서 먼저 읽기 락을 겁니다. 그 후에 RedissonClient를통해서 분산 락이 걸리게 되어서 읽기 락이 끝날 때까지 분산 락이 기다리고 있는데 읽기 락은 해당 로직이 다 끝나야 해제되니까 DeadLock이 발생하는 겁니다.

그렇다면 기존에 클래스 수준에 달린 @Transactional(readOnly = true)와 메서드 수준에서의 @Transactional은 어떻게 공존할 수 있을까요? @Trasactional을 달면 읽기-쓰기 트랜잭션이 걸리는데 클래스 수준에서 상속된 읽기 전용 동작을 재정의하여서 읽기-쓰기 트랜잭션이 실행됩니다. 그리고 필요할 때만 쓰기 락을 걸어줘서 교착 상태를 유발하지 않고 공존할 수 있습니다.

그럼 분산 락은 공존할 수 없나요? 없습니다. 일반적으로 분산 락을 사용하는 이유가 여러 애플리케이션 서버에서 DB로 접근할 때 배타적 엑세스를 보장해야 하는데 읽기 락과 쓰기 락이 공존하도록 허용한다면 정합성이 깨지고 충돌이 발생할수 있습니다.

그래서 이 문제를 해결하는 방법이 2개가 있는데

PostService에서 @Transactional(readOnly = true)를 제거한다.
하나의 컴포넌트나 상위 계층을 만들어서 @DistributedLock 어노테이션을 거기에 붙이고 기존 로직을 수행하는 것이다. (위에서 구현한 AopForTransaction 처럼)

저는 1번 방법을 이용했습니다. 그 이유는 또 다른 컴포넌트를 만든다는 것이 관리 포인트가 늘어나고 또 특정 컴포넌트를 PostService에서 사용하기 위해선 파라미터로 설정을 해야 하는데 퍼블릭 인터페이스가 바뀌는 것이기에 수정 포인트가 많을 것 입니다. 이런 것들과 Transactional(readOnly = true)의 이점인 스냅샷을 유지 하지 않고 플러시가 이루어지지 않아 성능을 최적화한다를 비교했을 때 매력적이라고 느껴지지 않아서 1번 방법을 선택했습니다.

waitTime 설정

테스트 코드를 만들 때 몇 개의 스레드를 이용해서 동시성 테스트할 것인지 결정하게 되는데 이 때 waitTime에 따라서 실패할 수 있습니다. 그 이유는 분산 락은 pub/sub을 이용해서 retry를 하지만 스핀 락관 다르게 무한정 retry하지 않기에 waitTime을 적절히 설정하지 않으면 시간이 초과되어서 업데이트 되지 않는 요청들이 생깁니다.

테스트 결과

	@Test
	void 게시글_동시_좋아요100개_수행시_횟수만큼_증가한다() throws InterruptedException {
		// given
		int numberOfThreads = 100;
		List<Principal> principalList = new ArrayList<>();
		IntStream.rangeClosed(1, numberOfThreads)
			.forEach(i -> principalList.add(new TestAuthentication(
				new MemberDetails(String.valueOf(memberRepository.save(createMember()).getId())))));
		ExecutorService executorService = Executors.newFixedThreadPool(numberOfThreads);
		CountDownLatch countDownLatch = new CountDownLatch(numberOfThreads);

		// when
		for (int i = 0; i < numberOfThreads; i++) {
			final int I = i;
			// postService.likePost(principalList.get(I), post.getId());
			executorService.submit(() -> {
				try {
					postService.likePost(principalList.get(I), post.getId());
				} finally {
					countDownLatch.countDown();
				}
			});
		}

		// then
		countDownLatch.await();
		Post findPost = postRepository.findById(post.getId())
			.orElseThrow();

		assertThat(findPost.getLikeCount()).isEqualTo(numberOfThreads);
	}

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface DistributedLock {
	String lockName();

	long waitTime() default 10L;

	long leaseTime() default 3L;

	TimeUnit unit() default TimeUnit.SECONDS;
}

1000개를 할려면 waitTime을 100초 정도로 늘려야했습니다.

마무리

AOP를 로깅할 때만 사용해보고 다른 곳에서 어떤 식으로 사용하면 좋을 지 크게 느낀 적이 없었는데 이를 구현해보면서 정말 잘 쓰면 엄청 유용하겠구나를 느꼈습니다. AOP를 깊게 공부해서 더 잘 사용할 수 있도록 노력 해봐야겠습니다.

Reference

Redisson 라이브러리를 이용한 분산 락 동시성 처리 (1)

이왕할꺼 제대로 하자.