From 4f5fa96ef521b16be24d567b7d987d8944877d16 Mon Sep 17 00:00:00 2001 From: cho Date: Thu, 5 Mar 2026 14:48:37 +0900 Subject: [PATCH] =?UTF-8?q?=EB=82=9C=EC=88=98=20=EC=83=9D=EC=84=B1=20?= =?UTF-8?q?=EB=AC=B8=EC=A0=9C=EB=A1=9C=20=ED=92=80=EB=A7=81=20=EC=A0=9C?= =?UTF-8?q?=EA=B1=B0=20=EC=9B=90=EB=B3=B5=20=EB=B0=8F=20=ED=81=90=20?= =?UTF-8?q?=EC=82=AC=EC=9A=A9=20=EB=B9=84=EB=8F=99=EA=B8=B0=EC=B2=98?= =?UTF-8?q?=EB=A6=AC.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../eai/custom/alarm/AlarmStateManager.java | 179 +++++++++--------- .../eai/custom/alarm/ums/UmsSendManager.java | 87 +++++---- 2 files changed, 140 insertions(+), 126 deletions(-) diff --git a/src/main/java/com/eactive/eai/custom/alarm/AlarmStateManager.java b/src/main/java/com/eactive/eai/custom/alarm/AlarmStateManager.java index 83e4e14..f03ae3b 100644 --- a/src/main/java/com/eactive/eai/custom/alarm/AlarmStateManager.java +++ b/src/main/java/com/eactive/eai/custom/alarm/AlarmStateManager.java @@ -18,6 +18,7 @@ import com.eactive.eai.custom.alarm.event.AlarmEvent; import com.eactive.eai.custom.alarm.key.AlarmKey; import com.eactive.eai.custom.alarm.policy.AlarmPolicy; import com.eactive.eai.custom.alarm.policy.SendPolicy; +import com.eactive.eai.custom.alarm.ums.UmsSendManager; import com.eactive.eai.custom.alarm.ums.UmsService; import com.eactive.eai.custom.alarm.ums.payload.ObpMonitoringUmsTemplate; import com.eactive.eai.custom.alarm.ums.payload.UmsRequestPayload; @@ -25,121 +26,121 @@ import com.eactive.eai.custom.alarm.ums.payload.UmsRequestPayload; @Component public class AlarmStateManager { - List registerAlarmList = new ArrayList(); + List registerAlarmList = new ArrayList(); - private final ScheduledExecutorService scheduler; + private final ScheduledExecutorService scheduler; - protected static Logger logger = Logger.getLogger(Logger.LOGGER_DEFAULT); + protected static Logger logger = Logger.getLogger(Logger.LOGGER_DEFAULT); - private static AlarmStateManager alarmStateManager; + private static AlarmStateManager alarmStateManager; - @Autowired - AlarmService alarmService; + @Autowired + AlarmService alarmService; - @Autowired - UmsService usmService; + @Autowired + UmsSendManager umsSendManager; - public static AlarmStateManager getAlarmStateManager() { - if (alarmStateManager == null) { - alarmStateManager = ApplicationContextProvider.getContext().getBean(AlarmStateManager.class); - } - return alarmStateManager; - } - - public AlarmStateManager() { - - this.scheduler = Executors.newSingleThreadScheduledExecutor(r -> { - Thread t = new Thread(r); - t.setName("AlarmStateManager-worker"); - t.setDaemon(false); - return t; - }); - - scheduler.scheduleWithFixedDelay(() -> { - - try { - // === 여기에 지속 실행할 로직 === - fireAlarm(); - - } catch (Throwable e) { - // 반드시 예외 잡기 (안 잡으면 스케줄 자체가 죽음) - logger.error("fireAlarm fail", e); - } - - }, 0, 5, TimeUnit.SECONDS); // 최초 0초, 이후 5초 간격 - } - - public void onEvnet(AlarmEvent alaramEvent) { - - Optional existEvent = registerAlarmList.stream().filter(e -> e.equals(alaramEvent)).findFirst(); - - if (existEvent.isPresent()) { - AlarmEvent event = existEvent.get(); - event.onOcurr(); - } else { - registerAlarmList.add(alaramEvent); + public static AlarmStateManager getAlarmStateManager() { + if (alarmStateManager == null) { + alarmStateManager = ApplicationContextProvider.getContext().getBean(AlarmStateManager.class); + } + return alarmStateManager; } - } + public AlarmStateManager() { - private void fireAlarm() { + this.scheduler = Executors.newSingleThreadScheduledExecutor(r -> { + Thread t = new Thread(r); + t.setName("AlarmStateManager-worker"); + t.setDaemon(false); + return t; + }); - Map alarmPolicyMap = alarmService.getAlarmPolicy(); + scheduler.scheduleWithFixedDelay(() -> { - Map sendPolicyMap = alarmService.getSendPolicy(); + try { + // === 여기에 지속 실행할 로직 === + fireAlarm(); - Iterator iterator = registerAlarmList.iterator(); + } catch (Throwable e) { + // 반드시 예외 잡기 (안 잡으면 스케줄 자체가 죽음) + logger.error("fireAlarm fail", e); + } - while (iterator.hasNext()) { - AlarmEvent event = iterator.next(); - AlarmPolicy alarmPolicy = alarmPolicyMap.get(event.getKey()); + }, 0, 5, TimeUnit.SECONDS); // 최초 0초, 이후 5초 간격 + } - if (alarmPolicy != null && alarmPolicy.isAlarmTriggered(event)) { - // 알람 발송 로직 - SendPolicy sendPolicy = sendPolicyMap.get(event.getKey()); - if (alarmService.isAlarmEnabled()) { - send(sendPolicy, event); + public void onEvnet(AlarmEvent alaramEvent) { + + Optional existEvent = registerAlarmList.stream().filter(e -> e.equals(alaramEvent)).findFirst(); + + if (existEvent.isPresent()) { + AlarmEvent event = existEvent.get(); + event.onOcurr(); } else { - logger.info("Alarm is disabled, event = {}, policy = {}", event, sendPolicy); + registerAlarmList.add(alaramEvent); } - // 리스트에서 안전하게 제거 - iterator.remove(); - }else { //트리거 조건이 안되더라도, 이벤트발생한지 오래 되었으면 제거하여 OOM 예방. - - - long firstOcurredAt = event.getCondition().getOcurredAt(); - - long diffInMs = System.currentTimeMillis() - firstOcurredAt; - - if ( diffInMs >= TimeUnit.MINUTES.toMillis(60) ) { //1시간 지난거면 삭제 - iterator.remove(); - } - } + } - } + private void fireAlarm() { - private void send(SendPolicy sendPolicy, AlarmEvent event) { + Map alarmPolicyMap = alarmService.getAlarmPolicy(); - UmsRequestPayload payload = null; + Map sendPolicyMap = alarmService.getSendPolicy(); - try { + Iterator iterator = registerAlarmList.iterator(); - List cellPhoneList = sendPolicy.getCellPhoneList(); + while (iterator.hasNext()) { + AlarmEvent event = iterator.next(); + AlarmPolicy alarmPolicy = alarmPolicyMap.get(event.getKey()); - for (String cellPhone : cellPhoneList) { + if (alarmPolicy != null && alarmPolicy.isAlarmTriggered(event)) { + // 알람 발송 로직 + SendPolicy sendPolicy = sendPolicyMap.get(event.getKey()); + if (alarmService.isAlarmEnabled()) { + send(sendPolicy, event); + } else { + logger.info("Alarm is disabled, event = {}, policy = {}", event, sendPolicy); + } + // 리스트에서 안전하게 제거 + iterator.remove(); + } else { // 트리거 조건이 안되더라도, 이벤트발생한지 오래 되었으면 제거하여 OOM 예방. - payload = UmsRequestPayload.builder() - .content(ObpMonitoringUmsTemplate.builder().message(event.getMessage()).build()) - .cellphone(cellPhone).build(); - logger.info("Alarm is disabled, UmsRequestPayload = {}", payload); - usmService.send(payload); - } + long firstOcurredAt = event.getCondition().getOcurredAt(); - } catch (Exception e) { - logger.error(String.format("Alarm message delivery failed (UMS), payload=%s", payload), e); + long diffInMs = System.currentTimeMillis() - firstOcurredAt; + + if (diffInMs >= TimeUnit.MINUTES.toMillis(60)) { // 1시간 지난거면 삭제 + iterator.remove(); + } + } + } } - } + + private void send(SendPolicy sendPolicy, AlarmEvent event) { + + UmsRequestPayload payload = null; + + try { + + List cellPhoneList = sendPolicy.getCellPhoneList(); + + for (String cellPhone : cellPhoneList) { + + payload = UmsRequestPayload.builder() + .content(ObpMonitoringUmsTemplate.builder().message(event.getMessage()).build()) + .cellphone(cellPhone).build(); + logger.info("Alarm is disabled, UmsRequestPayload = {}", payload); + + umsSendManager.enqueue(payload); + } + + } catch (Exception e) { + logger.error(String.format("Alarm message delivery failed (UMS), payload=%s", payload), e); + + } + } } \ No newline at end of file diff --git a/src/main/java/com/eactive/eai/custom/alarm/ums/UmsSendManager.java b/src/main/java/com/eactive/eai/custom/alarm/ums/UmsSendManager.java index 396705f..8e9c5d5 100644 --- a/src/main/java/com/eactive/eai/custom/alarm/ums/UmsSendManager.java +++ b/src/main/java/com/eactive/eai/custom/alarm/ums/UmsSendManager.java @@ -3,10 +3,13 @@ package com.eactive.eai.custom.alarm.ums; import java.util.Queue; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import javax.annotation.PreDestroy; + import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; @@ -15,59 +18,69 @@ import com.eactive.eai.custom.alarm.ums.payload.UmsRequestPayload; @Component public class UmsSendManager { - + + protected static Logger logger = Logger.getLogger(Logger.LOGGER_DEFAULT); + // 1. I/O 처리를 위한 전용 스레드 풀 (예: 최대 10개 스레드) + private final ExecutorService ioExecutor = Executors.newFixedThreadPool(10); private final ScheduledExecutorService scheduler; - - protected static Logger logger = Logger.getLogger(Logger.LOGGER_DEFAULT); - Queue queue = new ConcurrentLinkedQueue(); - + // 2. 메모리 보호를 위해 크기 제한이 있는 큐 권장 (선택사항) + private final Queue queue = new ConcurrentLinkedQueue<>(); + @Autowired UmsService umsService; - + public UmsSendManager() { - this.scheduler = Executors.newSingleThreadScheduledExecutor(r -> { - Thread t = new Thread(r); - t.setName("AlarmStateManager-worker"); - t.setDaemon(false); - return t; - }); + this.scheduler = Executors.newSingleThreadScheduledExecutor(r -> { + Thread t = new Thread(r); + t.setName("UmsSendManager-worker"); + return t; + }); - scheduler.scheduleWithFixedDelay(() -> { - - try { - // === 여기에 지속 실행할 로직 === - sendMessage(); - - } catch (Throwable e) { - // 반드시 예외 잡기 (안 잡으면 스케줄 자체가 죽음) - logger.error("fireAlarm fail", e); - } - - }, 0, 5, TimeUnit.SECONDS); // 최초 0초, 이후 5초 간격 + scheduler.scheduleWithFixedDelay(() -> { + try { + sendMessage(); + } catch (Throwable e) { + logger.error("UMS 스케줄러 실행 중 오류", e); + } + }, 0, 30, TimeUnit.SECONDS); } - - void sendMessage() throws Exception { - UmsRequestPayload payload; - while ((payload = queue.poll()) != null) { + + void sendMessage() { + UmsRequestPayload payload; + // 3. 한 번의 주기당 최대 처리량 제한 (필요시) + int count = 0; + while ((payload = queue.poll()) != null && count < 1000) { final UmsRequestPayload finalPayload = payload; + + // 전용 Executor를 사용하여 공용 풀 보호 CompletableFuture.runAsync(() -> { - UmsCallResult result = null; try { - result = umsService.send(finalPayload); + umsService.send(finalPayload); } catch (Exception e) { logger.error("UMS 전송 실패", e); - }finally { - logger.info(result); } - }); + }, ioExecutor); + count++; } - } - + + // 4. 안전한 종료 처리 + @PreDestroy + public void stop() { + scheduler.shutdown(); + ioExecutor.shutdown(); + try { + if (!ioExecutor.awaitTermination(60, TimeUnit.SECONDS)) { + ioExecutor.shutdownNow(); + } + } catch (InterruptedException e) { + ioExecutor.shutdownNow(); + logger.error("UmsSendManager shutdownNow.", e); + } + } + public void enqueue(UmsRequestPayload payload) { queue.offer(payload); } - - }