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); } - - } diff --git a/src/main/java/com/eactive/eai/custom/alarm/ums/UmsService.java b/src/main/java/com/eactive/eai/custom/alarm/ums/UmsService.java index cc2b114..74777de 100644 --- a/src/main/java/com/eactive/eai/custom/alarm/ums/UmsService.java +++ b/src/main/java/com/eactive/eai/custom/alarm/ums/UmsService.java @@ -1,15 +1,20 @@ package com.eactive.eai.custom.alarm.ums; +import java.io.IOException; import java.nio.charset.StandardCharsets; +import javax.annotation.PreDestroy; + import org.apache.hc.client5.http.classic.methods.HttpPost; import org.apache.hc.client5.http.config.RequestConfig; import org.apache.hc.client5.http.entity.EntityBuilder; import org.apache.hc.client5.http.impl.classic.CloseableHttpClient; import org.apache.hc.client5.http.impl.classic.HttpClients; +import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManagerBuilder; +import org.apache.hc.client5.http.io.HttpClientConnectionManager; import org.apache.hc.core5.http.ContentType; -import org.apache.hc.core5.http.HttpVersion; import org.apache.hc.core5.http.io.entity.EntityUtils; +import org.apache.hc.core5.util.TimeValue; import org.apache.hc.core5.util.Timeout; import org.springframework.stereotype.Component; @@ -38,6 +43,24 @@ public class UmsService { private final static String DEFAULT_API_KEY = "7cca6d58-e4f7-474d-9edd-525806bc99ff"; private final static String AUTHORIZATION_FMT = "Bearer %s"; + // 재사용되는 HttpClient (Thread-Safe) + private final CloseableHttpClient httpClient; + + public UmsService() { + // 커넥션 풀 설정: 매번 소켓을 열고 닫지 않고 재사용함 + HttpClientConnectionManager cm = PoolingHttpClientConnectionManagerBuilder.create() + .setMaxConnTotal(100) // 전체 최대 커넥션 수 + .setMaxConnPerRoute(20) // 호스트당 최대 커넥션 수 + .setConnectionTimeToLive(TimeValue.ofMinutes(3)) // TTL + .setValidateAfterInactivity(TimeValue.ofSeconds(5)) // 🔥 중요 + .build(); + + this.httpClient = HttpClients.custom() + .setConnectionManager(cm) + .evictExpiredConnections() // 만료 제거 + .evictIdleConnections(TimeValue.ofSeconds(20)) // 20초 idle 제거 + .build(); + } public UmsCallResult send(UmsRequestPayload payload) throws Exception { if (payload.getContent() == null) { @@ -72,9 +95,6 @@ public class UmsService { .setContentType(ContentType.APPLICATION_JSON) .setContentEncoding("UTF-8") .build()); - - httpPost.setVersion(HttpVersion.HTTP_1_0); - httpPost.setHeader("Connection", "close"); // 3. 타임아웃 설정 (설정파일 값 적용) RequestConfig config = RequestConfig.custom() @@ -89,7 +109,7 @@ public class UmsService { } // 4. 실행 (HttpClient 재사용) - try (CloseableHttpClient httpClient = HttpClients.createDefault()) { + try { return httpClient.execute(httpPost, response -> { String body = EntityUtils.toString(response.getEntity(), StandardCharsets.UTF_8); @@ -120,4 +140,15 @@ public class UmsService { } } + @PreDestroy + public void destroy() { + try { + if (httpClient != null) { + httpClient.close(); + logger.info("UMS HttpClient closed."); + } + } catch (IOException e) { + logger.error("Error closing UMS HttpClient", e); + } + } } \ No newline at end of file