[Spring] TaskScheduler 이론과 구현

0. 개요

외부 API에 보낸 요청의 상태를 모니터링하는 기능을 개발하던 도중, 기존의 polling 방식이 서버에 불필요한 부하를 만들고 있었음을 파악했다. Polling 했던 목적이 클라이언트(관리자)가 스케줄러를 통한 API 요청에 대한 응답이 잘 오고 있는지 확인하기 위함이었다. 요청 이후 30초안에 응답이 오지 않으면 '지연', 50초 안에 응답이 오지 않으면 '초과'임을 클라이언트에게 전달해주어야 했기에 이를 위해 최대한 빠르게 문제 상황을 전달해 주기 위해서는 짧은 주기의 polling이 요구되었다.
 
하지만, 이벤트 발생 시에만 클라이언트에게 정보를 전달해주면 되었기에 SSE를 고려하게 되었고, 서버 측에서 정해진 특정 시점에만 확인 후 이벤트를 발생시키면 되었기에 방법을 찾아보았다. TaskScheduler로 key별 스케줄을 등록해 사용자에게 더욱 정확한 상태 데이터를 전달할 수 있는 방법을 알게되었고, 이를 적용해 결국 서버 부하가 10배 감소했고, 상태 정확도는 약 2배 향상되었다. 

---

TaskScheduler는 생소하지만, TaskExecutor는 많이 들어봤을 것 같다. TaskExecutor는 작업의 비동기 실행을 위한 인터페이스이고, TaskScheduler는 작업 예약(스케줄링)을 위한 인터페이스다. 해당 글에서는 TaskScheduler에 대해 다뤄보겠다. 
 

1. TaskScheduler란? (공식문서 참고)

Spring has a TaskScheduler SPI with a variety of methods for scheduling tasks to run at some point in the future.

공식문서를 보면 알 수 있다싶이, TaskScheduler는 미래의 특정 시점에 작업을 실행하기 위한 다양한 메서드를 제공한다. 

public interface TaskScheduler {

	Clock getClock();

	ScheduledFuture schedule(Runnable task, Trigger trigger);

	ScheduledFuture schedule(Runnable task, Instant startTime);

	ScheduledFuture scheduleAtFixedRate(Runnable task, Instant startTime, Duration period);

	ScheduledFuture scheduleAtFixedRate(Runnable task, Duration period);

	ScheduledFuture scheduleWithFixedDelay(Runnable task, Instant startTime, Duration delay);

	ScheduledFuture scheduleWithFixedDelay(Runnable task, Duration delay);

위과 같은 여러 메서드들을 제공하는데, 크게 세 가지 유형으로 나뉜다. 
 

• schedule(Runnable task, Instant startTime): 지정된 시간(startTime)에 딱 한 번만 실행
• scheduleAtFixedRate(...); 작업 완료 여부와 상관없이 설정된 간격(period)마다 작업을 시작
• scheduleWithFixedDelay(...): 이전 작업이 끝난 시점부터 일정 시간(delay) 후에 다음 작업을 시작

Trigger를 인자로 받는 메서드는 작업이 끝날 때마다 trigger.nextExecution(context)를 호출해서 다음 실행 시각은 계산한다. 
다음 실행 시간을 결정할 때, 이전 실행 결과를 고려해야 한다면, 해당 정보를 TriggerContext 내에서 확인할 수 있다. 

public interface Trigger {
    Instant nextExecution(TriggerContext triggerContext);
}

 
Trigger 인터페이스를 통해 triggerContext(이력 정보)를 받아서 다음 실행 시점(Instant)를 반환하는 것이다. TriggerContext 안에는 다음과 같은 정보들이 있다.  
 

• lastScheduledExecution(): 마지막 작업이 원래 실행되기로 예약되었던 시각
• lastActualExecution(): 마지막 작업이 실제로 시작된 시각
• lastCompletion(): 마지막 작업이 종료된 시각

이처럼 필요한 모든 데이터를 캡슐화하고 있으며, 필요하다면 향후 확장이 가능하도록 설계되었다. 
 
Spring은 Trigger 인터페이스의 두 가지 구현체를 제공한다.
- CronTrigger는 크론 표현식을 기반으로 시점(날짜, 시간 등)을 명확히 제시한다. 
 
- PeriodicTrigger는 고정된 주기(period), 선택적인 초기 지연 시간(initial delay), 그리고 이 주기를 '고정률(fixed-rate)'로 볼지 '고정 지연(fixed-delay)'으로 볼지 결정하는 불리언(boolean) 값을 인자로 받아 정해진 주기로 시점을 계산해 실행한다. 
 
이미 여러 유용한 메서드들이 있는데, 설정 파일이나 외부 조건에 따라 크론 방식과 주기적 방식을 자유롭게 갈아끼워야 할 때나 이력 정보로 실행 시점을 조정해야 할 때에는 Trigger 객체를 주입받아 인자로 받는 메서드를 사용할 수 있을 것 같다. 
 

2. ScheduledFuture schedule(Runnable task, Instant startTime) 적용기

하지만, 나의 경우에는 이러한 부가 기능은 필요가 없었고, 특정 이벤트가 발생한 시점을 기준으로 지정된 시점에 딱 한 번 설정한 작업을 실행하면 되었기에 ScheduledFuture schedule(Runnable task, Instant startTime)을 선택했다. *Runnable은 실행 작업이며, Instant는 실행 시점이다. 

@Slf4j
@Component
@RequiredArgsConstructor
public class PhisFeignScheduler {
	private final PhisMonitorCache phisMonitorCache;
	private final ApplicationEventPublisher publisher;
	private final TaskScheduler scheduler;

	// 요청마다 Runnable이 등록되고, Runnable은 각각 스케줄러 내부 스레드 풀에서 병렬로 실행
	public void scheduleMonitoring(String key) {
		// 30초 후
		scheduler.schedule(() -> checkAndPublish(key), Instant.now().plusMillis(30000));
		// 50초 후
		scheduler.schedule(() -> checkAndPublish(key), Instant.now().plusMillis(50000));
	}

	private void checkAndPublish(String key) {
		PhisMonitorCache.RecentStatus recentStatus = phisMonitorCache.recentStatusMap.get(key);
		if (recentStatus == null || !recentStatus.getIsResponding()) // 요청이 진행중일 때만 이벤트 발행
			return;

		// SCHEDULER 이벤트 발행
		publisher.publishEvent(
			FeignResponseEventDto.builder()
				.eventType(EventType.SCHEDULER)
				.hospitalCode(key.split(":")[0])
				.phisApiCategory(key.split(":")[1])
				.responseTime(LocalDateTime.now()) // 스케쥴러 이벤트 발생 시각
				.responseStatus(null)
				.build()
		);
	}
}

 
스케줄러로 등록된 api 요청시 TaskScheduler로 작업 예약하고, 정해진 Instant 시점에 이벤트를 발생시키는 전 과정을 도식화해보면 다음과 같다. 

 
 

3. ThreadPoolTaskScheduler의 ThreadPool

해당 메서드가 실행될 때마다 TaskScheduler 내부적으로 Runnable(작업)을 thread pool에서 사용 가능한 스레드를 할당하고, 실행을 예약한다. 만일 스레드 풀에 여유가 없다면, 대기하다가 스레드 풀에서 스레드가 확보되면 실행한다. 
 
schedule() 메서드를 호출하는 순간, 스케줄러 내부의 지연 큐(DelayedWorkQueue)에 작업이 등록된다. 이후, 설정한 시간(Instant)이 되면, 스케줄러(ex. ThreadPoolTaskScheduler)의 Thread Pool에서 유휴 스레드 하나를 깨워 작업을 수행한다. 만일 스레드 풀의 모든 스레드가 작업 중이라면, 앞선 작업이 끝나서 스레드가 반환될 때까지 대기(Queueing) 후 실행된다. 
https://docs.oracle.com/en/java/javase/17/docs/api/java.base/java/util/concurrent/ScheduledThreadPoolExecutor.html

ScheduledThreadPoolExecutor

schedule(), @Scheduled로 예약된 작업을 처리하는 건 TaskScheduler 스레드 풀인데, Spring Boot에서 별도의 설정을 하지 않으면 ThreadPoolTaskScheduler의 기본 poolSize는 1이다. 그런데, 해당 프로젝트에서 cache warm up, data retention, soft delete와 hard delete, 1분 주기 api 요청 등 여러 스케줄러 작업이 있었고, 실행 시간이 오래 걸리는 작업도 있었기에 스레드 1개로는 정확한 시점의 작업이 어려워지거나 작업 종료 시각을 예측하기 힘들 수 있겠다 판단하여 동시간 대의 요청을 문제 없이 처리할 수 있는 최소한의 스레드 개수인 4로 poolSize를 늘려 설정했다. 

@Bean
public TaskScheduler taskScheduler() {
    ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
    scheduler.setPoolSize(4);
    scheduler.setThreadNamePrefix("scheduling-");
    scheduler.setWaitForTasksToCompleteOnShutdown(true); // 종료 시 진행 중 작업 기다림 (graceful shutdown)
    scheduler.setAwaitTerminationSeconds(30); // 최대 30초 대기
    scheduler.initialize(); // 스레드 풀 생성과 초기화
    scheduler.setRemoveOnCancelPolicy(true); // 취소된 작업 즉시 제거
    return scheduler;
}

ThreadPoolTaskScheduler

 

4. SSE 작업 내용 

내용이 길어 한 포스팅에 전부 담지 못했다.
SSE 관련 내용은 https://persi0815.tistory.com/162, Redis Pub/Sub 관련 내용은 https://persi0815.tistory.com/163에 적어두었다.