난수 생성 문제로 풀링 제거 원복 및 큐 사용 비동기처리.
This commit is contained in:
cho
@@ -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<AlarmEvent> registerAlarmList = new ArrayList<AlarmEvent>();
|
||||
List<AlarmEvent> registerAlarmList = new ArrayList<AlarmEvent>();
|
||||
|
||||
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<AlarmEvent> 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<AlarmKey, AlarmPolicy> alarmPolicyMap = alarmService.getAlarmPolicy();
|
||||
scheduler.scheduleWithFixedDelay(() -> {
|
||||
|
||||
Map<AlarmKey, SendPolicy> sendPolicyMap = alarmService.getSendPolicy();
|
||||
try {
|
||||
// === 여기에 지속 실행할 로직 ===
|
||||
fireAlarm();
|
||||
|
||||
Iterator<AlarmEvent> 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<AlarmEvent> 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<AlarmKey, AlarmPolicy> alarmPolicyMap = alarmService.getAlarmPolicy();
|
||||
|
||||
UmsRequestPayload payload = null;
|
||||
Map<AlarmKey, SendPolicy> sendPolicyMap = alarmService.getSendPolicy();
|
||||
|
||||
try {
|
||||
Iterator<AlarmEvent> iterator = registerAlarmList.iterator();
|
||||
|
||||
List<String> 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<String> 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);
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<UmsRequestPayload> queue = new ConcurrentLinkedQueue<UmsRequestPayload>();
|
||||
|
||||
// 2. 메모리 보호를 위해 크기 제한이 있는 큐 권장 (선택사항)
|
||||
private final Queue<UmsRequestPayload> 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);
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user