[FEAT] 커스텀 스케줄러 구현 (#160)

## 커스텀 스케줄러 구현 - 런타임중에 동적으로 스케줄 변경이 가능한 커스텀 스케줄러 구현 - 하나의 작업에 크론 여러개를 등록하고 싶을 시 "|" 를 구분자로 등록 가능 - DynamicScheduled(name = "", cron = "") 어노테이션으로 스케줄링 등록 - /admin/scheduler 페이지에서 관리 가능 --------- Co-authored-by: Photogrammer <[email protected]>
Kernel360 · Aug 30, 2024 · d74480e · d74480e
1 parent d69c0a9
commit d74480e
Show file tree

Hide file tree

Showing 11 changed files with 316 additions and 38 deletions.
diff --git a/server/src/main/java/com/talkka/server/admin/controller/AdminController.java b/server/src/main/java/com/talkka/server/admin/controller/AdminController.java
@@ -1,10 +1,13 @@
 package com.talkka.server.admin.controller;
 
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Lazy;
 import org.springframework.stereotype.Controller;
 import org.springframework.ui.Model;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.RequestMapping;
 
+import com.talkka.server.admin.scheduler.DynamicSchedulingConfig;
 import com.talkka.server.admin.service.AdminService;
 import com.talkka.server.admin.service.CollectBusRouteService;
 import com.talkka.server.admin.service.PublicApiKeyService;
@@ -20,7 +23,9 @@ public class AdminController {
 	private final UserService userService;
 	private final CollectBusRouteService collectBusRouteService;
 	private final PublicApiKeyService publicApiKeyService;
-	// private final DynamicSchedulingConfig dynamicSchedulingConfig;
+	@Lazy
+	@Autowired
+	private DynamicSchedulingConfig dynamicSchedulingConfig;
 
 	@GetMapping("")
 	public String index() {
@@ -56,11 +61,11 @@ public String publicApiKey(Model model) {
 		model.addAttribute("apiKeys", publicApiKeyService.getKeyList());
 		return "admin/key";
 	}
-	//
-	// @GetMapping("/scheduler")
-	// public String scheduler(Model model) {
-	// 	var schedulers = dynamicSchedulingConfig.getSchedulers();
-	// 	model.addAttribute("schedulers", schedulers);
-	// 	return "admin/scheduler";
-	// }
+
+	@GetMapping("/scheduler")
+	public String scheduler(Model model) {
+		var schedulers = dynamicSchedulingConfig.getSchedulers();
+		model.addAttribute("schedulers", schedulers);
+		return "admin/scheduler";
+	}
 }
diff --git a/server/src/main/java/com/talkka/server/admin/controller/SchedulerController.java b/server/src/main/java/com/talkka/server/admin/controller/SchedulerController.java
@@ -0,0 +1,38 @@
+package com.talkka.server.admin.controller;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Lazy;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import com.talkka.server.admin.dto.SchedulerReqDto;
+import com.talkka.server.admin.exception.SchedulerNotFoundException;
+import com.talkka.server.admin.scheduler.DynamicSchedulingConfig;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+@RestController
+@RequiredArgsConstructor
+@RequestMapping("/admin/api/scheduler")
+@Slf4j
+public class SchedulerController {
+	@Lazy
+	@Autowired
+	private DynamicSchedulingConfig dynamicSchedulingConfig;
+
+	// cron 유효성 검사 로직 필요
+	@PostMapping("")
+	public ResponseEntity<?> updateScheduler(@RequestBody SchedulerReqDto dto) {
+		try {
+			dynamicSchedulingConfig.updateCronExpression(dto);
+		} catch (SchedulerNotFoundException exception) {
+			log.error(exception.getMessage());
+			return ResponseEntity.badRequest().body(exception.getMessage());
+		}
+		return ResponseEntity.ok().build();
+	}
+}
diff --git a/server/src/main/java/com/talkka/server/admin/dto/SchedulerReqDto.java b/server/src/main/java/com/talkka/server/admin/dto/SchedulerReqDto.java
@@ -0,0 +1,7 @@
+package com.talkka.server.admin.dto;
+
+public record SchedulerReqDto(
+	String name,
+	String cronString
+) {
+}
diff --git a/server/src/main/java/com/talkka/server/admin/dto/SchedulerRespDto.java b/server/src/main/java/com/talkka/server/admin/dto/SchedulerRespDto.java
@@ -0,0 +1,12 @@
+package com.talkka.server.admin.dto;
+
+import java.util.List;
+
+public record SchedulerRespDto(
+	String name,
+	String cronString
+) {
+	public SchedulerRespDto(String name, List<String> cronList) {
+		this(name, String.join(" | ", cronList));
+	}
+}
diff --git a/server/src/main/java/com/talkka/server/admin/exception/SchedulerNotFoundException.java b/server/src/main/java/com/talkka/server/admin/exception/SchedulerNotFoundException.java
@@ -0,0 +1,9 @@
+package com.talkka.server.admin.exception;
+
+public class SchedulerNotFoundException extends RuntimeException {
+	static final String message = "Scheduler not found";
+
+	public SchedulerNotFoundException() {
+		super(message);
+	}
+}
diff --git a/server/src/main/java/com/talkka/server/admin/scheduler/DynamicScheduled.java b/server/src/main/java/com/talkka/server/admin/scheduler/DynamicScheduled.java
@@ -0,0 +1,14 @@
+package com.talkka.server.admin.scheduler;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@Retention(RetentionPolicy.RUNTIME)
+@Target(ElementType.METHOD)
+public @interface DynamicScheduled {
+	String cron() default "0 0/1 * * * *";  // 기본 cron 표현식
+
+	String name();  // 작업 이름
+}
diff --git a/server/src/main/java/com/talkka/server/admin/scheduler/DynamicSchedulingConfig.java b/server/src/main/java/com/talkka/server/admin/scheduler/DynamicSchedulingConfig.java
@@ -0,0 +1,98 @@
+package com.talkka.server.admin.scheduler;
+
+import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ScheduledFuture;
+
+import org.springframework.context.ApplicationContext;
+import org.springframework.scheduling.annotation.EnableScheduling;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
+import org.springframework.scheduling.config.ScheduledTaskRegistrar;
+import org.springframework.scheduling.support.CronTrigger;
+import org.springframework.stereotype.Component;
+import org.springframework.util.LinkedMultiValueMap;
+import org.springframework.util.MultiValueMap;
+import org.springframework.util.ReflectionUtils;
+
+import com.talkka.server.admin.dto.SchedulerReqDto;
+import com.talkka.server.admin.dto.SchedulerRespDto;
+import com.talkka.server.admin.exception.SchedulerNotFoundException;
+
+import jakarta.annotation.PostConstruct;
+import lombok.RequiredArgsConstructor;
+
+@EnableScheduling
+@Component
+@RequiredArgsConstructor
+// 나중에 Quartz Scheduler 로 리팩토링
+public class DynamicSchedulingConfig {
+
+	private ScheduledTaskRegistrar registrar;
+	private final ApplicationContext applicationContext;
+	private final Map<String, Runnable> taskMap = new HashMap<>();
+	private final MultiValueMap<String, ScheduledFuture<?>> scheduledTasks = new LinkedMultiValueMap<>();
+	private final MultiValueMap<String, String> cronMap = new LinkedMultiValueMap<>();
+
+	@PostConstruct
+	public void init() {
+		registrar = new ScheduledTaskRegistrar();
+		ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
+		taskScheduler.setPoolSize(20);
+		taskScheduler.initialize();
+		registrar.setTaskScheduler(taskScheduler);
+
+		String[] beanNames = applicationContext.getBeanDefinitionNames();
+		for (String beanName : beanNames) {
+			if (beanName.equals("dynamicSchedulingConfig")) {
+				continue;
+			}
+			Object bean = applicationContext.getBean(beanName);
+			Method[] methods = bean.getClass().getDeclaredMethods();
+
+			for (Method method : methods) {
+				DynamicScheduled annotation = method.getAnnotation(DynamicScheduled.class);
+				if (annotation != null) {
+					String[] crons = annotation.cron().split("\\|");
+					String name = annotation.name();
+					Runnable task = () -> ReflectionUtils.invokeMethod(method, bean);
+					taskMap.put(name, task);
+
+					for (String cron : crons) {
+						cron.trim();
+						ScheduledFuture<?> future = registrar.getScheduler()
+							.schedule(task, triggerContext -> new CronTrigger(cron).nextExecution(triggerContext));
+						scheduledTasks.add(name, future);
+						cronMap.add(name, cron);
+					}
+				}
+			}
+		}
+	}
+
+	public List<SchedulerRespDto> getSchedulers() {
+		List<SchedulerRespDto> schedulers = new ArrayList<>();
+		for (String name : taskMap.keySet()) {
+			schedulers.add(new SchedulerRespDto(name, cronMap.get(name)));
+		}
+		return schedulers;
+	}
+
+	public void updateCronExpression(SchedulerReqDto dto) throws SchedulerNotFoundException {
+		// 현재 해당 메소드의 스케줄링 작업 전부 멈추고 맵에서 삭제
+		scheduledTasks.get(dto.name()).forEach(future -> future.cancel(false));
+		scheduledTasks.remove(dto.name());
+		// 해당 메소드 가져옴
+		Runnable task = taskMap.get(dto.name());
+		// 크론식마다 스케줄 작업 추가
+		for (String cron : dto.cronString().split("\\|")) {
+			cron.trim();
+			ScheduledFuture<?> future = registrar.getScheduler()
+				.schedule(task, triggerContext -> new CronTrigger(cron).nextExecution(triggerContext));
+			scheduledTasks.add(dto.name(), future);
+			cronMap.add(dto.name(), cron);
+		}
+	}
+}
diff --git a/server/src/main/java/com/talkka/server/api/datagg/config/PersistenceApiKeyProvider.java b/server/src/main/java/com/talkka/server/api/datagg/config/PersistenceApiKeyProvider.java
@@ -6,11 +6,11 @@
 import java.util.TreeMap;
 
 import org.springframework.context.annotation.Primary;
-import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 
 import com.talkka.server.admin.dao.PublicApiKeyEntity;
 import com.talkka.server.admin.dao.PublicApiKeyRepository;
+import com.talkka.server.admin.scheduler.DynamicScheduled;
 import com.talkka.server.api.core.config.ApiKeyProvider;
 import com.talkka.server.api.core.exception.ApiClientException;
 
@@ -47,7 +47,7 @@ public String getApiKey(String path) throws ApiClientException {
 
 	// 매일 자정에 키 사용량 리셋
 	@PostConstruct
-	@Scheduled(cron = "0 0 0 * * *")
+	@DynamicScheduled(name = "Api Key Usage Reset", cron = "0 0 0 * * *")
 	public void init() {
 		rollingKeyIndex = 0;
 		keyList.clear();

diff --git a/server/src/main/java/com/talkka/server/bus/service/LocationCollectingScheduler.java b/server/src/main/java/com/talkka/server/bus/service/LocationCollectingScheduler.java
@@ -9,18 +9,16 @@
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.scheduling.annotation.EnableScheduling;
-import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Service;
 
+import com.talkka.server.admin.scheduler.DynamicScheduled;
 import com.talkka.server.admin.util.CollectedRouteProvider;
 import com.talkka.server.bus.util.LocationCollectingSchedulerConfigProperty;
 
 import lombok.extern.slf4j.Slf4j;
 
 @Slf4j
 @Service
-@EnableScheduling
 public class LocationCollectingScheduler {
 	private final LocationCollectingSchedulerConfigProperty locationCollectingSchedulerConfigProperty;
 	private final BusLocationCollectService busLocationCollectService;
@@ -38,37 +36,35 @@ public LocationCollectingScheduler(
 	}
 
 	// 병렬 버스 위치 api 호출 메소드
-	@Scheduled(fixedRate = 1000 * 60) // per minute
+	@DynamicScheduled(name = "collect bus", cron = "0 */3 * * * *")
 	public void runParallelLocationScheduler() {
-		if (isEnabled()) {
-			List<String> targetList = collectedRouteProvider.getTargetIdList();
-			ExecutorService executor = Executors.newFixedThreadPool(20);
+		List<String> targetList = collectedRouteProvider.getTargetIdList();
+		ExecutorService executor = Executors.newFixedThreadPool(20);
 
-			// CompletableFuture 리스트 생성
-			List<CompletableFuture<Void>> futures = targetList.stream()
-				.map(targetId -> CompletableFuture.runAsync(() -> {
-					try {
-						busLocationCollectService.collectLocationsByRouteId(targetId);
-					} catch (Exception e) {
-						// 재시도 후에도 실패한 경우, 로그 남기기
-						log.error(" 버스 위치 저장 실패 {} : {}", targetId, e.getMessage());
-					}
-				}, executor))
-				.toList();
+		// CompletableFuture 리스트 생성
+		List<CompletableFuture<Void>> futures = targetList.stream()
+			.map(targetId -> CompletableFuture.runAsync(() -> {
+				try {
+					busLocationCollectService.collectLocationsByRouteId(targetId);
+				} catch (Exception e) {
+					// 재시도 후에도 실패한 경우, 로그 남기기
+					log.error(" 버스 위치 저장 실패 {} : {}", targetId, e.getMessage());
+				}
+			}, executor))
+			.toList();
 
-			// 모든 작업이 완료될 때까지 기다림
-			CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
+		// 모든 작업이 완료될 때까지 기다림
+		CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
 
-			// ExecutorService 종료
-			executor.shutdown();
-			try {
-				if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
-					executor.shutdownNow();
-				}
-			} catch (InterruptedException e) {
+		// ExecutorService 종료
+		executor.shutdown();
+		try {
+			if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
 				executor.shutdownNow();
-				Thread.currentThread().interrupt();
 			}
+		} catch (InterruptedException e) {
+			executor.shutdownNow();
+			Thread.currentThread().interrupt();
 		}
 	}
 

diff --git a/server/src/main/resources/templates/admin/index.html b/server/src/main/resources/templates/admin/index.html
@@ -63,6 +63,15 @@ <h5 class="card-title">Api Key Management</h5>
                 </div>
             </div>
         </div>
+        <div class="col-md-3 mb-3">
+            <div class="card">
+                <div class="card-body text-center">
+                    <h5 class="card-title">Scheduler Management</h5>
+                    <p class="card-text">Manage schedulers.</p>
+                    <a href="/admin/scheduler" class="btn btn-primary">Go to Schedulers</a>
+                </div>
+            </div>
+        </div>
     </div>
 </div>