// 1. Kafka Streams 설정을 위한 Properties 생성
Properties props = new Properties();
// application.id
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "alarm-manager-app");
// bootstrap.servers
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9093");
// key/value serde 설정 등
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
// (선택) 에러 로그 보기 쉽게 하기 위해 record 캐싱 끔
props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, "0");
// (선택) commit 주기 짧게
props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000);
설정 키
설명
APPLICATION_ID_CONFIG
Kafka Streams 앱 식별자 (state store 이름 등에 사용됨)
BOOTSTRAP_SERVERS_CONFIG
Kafka 브로커 주소
DEFAULT_KEY_SERDE_CLASS_CONFIG
메시지 Key 직렬화/역직렬화 방식
CACHE_MAX_BYTES_BUFFERING_CONFIG
캐시 비활성화 (실시간 디버깅에 도움됨)
COMMIT_INTERVAL_MS_CONFIG
오프셋 커밋 주기 설정 (기본 30초 → 1초로 단축)
(+) 알람 발생 조건 AlarmCondition.java 파일 생성
public class AlarmCondition {
public String metricKey;
public double thresholdValue;
public String operator;
public int severity;
public int category;
public AlarmCondition(String metricKey, double thresholdValue,
String operator, int severity, int category) {
this.metricKey = metricKey;
this.thresholdValue = thresholdValue;
this.operator = operator;
this.severity = severity;
this.category = category;
}
}
각 파라미터 값은 알람 발생 조건 DB에 맞게 작성한다.
(2) 알람 발생 조건을 저장할 Map 초기화
// 2. 알람 발생 조건을 저장할 Map 초기화
// key: metric_key (예: login_fail)
// value: 조건 객체 (threshold, operator, severity, category 등)
Map<String, AlarmCondition> alarmConditions = new ConcurrentHashMap<>();
Kafka Streams가 멀티스레드 환경이니 동시성 문제를 방지하기 위해 ConcurrentHashMap을 사용한다.
(3) StreamBuilder 생성
// 3. StreamsBuilder 생성
StreamsBuilder builder = new StreamsBuilder();
(4) alarm-condition 토픽 구독 (조건 업데이트 수신)
// 4. alarm-condition 토픽 구독 (조건 업데이트 수신)
// alarm-condition 메시지를 받아서 JSON 파싱
ObjectMapper objectMapper = new ObjectMapper();
KStream<String, String> conditionStream = builder.stream("alarm-condition");
// Map<String, AlarmCondition> 형태로 조건 저장 (기존 조건 갱신 포함)
conditionStream
.peek((key, value) -> System.out.println("[조건 수신] key: " + key + ", value: " + value))
.foreach((key, value) -> {
try {
// 메시지를 Map으로 파싱
Map<String, Object> conditionData = objectMapper.readValue(value, Map.class);
String metricKey = (String) conditionData.get("metric_key");
double threshold = Double.parseDouble(conditionData.get("threshold_value").toString());
String operator = (String) conditionData.get("condition_operator");
int severity = Integer.parseInt(conditionData.get("severity").toString());
int category = Integer.parseInt(conditionData.get("category").toString());
// Map에 조건 저장 또는 갱신
alarmConditions.put(metricKey, new AlarmCondition(metricKey, threshold, operator, severity, category));
System.out.println("[조건 갱신] metric_key: " + metricKey);
} catch (Exception e) {
System.out.println("alarm-condition 파싱 실패: " + e.getMessage());
}
});
알람매니저는 event-topic을 구독하여 데이터를 수신 후 알람 발생 조건과 비교하여 알람 발생 유무를 판단한다.
알람 발생 조건 만족 시 alarm-topic으로 해당 데이터를 전송하고 웹 서버는 해당 토픽을 구독하여 알람을 발생시킨다.
알람 발생 조건 불만족 시 event-log-topic으로 해당 데이터를 전송하고 웹 서버는 해당 토픽을 구독하여 DB에 저장한다.
먼저 웹 서버 최초 시작 시
Kafka Producer와 Kafka Producer Config는 저번 게시글의 코드와 동일하다.
// Kafaka Topic(=Message) 를 Produce(=Send) 구현체
@Service
@RequiredArgsConstructor
public class KafkaProducer {
private final KafkaTemplate<String, String> kafkaTemplate;
private static final String event = "event-topic"; // 웹 서버에서 발생한 이벤트
private static final String alarm = "alarm-topic"; // 발생한 알람
private static final String log = "event-log-topic"; // 알람이 아닌 이벤트
private static final String condition = "alarm-condition"; // 알람 발생 조건
// 토픽 전송
public void sendMessageTo(String topic, String key, String message) {
kafkaTemplate.send(topic, key, message);
}
}
// Kafaka Topic(=Message)를 Produce(=Send) 하는 설정 클래스
@Configuration
public class KafkaProducerConfig {
// ProducerFactory: Kafka 메시지 전송을 위한 프로듀서 팩토리
public ProducerFactory<String,String> producerFactory() {
Map<String, Object> configProps = new HashMap<String, Object>();
configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9093"); // Kafka 서버 주소 (로컬 PC에서 구현 시)
//configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092"); // Kafka 서버 주소 (이후 도커에서 구현 시)
configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); // Kafka 메시지의 키를 문자열로 직렬화
configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class); // Kafka 메시지의 값을 문자열로 직렬화
return new DefaultKafkaProducerFactory<>(configProps);
}
// KafkaTemplate: Kafka에 메시지를 보낼 때 사용하는 핵심 객체
@Bean
public KafkaTemplate<String, String> kafkaTemplate() {
return new KafkaTemplate<>(producerFactory());
}
}
웹 서버 애플리케이션 실행 시 (참고로 자바 파일 코드도 실행시켜야 한다)
알람 발생 조건이 alarm-condition 토픽으로 전송이 되고, 알람 매니저는 해당 토픽을 구독하여 메시지를 수신받는다.
이후 웹에서 발생한 이벤트 데이터를 Kafka의 event-topic으로 전송한다.
기능 구현 테스트를 위해 알람 발생 조건에 해당하는 값을 포스트맨으로 전송한다.
@RestController
@RequiredArgsConstructor
public class KafkaController {
private final KafkaProducer producer;
private final KafkaProducer kafkaProducer;
@GetMapping("/kafka/send")
public String sendMessage(@RequestParam String topic, @RequestParam String key,@RequestParam String message) {
kafkaProducer.sendMessageTo(topic, key, message);
return "전송됨 → 토픽: " + topic + ", key: " + key + ", 메시지: " + message;
}
}
해당 값이 알람 발생 조건에 해당하므로 알람 매니저는 해당 내용을 alarm-topic으로 전송한다.
UI 확인 결과 alarm-topic에 해당 내용이 저장되었음을 알 수 있다.
웹 서버에서 토픽을 구독하는 Consumer, Consumer Config 코드도 이전과 동일하다.
// Kafaka Topic(=Message) 를 Consume(=Recv) 하는 구현체
@Slf4j
@Service
@RequiredArgsConstructor
public class KafkaConsumer {
private final SseService sseService;
private final AlarmService alarmService;
@KafkaListener(topics = "event-topic", groupId = "cplug-consumer-group")
public void consumeEvnet(String message) {
log.warn("[event] Kafka 메시지 수신: " + message);
// 여기에 알람 UI 처리, 로그 저장 등
}
@KafkaListener(topics = "alarm-topic", groupId = "cplug-consumer-group")
public void consumeAlarm(@Header(KafkaHeaders.RECEIVED_KEY) String key, String message) {
log.warn("[alarm] Kafka key: " + key); // 1 (orch) -> network_type
log.warn("[alarm] Kafka message: " + message); // {"type":"login_fail","count":6,"target":"user01","time":"2025-05-22T18:15:00","severity":3,"category":1}
// SSE: 알람 UI 처리
Map<String, Object> payload = new HashMap<>();
try {
ObjectMapper mapper = new ObjectMapper();
payload = mapper.readValue(message, Map.class); // map -> {"type":"login_fail","count":6,"target":"user01","time":"2025-05-22T18:15:00"}
String target = (String) payload.get("target");
sseService.send(target, message);
} catch (Exception e) {
log.warn("알람 메시지 처리 실패: {}", message, e);
}
log.warn("key: " + key);
log.warn("networkType class1: " + key.getClass());
int networkType = Integer.parseInt(key);
payload.put("network_type", networkType);
log.warn("payload: {}", payload);
// {type=login_fail, count=6, target=user01, time=2025-05-22T18:15:00, severity=3, category=1, network_type=1}
// DB 저장
int createAlarm = alarmService.createAlarm(payload);
log.warn("알람 DB 저장 결과: " + createAlarm);
}
}
// Kafaka Topic(=Message) 를 Consume(=Recv) 하는 설정 클래스
@EnableKafka
@Configuration
public class KafkaConsumerConfig {
public ConsumerFactory<String, String> consumerFactory() {
Map<String, Object> props = new HashMap<>();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9093"); // Kafka 브로커
//props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092"); // Kafka 브로커
props.put(ConsumerConfig.GROUP_ID_CONFIG, "cplug-consumer-group"); // Consumer Group ID (같은 group-id를 공유하는 컨슈머끼리는 메시지를 나눠가짐)
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
return new DefaultKafkaConsumerFactory<>(props);
}
// kafkaListenerContainerFactory()
@Bean
public ConcurrentKafkaListenerContainerFactory<String, String>
kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<String, String> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
return factory;
}
}
alarm-topic을 구독하고 있는 웹 서버는 메시지를 수신받는다.
수신받은 알람 데이터를 토대로 SSE 연결 (위 로그에서는 SSE 연결이 기존에 되지 않음) 및
DB 저장이 됨을 확인할 수 있다.
SSE 연결은 프론트에서 위 기능을 실행하기 전에 SSE 연결에 대한 요청을 전송하면 된다.
@RestController
@RequiredArgsConstructor
@RequestMapping("/alarm")
public class AlarmSseController {
private final SseService sseService;
// 프론트에서는 /alarm/subscribe?clientId=user01 로 접속해서 실시간 알람 수신
@GetMapping("/subscribe")
public SseEmitter subscribe(@RequestParam String clientId){
return sseService.subscribe(clientId);
}
}
@Slf4j
@Service
public class SseService {
private final Map<String, SseEmitter> emitters = new ConcurrentHashMap<>();
// 사용자 ID에 대한 SSE(Server-Sent Events) 연결을 생성하고, 이 연결을 통해 실시간으로 서버와 클라이언트 간에 데이터 스트림이 설정된다.
public SseEmitter subscribe(String clientId) {
SseEmitter emitter = new SseEmitter(Long.MAX_VALUE);
emitters.put(clientId, emitter);
emitter.onCompletion(() -> emitters.remove(clientId));
emitter.onTimeout(() -> emitters.remove(clientId));
log.warn("SSE 구독 연결: {}", clientId);
return emitter;
}
// 알람 전송
public void send(String clientId, String message) {
SseEmitter emitter = emitters.get(clientId);
if (emitter != null) {
try {
emitter.send(SseEmitter.event()
.name("alarm")
.data(message));
log.warn("SSE 전송 완료 → {}", message);
} catch (IOException e) {
log.warn("SSE 전송 실패 → {}", message, e);
emitters.remove(clientId);
}
} else {
log.warn("연결된 SSE 없음: {}", clientId);
}
}
}
SSE 연결 요청은 일회성 연결인 포스트맨이 아닌 웹 브라우저를 통해 요청할 수 있도록 한다.
