프로젝트 개발 중 도로명 주소를 위경도로 변환하는 작업을 카카오 geo api를 통해 해결하고 있었다.
처리 속도 문제로 비동기 통신(이전 글)을 적용했는데 카카오에서 사용량 제한을 분당 10만 요청으로 걸어놨기 때문에 429 too many requests 예외가 발생했었다.
분당 10만이라는 것이 횟수 제한뿐만 아니라 속도 제한도 있어서 추정이지만 1ms당 약 1.67번 이상의 요청을 보낸다면 429가 발생한다.

통신 간 delay 주기

처음에 시도한 방법은 Flux에 .delayElements() 라는 요소 사이에 delay를 줄 수 있는 메소드가 있었고 delay 1ms 를 줘서 해결하려고 했다.

하지만 정말 바보 같은 생각이었다.
왜냐하면 멀티 스레드 환경에서 메소드 실행은 독립적으로 수행하기 때문에 스레드 당 Flux가 하나씩 생성된다.
위 이미지처럼 동시에 요청을 보내면 각 Flux에서 요소들 사이는 delay가 있지만 여러 Flux가 동시에 비동기 통신을 요청하기 때문에 429를 막을 수 없었다.

스레드 동기화

synchronized

다수의 사용자가 동시에 요청을 보낼 경우 문제가 발생하기 때문에 메소드에 synchronized 키워드로 동기화 하려고 했다.
하지만 비동기 통신을 처리하는 스레드는 별도로 작업 스레드에서 처리하고 메소드를 실행한 스레드는 스레드풀에서 가져왔기 때문에 비동기 통신이 진행 중인데도 불구하고 바로 락을 반납하게 된다. 따라서 여러 스레드가 동시에 비동기 통신을 하는 것과 다름이 없어서 synchronized 로 해결할 수 없었다.

Semaphore

세마포어(Semaphore)는 여러 프로세스나 스레드가 공유 자원에 접근하는 것을 제어하여 동시성 문제를 해결하는 것으로 자바에서의 세마포어는 java.util.concurrent 패키지의 Semaphore 클래스를 통해 구현된다.
접근 가능할 수를 지정하여 지정된 수만큼 여러 프로세스나 스레드가 공유 자원에 접근할 수 있게 된다.

private final Semaphore semaphore = new Semaphore(1);

현재 프로젝트에서는 비동기 통신을 하나의 스레드만 수행해야 되기 때문에
접근 가능 수를 1 로 지정하였다.
그리고 락을 얻기 위해 tryAcquire() 메소드를 사용하여 지정한 시간에 락을 못 얻을 경우 예외가 발생하도록 하였다.

사용자의 오랜 기다림

429 문제는 해결했지만 동기화로 인해 여러 사용자들이 동시에 요청을 보낼 경우 다른 사용자들은 락을 얻기까지 계속 기다려야 되는 문제가 있다.
이렇게 되면 스레드를 오랜 시간 점유하기 때문에 비효율적이다.

따라서 사용자의 오랜 기다림을 방지할 수 있는 장치가 필요하였다.

생각한 방법은 현재 처리해야 될 비동기 통신 작업 수를 저장하고 작업 수가 많을 경우 사용자에게 잠시 후 이용하라고 알려주면 되지 않을까?

AtomicInteger

현재 처리해야 될 비동기 통신 작업 수를 저장하기 위해 AtomicInteger 를 사용하였다.
AtomicInteger는 멀티 스레드 환경에서 동시성과 가시성을 해결한 Integer 타입이다.
여러 스레드가 동시에 같은 변수를 접근해도 문제가 없다.
자세한 내용은 AtomicInteger란? 을 참고하면 될 것 같다.

현재 적용한 프로젝트에서는 AtomicInteger 로 현재 처리해야 되는 카카오 geocode API 요청 수를 관리하는데 다음과 같은 작업을 수행한다.

• 사용자의 엑셀을 통한 장소 등록 요청하기 전에 빠른 서비스 이용 가능 여부를 판단하기 위해 AtomicInteger 값을 확인한다.
  • 값이 크다면 오래 기다려야 하므로 "잠시 후 이용" 알림창을 띄운다.
  • 값이 작다면 빠른 시간에 처리가 가능하므로 AtomicInteger 에 엑셀에 저장된 장소 수만큼 통신 작업 수를 증가시킨다.
    • 통신 작업이 모두 끝나면 AtomicInteger 에 통신 작업 수를 감소시킨다.

최종

클래스 다이어그램은 위와 같다.
WebClientController와 Geocodr가 프록시인 이유는 부가 기능들을 분리했기 때문이다.

그렇다면 요청이 들어올 때 어떻게 수행하는지 보자.
우선 맨 처음 요청이 들어오면 오래 기다리지 않고 서비스를 이용할 수 있는지 확인한다.

1 - AtomicInteger 증가
	2 - 세마포어 증가
		3 - 위경도 변환
	2 - 세마포어 감소
1 - AtomicInteger 감소

그런 다음 서비스를 이용할 수 있다면 위와 같은 순서로 로직을 수행하게 된다.

저런 구조로 된 이유는 한 클래스에 여러 책임이 있어서 세마포어 관리와 통신 작업 수 관리를 각각 AOP로 분리하여 위경도 변환 이라는 핵심 기능에 부가 기능으로 추가하였다.

참조

https://www.baeldung.com/java-semaphore
https://javaplant.tistory.com/23