[Spring] SSE & Kafka 를 활용해서 알림 기능 구현하기

필자는 처음에는 알림 테이블을 구현한후 댓글이 달리면 알림테이블에 데이터로 저장하였고, API가 호출되면 해당 사용자의 알림 데이터를 DB에서 가져오는 방식으로 구현하였다. 

하지만 이런 polling 방식은 정해진 주기마다 무조건 서버에 요청을 보내기 때문에 데이터가 없을 때도 트래픽을 발생시키며, 실시간으로 데이터가 업데이트 되지않는다는 단점이 있다. 

 

그래서 필자는 Kafka를 사용하여 알림 이벤트를 비동기적으로 처리하고, Server-Sent Events(SSE)를 통해 클라이언트에 실시간으로 알림을 전달하는 시스템을 구현해볼것이다. 게시판의 글에 댓글이 달리면 알람이 생기고, 서버는 Kafka를 통해 비동기적으로 알람 이벤트를 처리하며, 처리된 알람은 SSE를 통해 실시간으로 클라이언트에 전송하는 로직으로 구현할것이다.

 

---

 

 

전체적인 흐름

사용자의 행동이나 특정 이벤트가 발생하면 알림 이벤트가 생성되고 Kafka를 통해 처리된다. 이벤트는 AlarmConsumer에 의해 수신되고, AlarmService에서 처리되어 실제 알림으로 변환된 후, 해당 사용자에게 SSE를 통해 실시간으로 전송한다.

 

주요 클래스의 역할

• AlarmProducer: 사용자가 댓글을 달면 호출되어, Kafka의 "alarm" 토픽으로 알림 이벤트를 전송한다.
• AlarmConsumer: Kafka에서 "alarm" 토픽의 알림 이벤트를 수신하고, 해당 이벤트를 처리하기 위해 AlarmService의 send 메서드를 호출한다.
• AlarmService: 알림 데이터를 처리하여 데이터베이스에 저장하고, 해당 사용자의 SseEmitter가 있을 경우 실시간으로 알림을 전송한다. 또한, 사용자가 알림 구독을 요청할 때 (connectNotification 메서드) 새 SseEmitter를 생성하고 저장한다.
• EmitterRepository: 사용자별 SseEmitter 객체를 메모리 내에서 관리하며, 알림 전송 시 사용자의 SseEmitter를 조회하여 알림을 전송한다.
• KafkaProducerConfig: Kafka 프로듀서와 컨슈머의 설정을 정의한다.

 

---

 

Controller.java

    @GetMapping( "alarm/subscribe")
    public SseEmitter subscribe(@RequestParam Long userId) {
        log.info("subscribe");
        return alarmService.connectNotification(userId);
    }

    @GetMapping("alarm")
    public ResponseEntity<Page<AlarmReponse>> alarm(@RequestParam Long userId, Pageable pageable) {
        return ResponseEntity.ok().body(alarmService.getAlarmList(userId, pageable));
    }

 

@GetMapping( "alarm/subscribe")

• 사용자 ID에 대한 SSE(Server-Sent Events) 연결을 생성하고, 이 연결을 통해 실시간으로 서버와 클라이언트 간에 데이터 스트림이 설정된다.

@GetMapping("alarm")

• 해당 사용자의 알림 목록을 페이지네이션된 형태로 가져온다.

 

KafkaProducerConfig.java

• Kafka 프로듀서 설정을 위한 클래스
• Kafka와 통신하기 위한 여러 Bean들을 정의

@Configuration
public class KafkaProducerConfig {
    @Bean
    public ProducerFactory<String, AlarmEvent> producerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        return new DefaultKafkaProducerFactory<>(config);
    }

    @Bean
    public KafkaTemplate<String, AlarmEvent> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
            ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
            ConsumerFactory<Object, Object> kafkaConsumerFactory) {

        ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
        configurer.configure(factory, kafkaConsumerFactory);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        return factory;
    }
}

 

producerFactory() 메소드

• ProducerFactory<String, AlarmEvent> : Kafka로 메시지를 보낼 때 사용된다.
• ProducerConfig를 설정하는데, 여기에는 Kafka 서버 주소(BOOTSTRAP_SERVERS_CONFIG), 메시지 키의 직렬화 방식(KEY_SERIALIZER_CLASS_CONFIG), 메시지 값의 직렬화 방식(VALUE_SERIALIZER_CLASS_CONFIG) 등이 포함된다.
• 여기서 AlarmEvent는 메시지의 본문을 나타내는 클래스이며, JsonSerializer는 이 AlarmEvent 인스턴스를 JSON 형식으로 직렬화한다.

kafkaTemplate() 메소드

• KafkaTemplate<String, AlarmEvent> 타입의 Bean을 반환하며, 이는 Kafka로 메시지를 보내기 위한 템플릿으로 사용한다.
• 이 템플릿은 producerFactory()에서 생성한 ProducerFactory를 사용한다.

kafkaListenerContainerFactory() 메소드

• Kafka 메시지 리스너 컨테이너를 설정하기 위한 메소드
• ConcurrentKafkaListenerContainerFactory<?, ?> 는 Kafka 메시지 리스너들을 위한 컨테이너를 생성하는데 사용된다.
• ConcurrentKafkaListenerContainerFactoryConfigurer와 ConsumerFactory<Object, Object>를 이용해 리스너 컨테이너 팩토리를 구성한다.
• setAckMode(ContainerProperties.AckMode.MANUAL)는 메시지가 성공적으로 처리되었음을 애플리케이션에서 Kafka에 알리는 역할을 한다.

 

AlarmService.java

• 사용자에게 알람을 전송하고, 알람 연결을 관리하며, 알람 목록을 조회하는 클래스

@Slf4j
@RequiredArgsConstructor
@Service
public class AlarmService {
    private final AlarmRepository alarmRepository;
    private final EmitterRepository emitterRepository;
    private final UserRepository userRepository;
    private static final Long DEFAULT_TIMEOUT = 60L * 1000 * 60;
    
    private final static String ALARM_NAME = "alarm";

    @Transactional
    public void send(AlarmType type, AlarmArgs args, Long receiverId) {
        EntityUser user = userRepository.findByUserId(receiverId).orElseThrow(
                () -> new AppException(ErrorCode.USER_NOT_FOUND, "회원이 존재하지 않습니다."));
        HashMap<Long, Long> newArgs = new HashMap<>();
        newArgs.put(args.getFromUserId(), args.getTargetId());
        EntityAlarm alarm = EntityAlarm.of(type, newArgs, user);

        try {
            alarmRepository.save(alarm);
        } catch (Exception e){
            throw new AppException(ErrorCode.INTERNAL_SERVER_ERROR, "일시적 오류입니다");
        }

        emitterRepository.get(receiverId).ifPresentOrElse(it -> { //emitterRepo에서 인스턴스를 가지고 온다.
                    try {
                        it.send(SseEmitter.event() // 새로운 알람 전송
                                .id(alarm.getAlarmId().toString())
                                .name(ALARM_NAME)
                                .data(new AlarmNoti()));

                    } catch (IOException exception) {
                        emitterRepository.delete(receiverId);
                        throw new AppException(ErrorCode.NOTIFICATION_CONNECT_ERROR, "알림 전송에 실패했습니다.");
                    }
                },
                () -> log.info("No emitter founded")
        );
    }

    @Transactional
    public SseEmitter connectNotification(Long userId) {
        log.info("{userId = }", userId);
        SseEmitter emitter = new SseEmitter(DEFAULT_TIMEOUT);
        emitterRepository.save(userId, emitter);
        emitter.onCompletion(() -> emitterRepository.delete(userId));
        emitter.onTimeout(() -> emitterRepository.delete(userId));

        try {
            log.info("send");
            emitter.send(SseEmitter.event()
                    .id("id")
                    .name(ALARM_NAME)
                    .data("connect completed"));
        } catch (IOException exception) {
            throw new AppException(ErrorCode.NOTIFICATION_CONNECT_ERROR, "통신에 실패했습니다.");
        }
        return emitter;
    }

    @Transactional(readOnly = true)
    public Page<AlarmReponse> getAlarmList(Long userId, Pageable pageable) {
        return alarmRepository.findAllByUserUserId(userId, pageable).map(AlarmReponse::fromEntity);
    }

 

send 메소드

• 알람을 전송
• alarmRepository 알람 데이터를 저장한다.
• emitterRepository에서 SSE (Server-Sent Events) 이벤트 스트림을 관리하고, 새로운 알람 이벤트를 스트림에 보낸다.

connectNotification 메소드

• 클라이언트가 알림 수신을 위해 연결
• SseEmitter 인스턴스를 생성하고, 이를 emitterRepository에 저장한다.  연결 완료되면, 연결 완료 이벤트를 클라이언트에 보낸다.
• 연결 완료나 타임아웃 시 emitterRepository에서 해당 SseEmitter 인스턴스를 제거한다.

getAlarmList 메소드

• 사용자 ID로 알람 목록을 페이지 단위로 조회

 

EmitterRepository.java

서버에서 클라이언트로 서버 전송 이벤트(Server-Sent Events, SSE)를 전송하는 데 사용되는 SseEmitter 인스턴스를 로컬 캐시로 관리하는 클래스

@Slf4j
@Repository
@RequiredArgsConstructor
public class EmitterRepository { // 로컬캐시로 emitter 저장

    private Map<String, SseEmitter> emitterMap = new HashMap<>();

    public SseEmitter save(Long userId, SseEmitter emitter) {
        final String key = getKey(userId);
        log.info("Set Emitter to Redis {}({})", key, emitter);
        emitterMap.put(key, emitter);
        return emitter;
    }

    public void delete(Long userId) {
        emitterMap.remove(getKey(userId));
    }

    public Optional<SseEmitter> get(Long userId) { //emitter가 null일수도 잇으니깐 optional 처리
        SseEmitter result = emitterMap.get(getKey(userId));
        log.info("Get Emitter from Redis {}", result);
        return Optional.ofNullable(result);
    }

    private String getKey(Long userId) {
        return "emitter:UID:" + userId;
    }
}

 

Map<String, SseEmitter> emitterMap = new HashMap<>()

• SSE 이벤트 스트림을 관리하는 SseEmitter 인스턴스를 저장. 키는 사용자 ID를 기반으로 생성된 문자열이다.

save 메소드

• 특정 사용자 ID에 대한 SseEmitter 인스턴스를 저장
• 사용자 ID를 기반으로 키를 생성하고, 이 키와 함께 emitterMap에 SseEmitter 인스턴스를 저장한다.
• 저장된 SseEmitter 인스턴스를 반환환다.

delete 메소드

• 주어진 사용자 ID에 해당하는 SseEmitter 인스턴스를 emitterMap에서 제거한다.
• 사용자 ID를 기반으로 키를 생성하고, 해당 키에 대응하는 인스턴스를 맵에서 제거한다.

get 메소드

• 주어진 사용자 ID에 해당하는 SseEmitter 인스턴스를 조회한다.
• 사용자 ID를 기반으로 키를 생성하고, 해당 키를 사용하여 emitterMap에서 SseEmitter 인스턴스를 조회한다.

getKey 메소드

• "emitter:UID:" 접두사에 사용자 ID를 붙여 문자열을 생성한다. 이는 emitterMap에서 사용자별 SseEmitter 인스턴스를 구분하는 데 사용된다.

 

AlarmRepository.java

알람을 저장하는 클래스

@Repository
public interface AlarmRepository  extends JpaRepository<EntityAlarm, Long>, JpaSpecificationExecutor<EntityAlarm> {

    Page<EntityAlarm> findAllByUserUserId(Long userId, Pageable pageable);
}

 

 

AlarmArgs.java, AlarmEvent.java, AlarmNoti.java, AlarmType.java
알람 내용, 알람 정보 클래스

@Data
@AllArgsConstructor
@NoArgsConstructor
public class AlarmArgs {
    private Long fromUserId; //알람 발생자번호
    private Long targetId; //알람이 발생한 객체번호
}

@Data
@AllArgsConstructor
@NoArgsConstructor
public class AlarmEvent {
    private AlarmType type;//알람 타입
    private AlarmArgs args; // 알람 정보
    private Long receiverUserId; //알람받는 사람
}

@Data
@AllArgsConstructor
@NoArgsConstructor
public class AlarmNoti {
    private String text;
}

@Getter
@AllArgsConstructor
public enum AlarmType {
    NEW_PRODUCT("새로운 금융상품!"),
    NEW_RECOMMEND("새로운 추천상품!"),
    NEW_COMMENT("새로운 댓글!");

    private final String alarmText;
}

 

 

AlarmProducer.java

Kafka로 알람 이벤트를 전송하는 클래스

@Slf4j
@Component
@RequiredArgsConstructor
public class AlarmProducer {

    private final KafkaTemplate<String, AlarmEvent> alarmEventKafkaTemplate;

    public void send(AlarmEvent event) {
        alarmEventKafkaTemplate.send("alarm", String.valueOf(event.getReceiverUserId()), event);
        log.info("send fin");
    }
}

 

alarmEventKafkaTemplate: KafkaTemplate<String, AlarmEvent> 

• Kafka로 메시지를 전송하는 데 사용
• String은 메시지의 키 타입, AlarmEvent는 메시지의 값 타입이다.

send 메소드

• 알람 이벤트를 Kafka로 전송하는 메소드
• alarmEventKafkaTemplate.send() 메소드를 호출하여 실제로 메시지를 Kafka의 "alarm" 토픽으로 전송한다. 이때 메시지의 키로는 알람 수신자를 명시한다. 키는 메시지를 구분하고 파티셔닝할 때 사용된다. 메시지의 값으로는 event 객체가 전달된다.

 

AlarmConsumer.java

Kafka로부터 알람 관련 메시지를 수신하고 처리하는 클래스

@Slf4j
@Component
@RequiredArgsConstructor
public class AlarmConsumer {

    private final AlarmService alarmService;
    private final ObjectMapper objectMapper; // JSON 변환을 위한 ObjectMapper

    @KafkaListener(topics = "alarm")
    public void consumeNotification(ConsumerRecord<String, String> record, Acknowledgment ack) {
        try {
            AlarmEvent event = objectMapper.readValue(record.value(), AlarmEvent.class); // 메시지를 AlarmEvent 객체로 변환
            log.info("Consume the event {}", event);
            alarmService.send(event.getType(), event.getArgs(), event.getReceiverUserId());
            ack.acknowledge(); // 메시지 처리 후 수동으로 ack
        } catch (IOException e) {
            log.error("Error deserializing message", e);
        }
    }
}

 

 

@KafkaListener(topics = "alarm")

•  Kafka의 "alarm" 토픽으로부터 메시지를 수신하도록 지정.

ConsumerRecord<String, String>

• 메소드 내에서 objectMapper를 사용하여 메시지의 값을 AlarmEvent 객체로 변환.

alarmService.send()

• 변환된 AlarmEvent 객체를 사용하여 alarmService의 send 메소드를 통해 알람을 발송.

ack.acknowledge()

• Kafka에서 메시지 소비자가 특정 메시지를 성공적으로 처리했음을 알림.

 

 

이제 실습해보자!

필자는 kafka를 도커에 띄워서 도커를 실행시켜주겠다.

 

 

"alarm/subscribe" API 호출하여 사용자 1번에 대한 SSE(Server-Sent Events) 연결을 생성한다.

 

그리고 kafka에 알람이 오는지 확인하기위해 kafka 컨테이너에 접속하자

// 도커 컨테이너 접속
docker exec -it [ContainerID] /bin/bash

// 컨슈머 경로로 이동
cd /opt/kafka_2.13-2.8.1/bin/
./kafka-console-consumer.sh --topic [토픽이름] --bootstrap-server kafka:9092

 

 

사용자 1번의 게시글을 만들어준다.

 

그리고 사용자 2번은 사용자 1번의 게시글에 댓글을 단다.

 

그리고 사용자 1번의 알림을 확인해보자.

 

 

알림리스트에도 생기고 kafka 컨테이너에도 알람이 정상적으로 생긴것을 확인할수있다!!

성공!!🥳