[Spring] kafka와 SSE(Server Send Event)를 사용한 실시간 알림 전송 방법

Kafka 선택 이유

알림 전송에 Kafka를 도입한 이유는 다음과 같다. 

필자의 과일샵 쇼핑몰 프로젝트는 아래 처럼 회원 서버와 어드민 서버로 나뉘어져 있다.

두 서버가 DB는 공유하고 있지만 ,어드민 서버에서 발생한 이벤트를 실시간으로 회원 서버로 전달할 수가 없었다.

또한 알림에 대한 처리를 기다릴 필요가 없어 비동기로 처리할 수 있었다.

바로 이 문제를 해결하기 위해 Kafka를 도입하게 되었다.

Kafka외에도 RabbitMQ나 Redis를 고려 해 볼 수도 있었지만, 분산 시스템이 적용되어 있는 큰 기업들에서 사용하고 있어 경험해보기 위해 Kafka를 선택하게 되었다.

[Kafka] 카프카란 ? 주요개념 정리 및 Pub/Sub 모델 비교

 

SSE 선택 이유

 SSE를 선택한 이유는 Kafka로 부터 수신한 메시지를 클라이언트에게 실시간으로 전달하기 위해서이다.

 

알림이 왔을 때 실시간으로 "빨간원" 표시

실시간 알림을 구현하기 위해서는 Websoket도 비교대상으로 고려해볼 수 있었다.

이에 대한 SSE와 Websoket의 차이점과 SSE에 대한 자세한 설명은 아래 포스팅에서 다룬다.

실시간 알림 - SSE(Server Sent Event), Websoket 차이점 정리

 

 

✔ 전체 흐름
1. 어드민 서버: 관리자가 회원의 리뷰에 답글을 남기면 해당 메시지에 맞는 Topic을 지정하여 알림 메시지를 Kafka로 발행.
2. Kafka: 어드민 서버에서 생성된 알림 메시지를 받아서 해당 Topic을 구독하는 회원 서버로 전달
3. 회원 서버: Kafka에서 받은 알림 메시지를 수신하여 저장. 그리고 해당 메시지를 SSE 연결된 사용자에게 전달 

 

참고로 이 포스팅의 코드는 알림에 관련한 코드만을 다루니 필자의 부가적인 기능에 대한 코드는 다루지 않았다.

더 많은 내용을 보고 싶으면 아래 저장소를 참고하는 것을 추천한다.

회원 서버 : https://github.com/hyeon0208/fruit-mall

괸라자 서버 : https://github.com/hyeon0208/fruit-mall-admin

---

Docker로 kafka 설치 및 Spring 의존성 추가

필자는 Kafka를 Docker로 설치했다.

 

kafka-compose.yml

아래 컴포즈 설정은 broker(카프카 서버)가 1개로 구성된 설정이다.

만약 broker 개수를 늘리고 싶으면 포트번호가 다른 kafka를 하나더 추가하고, KAFKA_BROKER_ID를 지정해주면 된다.

version: '3.8'
services:
  zookeeper:
    image: wurstmeister/zookeeper:latest
    container_name: zookeeper
    ports:
      - "2181:2181"
  kafka:
    image: wurstmeister/kafka:latest
    container_name: kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_ADVERTISED_HOST_NAME: 127.0.0.1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock

 

이제 터미널에서 아래 명령어를 입력해 컨테이너를 생성하고 실행하도록하자

docker-compose -f kafka-compose.yml up

 

도커 데스크탑을 통해 해당 컨테이너가 정상 실행되고 있는 것을 확인할 수 있다. 

(도커 데스크탑을 설치하지 않았다면 터미널에서 docker ps 명령어로 확인)

 

kafka 라이브러리 추가

kafka를 사용할 프로젝트의 build.grdle에 라이브러리 추가.

(필자는 회원 서버, 어드민 서버 2개의 서버에 해당 라이브러리를 추가했다.)

implementation 'org.springframework.kafka:spring-kafka'

 

---

Kafka Producer 구현 (어드민 서버)

 

KafkaProducerConfig - 카프카 프로듀서 설정 클래스

설정 방법은 application.properties에 설정 정보를 작성해도 되고 필자처럼 config 파일을 작성해도된다.

(직렬화 패키지에 주의해서 import)

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.kafka.support.serializer.JsonSerializer;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;

@Configuration
public class KafkaProducerConfig {
    @Bean
    public ProducerFactory<String, NotificationMessage> 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, NotificationMessage> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
}

 

◼ BOOTSTRAP_SERVERS_CONFIG

kafka 서버가 실행되는 포트를 입력해주면 된다.

 

◼ KEY_SERIALIZER_CLASS_CONFIG

Producer가 key 값의 데이터를 Kafka 브로커로 전송하기 전에 데이터를 Byte Array로 변환하는데 사용하는 직렬화 매커니즘 설정이다.

key값은 String이기 때문에 String으로 직렬화 했다.

 

◼ VALUE_SERIALIZER_CLASS_CONFIG

위와 동일한 Value값의 직렬화 메커니즘 설정이다.

필자는 NotificationMessage 객체를 브로커로 전달하기 때문에 Json으로 직렬화 했다.

 

◼ KafkaTemplate

Spring에서 제공하는 Kafka Producer를 Wrapping한 클래스로 kafka에 메시지를 보내는 여러 메서드를 제공한다.

 

NotificationMessage 

참고로 이 클래스로 전달되는 메시지를 Consumer가 받기 위해서는 Producer의 NotificationMessage  클래스의 패키지 위치와 동일해야한다.

(필자는 어드민 서버, 회원 서버 모두 com.fruit.common 위치에 해당 클래스를 생성두었다.)

@Getter
@AllArgsConstructor
public class NotificationMessage {
    private String userId;
    private String message;
}

 

NotificationsService

@Service
@Slf4j(topic = "elk")
@RequiredArgsConstructor
public class NotificationsService {
    private final KafkaTemplate<String, NotificationMessage> kafkaTemplate;

    public void commentNotificationCreate(String userId, Long reviewId, String message) {
        NotificationMessage notificationMessage = new NotificationMessage(reviewId, userId, message);
        log.info("리뷰 답글 알림 전송. userId : {}, message : {}",userId, message);
        kafkaTemplate.send("comment-notifications", notificationMessage);
    }
}

 

Producer -> Kafka 메시지 전송

@Service
@Slf4j
@RequiredArgsConstructor
public class ProducerService {
    private final KafkaTemplate<String, NotificationMessage> kafkaTemplate;

    public void commentNotificationCreate(String userId, String message) {
        NotificationMessage notificationMessage = new NotificationMessage(userId , message);
        log.info("리뷰 답글 알림 전송. userId : {}, message : {}",userId, message);
        kafkaTemplate.send("comment-notifications", notificationMessage);
    }
}

위 설정에서 빈 등록한 KafkaTemplate를 의존성 주입하여 kafkaTemplate의 send 메서드 인자로  토픽을 정하고 데이터를 담아  메시지를 kafka로 전달한다.

 

이제 해당 메서드를 알림을 이벤트 발생이 필요한 부분에 추가해주면 된다.

// 예시 코드
	@PostMapping("/api/v1//reply/notifications")
    public ResponseEntity<?> sendReplyNotifications(@RequestBody NotificationsDto dto) {
        Long userId = reviewService.selectUserIdByReviewId(dto.getReviewId());
        String message = dto.getProductName() + " 상품의 리뷰에 판매자가 댓글을 남겼습니다.";
        notificationsService.commentNotificationCreate(String.valueOf(userId), message);
        return ResponseEntity.ok().build();
    }

 

---

Kafka Consumer 구현 (회원 서버)

KafkaConsumerConfig - 카프카 컨슈머 설정 클래스

@Configuration
public class KafkaConsumerConfig {

    @Bean
    public ConsumerFactory<String, NotificationMessage> consumerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        config.put(ConsumerConfig.GROUP_ID_CONFIG, "group_1");
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

        JsonDeserializer<NotificationMessage> deserializer = new JsonDeserializer<>(NotificationMessage.class);
        deserializer.addTrustedPackages("com.fruit.common");

        return new DefaultKafkaConsumerFactory<>(config,
                new StringDeserializer(),
                deserializer);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, NotificationMessage> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, NotificationMessage> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }
}

설정은 ProducerCofing과 비슷하다.

다른점에 대해 설명하면 다음과 같다.

 

◼ GROUP_ID_CONFIG

Cosumer가 속할 그룹을 설정 했다. 

 

◼ JsonDeserializer<NotificationMessage> deserializer 와deserializer.addTrustedPackages("com.fruit.common");

JsonDeserializer를 사용해 메세지 Value값을 NotificationMessage 객체로 변환하고, 해당 패키지에 속하는 클래스를 안전하게 역질렬화 할 수 있도록 허용하였다.

이 addTrustedPackages로 실회할수있는 패키지를 추가하지 않으면 "신뢰할 수 없는 패키지 ... 머시기" 에러가 발생한다.

 

 

◼ ConcurrentKafkaListenerContainerFactory

Spring의 @KafkaListener 어노테이션이 붙은 메서드에 주입되어 사용되는 클래스이다.

String 타입의 키와 NotificationMessage 타입의 값을 가진 Kafka 메시지를 동시에 처리할 수 있는 Listener Container들을 생성한다.

 

---

SSE(Server Send Event) 코드 작성 (회원 서버)

 

Notifications

@Getter
@ToString
@NoArgsConstructor
public class Notifications {
    private Long notificationsId;
    private Long reviewId;
    private String notificationsMessage;
    private Timestamp notificationsCreatedAt;
    private Timestamp isRead;

    @Builder
    public Notifications(Long reviewId, String notificationsMessage) {
        this.reviewId = reviewId;
        this.notificationsMessage = notificationsMessage;
    }
}

 

 

NotificationsService

회원서버의 NotificationsService는 어드민 서버의 NotificationsService와 다르게

알림 테이블에 MyBatis를 사용해 데이터를 저장하고 있다.

따라서 NotificationsRepository의 구현부는 테이블 구조와 데이터베이스 프레임워크에 따라 다르기 때문에 생략하였다.

@RequiredArgsConstructor
@Service
@Transactional
public class NotificationsService {
    private final NotificationsRepository notificationsRepository;

    public void insertNotifications(Notifications notifications) {
        notificationsRepository.insertNotifications(notifications);
    }

    public void deleteOldestNotificationsByUserId(Long userIdNo, int delCount) {
        notificationsRepository.deleteOldestNotificationsByUserId(userIdNo, delCount);
    }

    public int countNotificationsByUserId(Long userIdNo) {
        return notificationsRepository.countNotificationsByUserId(userIdNo);
    }

    public List<NotificationsResDto> selectMessagesByUserId(Long userIdNo) {
        return notificationsRepository.selectMessagesByUserId(userIdNo);
    }

    public void updateRead(Long notificationsId) {
        notificationsRepository.updateRead(notificationsId);
    }

    public Long selectProductIdByNotificationsId(Long notificationsId) {
        return notificationsRepository.selectProductIdByNotificationsId(notificationsId);
    }
}

 

 

EmitterRepository

SSE연결 정보들을 DB가 아닌 Map<>에(JVM에 저장) CRUD 하는 메서드들을 작성하였다.

@Repository
public class EmitterRepository {
    private final Map<String, SseEmitter> emitters = new ConcurrentHashMap<>();
    private final Map<String, Object> eventCache = new ConcurrentHashMap<>();

    public SseEmitter save(String emitterId, SseEmitter sseEmitter) {
        emitters.put(emitterId,sseEmitter);
        return sseEmitter;
    }

    public void saveEventCache(String emitterId, Object event) {
        eventCache.put(emitterId,event);
    }

    public Map<String, SseEmitter> findAllEmitters() {
        return new HashMap<>(emitters);
    }

    public Map<String, SseEmitter> findAllEmitterStartWithById(String memberId) {
        return emitters.entrySet().stream()
                .filter(entry -> entry.getKey().startsWith(memberId))
                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
    }

    public Map<String, Object> findAllEventCacheStartWithById(String memberId) {
        return eventCache.entrySet().stream()
                .filter(entry -> entry.getKey().startsWith(memberId))
                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
    }

    public void deleteById(String emitterId) {
        emitters.remove(emitterId);
    }
}

◼ ConcurrentHashMap

동시성 문제를 해결가능한 Map 자료형이다.

여러 사용자가 동시에 접근하더라도 안전하게 데이터를 조작할 수 있도록 하기위해 해당 자료형을 사용했다.

 

◼ emitters

사용자별로 생성된 Key와 SseEmitter 객체를 저장하는 ConcurrentHashMap이다.

 

◼ eventCache

사용자별로 생성된 Key를 사용하여 해당 사용자에게 전송되지 못한 이벤트를 저장하는 ConcurrentHashMap이다.

네트워크 연결이 끊기는 등의 이유로 알림이 사용자에게 전달이 되지 못했을 때 이벤트 유실을 방지하기 위해 신뢰성을 보장하고자 해당 저장소를 추가하였다.

 

EmitterService

@Service
@Slf4j
@RequiredArgsConstructor
public class EmitterService {
    private final NotificationsService notificationsService;
    private final EmitterRepository emitterRepository;

    private static final int MAX_NOTIFICATIONS_COUNT = 6;

    @KafkaListener(topics = "comment-notifications", groupId = "group_1")
    public void listen(NotificationMessage message) {
        String userId = message.getUserId();
        Notifications notifications = Notifications.builder()
                .userIdNo(Long.valueOf(userId))
                .message(message.getMessage())
                .type("답글알림").build();

        notificationsService.insertNotifications(notifications);
        int curCnt = notificationsService.countNotificationsByUserId(Long.valueOf(userId));
        if (curCnt > MAX_NOTIFICATIONS_COUNT) {
            int delCount = curCnt - MAX_NOTIFICATIONS_COUNT;
            notificationsService.deleteOldestNotificationsByUserId(Long.valueOf(userId), delCount);
        }

        Map<String, SseEmitter> sseEmitters = emitterRepository.findAllEmitterStartWithById(userId);
        sseEmitters.forEach(
                (key, emitter) -> {
                    emitterRepository.saveEventCache(key, notifications);
                    sendToClient(emitter, key, notifications);
                }
        );
    }

    private void sendToClient(SseEmitter emitter, String emitterId, Object data) {
        try {
            emitter.send(SseEmitter.event()
                    .id(emitterId)
                    .data(data));
            log.info("Kafka로 부터 전달 받은 메세지 전송. emitterId : {}, message : {}", emitterId, data);
        } catch (IOException e) {
            emitterRepository.deleteById(emitterId);
            log.error("메시지 전송 에러 : {}", e);
        }
    }

    public SseEmitter addEmitter(String userId, String lastEventId) {
        String emitterId = userId + "_" + System.currentTimeMillis();
        SseEmitter emitter = emitterRepository.save(emitterId, new SseEmitter(DEFAULT_TIMEOUT));
        log.info("emitterId : {} 사용자 emitter 연결 ", emitterId);

        emitter.onCompletion(() -> {
            log.info("onCompletion callback");
            emitterRepository.deleteById(emitterId);
        });
        emitter.onTimeout(() -> {
            log.info("onTimeout callback");
            emitterRepository.deleteById(emitterId);
        });

        sendToClient(emitter, emitterId, "connected!"); // 503 에러방지 더미 데이터

        if (!lastEventId.isEmpty()) {
            Map<String, Object> events = emitterRepository.findAllEventCacheStartWithById(userId);
            events.entrySet().stream()
                    .filter(entry -> lastEventId.compareTo(entry.getKey()) < 0)
                    .forEach(entry -> sendToClient(emitter, entry.getKey(), entry.getValue()));
        }
        return emitter;
    }

    @Scheduled(fixedRate = 180000) // 3분마다 heartbeat 메세지 전달.
    public void sendHeartbeat() {
        Map<String, SseEmitter> sseEmitters = emitterRepository.findAllEmitters();
        sseEmitters.forEach((key, emitter) -> {
            try {
                emitter.send(SseEmitter.event().id(key).name("heartbeat").data(""));
                log.info("하트비트 메세지 전송");
            } catch (IOException e) {
                emitterRepository.deleteById(key);
                log.error("하트비트 전송 실패: {}", e.getMessage());
            }
        });
    }
}

 

◼ listen()

해당 메서드는 @KafkaListener 어노테이션으로 Kafka 에서 해당 토픽에 대한 메세지를 수신하여 SSE 연결된 클라이언트에게 실시간으로 알림을 전달한다.

 

◼ sendToClient()

SSE 연결된 사용자 정보를 받아 메시지를 전송해 준다.

 

◼ addEmitter()

SSE 연결을 추가하는 메서드로 userId와 현재시간을 조합한 emitterId를 만들었다.

이렇게 한이유는 lastEventId라는 값을 인자로 같이 받는것과 연관이 있다.

lastEventId는 SSE연결 요청을 받는 컨트롤러로 부터 전달받는데, 마지막으로 수신한 데이터의 키 값을 의미한다.

만약 단순히 userId로만 저장할 경우에는 네트워크 오류 등의 이유로 연결이 끊어지고 다시 연결이 되었을 때,

기존의 userId와 새로 연결된 userId와 동일하기 때문에 식별할 수가 없다.

그렇기 때문에 유실된 데이터를 정확히 식별하기 위해 userId와 현재시간을 조합해 emitterId를 만드는 것이다.

그리고 lastEventId가 비어있지 않다는 말은 유실된 데이터가 있다는 말이므로

eventCache Map에서 해당 userId로 시작하는 키값들을 찾아 유실된 데이터들을 전송해준다.

 

◼ sendHeartbeat()

 

필자의 경우에는 멀티 페이지 애플리케이션(MPA)이기 때문에 페이지 로드마다 새로 SSE 연결이 생성된다.

그렇기 때문에 새 페이지를 로드하게 되면 기존 SSE 연결은 제거 되지 않은채로 남아 있을 것이다.

이 제거 되지 않은 채로 남아있는 SSE연결들을 끊어내기 위해서  3분 주기로 HearBeat를 보내 사용되지 않는 SSE 연결을 제거하는 방법으로 구현했다.

 

이 메서드에 붙은 @Scheduled 어노테이션을 사용하기 위해서는 애플리케이션 클래스에 아래처럼 @EnableScheduling 어노테이션을 붙여줘야한다.

 

NotificationApiController - SSE 연결 요청을 받을 컨트롤러

@RestController
@Slf4j
@RequiredArgsConstructor
public class NotificationApiController {

    private final EmitterService emitterService;
    private final NotificationsService notificationsService;
    public static final Long DEFAULT_TIMEOUT = 3600L * 1000;

    @GetMapping(value = "/api/sse-connection", produces = "text/event-stream")
    public SseEmitter stream(@Login SessionUser sessionUser, @RequestHeader(value = "Last-Event-ID", required = false, defaultValue = "") String lastEventId) throws IOException {
        return emitterService.addEmitter(String.valueOf(sessionUser.getUserIdNo()), lastEventId);
    }
}

클라이언트로 부터 SSE 연결 요청을 받으면 addEmitter로 SSE 연결을 수행한다.

마지막으로 수신한 데이터를 알기 위해 요청 헤더에서 Last-Event-ID라는 값을 받아오는 것을 볼 수 있다.

중요한점은 SSE 통신을 위해서는 반환할 데이터 타입을 위처럼 "text/event-stream"로 받아야 한다.

 

---

SSE 연결 요청 (클라이언트)

필자의 프로젝트에서 알림 기능은 레이아웃 템플릿의 헤더부분에 적용되어 모든 페이지에 적용된다.

즉, 모든 페이지에는 알림 기능이  존재하고, 이 알림이 실시간으로 업데이트되어야 한다.

단일 페이지 애플리케이션(Single Page Application, SPA) 같은 경우에는 한 번의 SSE 연결로 전체 사이트에서 실시간 업데이트를 받아올 수 있다.

SPA는 새로운 페이지 로드를 발생시키지 않고, 필요한 부분만 동적으로 업데이트하기 때문이다.

 

필자와 같은 멀티 페이지 애플리케이션(Multi Page Application, MPA)인 경우에는 각각의 페이지 로드 시점마다 SSE 연결이 필요하다.

그렇기 때문에 공통 스크립트 파일에 SSE 연결 코드를 작성하여 모든 페이지에서 해당 스크립트를 로드하도록 아래와 같이 작성하였다.

 

let lastHeartbeat = Date.now();

$(document).ready(() => {
	// 로그인 여부 boolean으로 확인 (로그인시에만 SSE연결 요청)
    axios({
        method: "get",
        url: "/api/v1/isLogin"
    }).then(res => {
        if (res.data) {
            startSSE();
        }
    })
});

function startSSE() {
	// 해당 URL로 SSE 연결을 요청 (GET 요청)
    let sse = new EventSource("/api/sse-connection");

	// 더미 데이터(connected!)를 제외한 이벤트가 오면 알림 아이콘에 빨간색 알림이 뜨도록 했다.
    sse.onmessage = (event) => {
        if (event.data != "connected!") {
            $("#showNotifications").append('<span class="notification-dot"></span>');
        }
    };

	// hearbeat를 수신할 때마다 lastHeartbeat 시간을 업데이트
    sse.addEventListener("heartbeat", (event) => {
        lastHeartbeat = Date.now();
    });
    
    
    // 여기 부터 타임아웃 및 에러 발생시 재연결 시도 코드
    let retryCount = 0;
	
    if (retryCount >= 3) {
        return;
    }

    sse.onerror = (event) => {
        if (event.target.readyState === EventSource.CLOSED || Date.now() - lastHeartbeat > 1800000) {
            retryCount++;
            setTimeout(startSSE, 5000); // 5초후 SSE 재연결
        }
    };
}

 

마지막 코드에서

SSE연결 시간 타임아웃이 발생하거나, 마지막으로 받은 hearbeat 시간이 3분을 초과하면 재연결을 시도하도록 했는데

이 코드는 SPA처럼 로그인시에 한번만 SSE 연결을 진행한 후에도 연결을 유지하기 위해서 작성한 코드이다.

 

지금 필자와 같은 MPA라면 해당 코드는 필요 없을 것이다. 

하지만 누군가는 필자처럼 SSE 연결을 유지하기 위해서 여러 고민하며 삽질을 하지 않도록 필요한 사람이 있을 수도 있기에 남겨 둔다.

---

알림 기능이 있는 모든 페이지에 알림 데이터 Model에 담아 전달

모든 페이지에 있는 알림에 컨트롤러마다 알림 데이터를 model에 담아 전달하는 것은 상당히 번거로운 일이다.

반복적인 작업을 방법을 해결하기 위해서 필자는 AOP를 적용하였다.

해당 방법은 아래 포스팅에서 설명한다.  

[Spring AOP] 모든 페이지의 공통 헤더 영역에 공통 데이터 Model에 담아 전달