알림 구현(1) Kafka + Redis Pub/Sub + WebSocket MVP

개요

이번 프로젝트에서 나는 "알림 기능" 파트를 맡게 되었다. 프로젝트 주제는 실시간 게임 매칭 + 게임 공략글 공유 웹서비스이며, 알림은 전반적인 UX의 핵심 기능 중 하나였다.

 

알림은 두 가지 방향으로 나눠서 설계하기로 했다.

RDB 기반 알림: 로그인 시 확인할 수 있는 알림 목록, 읽음 처리 등 상태 저장용, 푸시 알림이 유실되면(상대 사용자가 WebSocket Session이 없으면) 이곳에 모은다.

푸시 기반 알림: 실시간 반응성과 UX 중심의 알림(브라우저 푸시)

우선 RDB 연동 없이 "글에 달린 댓글"의 실시간 알림 스트리밍 기능 MVP부터 구현하였다. 이 글은 그 과정을 정리한 것이다.

 

 

구성

1. 모든 유저는 웹 로그인 시 WebSocketSession을 부여받는다.

클라이언트에서 /ws?token=Bearer...로 WebSocket에 연결한다.

연결되면 유저 세션이 서버에 등록된다.

 

2. 댓글이 작성되면 Kafka로 알림 이벤트를 발행한다.

@TriggerNotification 어노테이션 기반으로 트리거를 자동 처리하고, 댓글 작성 시 글 작성자에게 알림 생성 이벤트를 발생한다.

AOP기반으로 알림 정보를 Kafka로 전송한다.

 

3. Redis Pub/Sub으로 해당 이벤트를 실시간 스트리밍한다.

Kafka Consumer가 이벤트를 수신하고 -> Redis Channel로 Publish한다.

어플리케이션은 해당 Channel을 구독하도록 설계된다.

 

4. 접속 중인 유저에게 WebSocket 알림을 전송한다

서버에서 Redis Pub/Sub 메시지를 수신하고, 현재 접속중인 유저 세션을 찾아 실시간으로 푸시 알림을 제공한다.

 

Kafka를 선택한 이유: 프로젝트는 분당 약 1만건의 게임 매칭 알림 트래픽을 처리해야 한다.

원래는 가벼운 메세지 큐 라이브러리인 RabbitMQ를 사용하려 했지만, RabbitMQ는 메세지 큐 기반이라 큐 수가 많아질 수록 운영과 스케일링에 한계가 있었고, 장기적인 확장성 측면에서도 부족함이 있었다.

반면 Kafka는 파티션 기반으로 분산 처리가 가능해 수평 확장에 유리하고, Throughput이 높아 우리 서비스의 대규모 알림 처리에 적합하다.

또한 실시간 알림 외에도 RDB 등으로 재처리 해야 하는 로직이 필요하기에 확장성이 좀 더 좋은 Kafka를 선택하게 되었다.

부하테스트를 통해 최적의 파티션 전략 등을 수립해볼 예정이다.

 

이 외에 Redis Pub/Sub은 실시간 메시지 브로드캐스트에 적합하고, 구조도 단순하며 딜레이가 거의 없어 실시간 알림 시스템에 매우 잘 맞는다.
또한 개인적으로 Redis에 익숙하여 빠르게 안정적인 구조를 구성할 수 있었기 때문에 채택하게 되었다.

WebSocket 역시 Spring에서 기본적으로 제공하는 기능을 활용할 수 있었고, 과거에 한 번 사용해본 경험이 있어 비교적 쉽게 구현에 들어갈 수 있었다.
이처럼 익숙한 도구를 기반으로 구현 속도를 높이되, 확장성과 운영 관점에서도 충분히 대응 가능한 선택이라고 판단했다.

 

 

흐름

여기서 JwtFilter까지 알림 서비스에 사용된다.

UML 다이어그램으로 표현하자면,

실시간 Notification 처리 로직(Kafka -> Redis -> WebSocket)은 이렇게 구성했다.

가장 핵심이 되는 곳이 Produce -> Consume인데,

@Around("@annotation(triggerNotification)")
public Object notify(ProceedingJoinPoint joinPoint, TriggerNotification triggerNotification) throws Throwable {
    Object result = joinPoint.proceed();
    Object[] args = joinPoint.getArgs();

    Long toMemberId = extractToMemberId(args);
    Long fromMemberId = extractFromMemberId(args);
    Long relatedId = extractCommentId(result);

    if (toMemberId.equals(fromMemberId)) return result;

    NotificationEvent event = NotificationEvent.builder()
            .target(triggerNotification.target())
            .fromMemberId(fromMemberId)
            .toMemberId(toMemberId)
            .relatedId(relatedId)
            .content("새 댓글이 달렸습니다.")
            .build();

    kafkaTemplate.send("notification.comment", event);
    log.info("댓글 알림 전송: {}", event);

    return result;
}

현재는 Comment만 처리하고 있어서 구조가 깔끔하지만 나중에 다른 알림도 같은 형태로 처리할 수 있게끔 확장하려면 어떻게 코드를 작성해야 할지 좀 고민된다.

Topic 별로 클래스를 나눌 전략을 잘 짜야 할 거 같다.

@Component
@RequiredArgsConstructor
public class NotificationListener implements MessageListener {

    private final WebSocketSessionManager sessionManager;

    @Override
    public void onMessage(Message message, byte[] pattern) {
        String channel = new String(message.getChannel());
        String body = new String(message.getBody());
        Long memberId = extractMemberIdFromChannel(channel);

        sessionManager.sendNotificationToMember(memberId, body);
    }

    private Long extractMemberIdFromChannel(String channel) {
        return Long.parseLong(channel.replace("noti:member:", ""));
    }
}

또한 Listener쪽에서도 해당 ToMember에게 webSocket 세션이 없으면 fallback 비슷하게 RDB에 저장하는 로직이 필요하다.

Producer에서 미리 RDB에 모두 저장해두고 처리할지, sessionManager에서 확인 후 세션이 존재하지 않으면 DB에 저장할지는 팀원들과 상의 후 결정해야겠다.

아무래도 미리 저장하는 로직이 좋지 않을까 싶다. session이야 뭐 없으면 없는 대로 유실 처리 하면 되고..

WebSocket 처리 로직(클라이언트 연결 -> 세션 등록)은 이렇게 구성했다.

filter에서 /ws로 요청이 들어오면 handler에서 이를 인식해 세션을 add한다.

현재는 HashMap 기반 Cache로 운영중이라 분산 환경에서 어떻게 관리할지 좀 고민해봐야 한다.

일단 WebSocket King Client로 테스트 결과, MVP 구현은 성공이당

 

 

이후 구현사항 + 회고

1. RDB에 알림 저장 및 Fallback 전략

실시간 알림 외에도 알림 목록 조회, 읽음 처리 등을 구현 가능하다.

접속 중이 아닌 유저에게는 RDB에 저장된 알림을 토대로, 프론트엔드에서 뱃지 형태로 쌓이게 구현할 수 있을 것이다.

WebSocket 연결 실패나 메세지 유실 시 fallback 전략도 세워야 한다.

적당히 부하테스트 후에, fallback 전략 세우고, 파티션 분리 하면 될 듯 하다.

 

2. 다중 접속 지원 / 분산 환경

하나의 유저가 여러 브라우저나 여러 기기로 접속해도 모두 실시간 알림을 받을 수 있어야 한다.

현재 WebSocket은 단일 세션만 지원해서, 다른 세션에서 접속 시 원래 세션의 연결이 끊긴다.

또한 Session 만료 시간 등등 생각할 게 많다. 분산 환경에서도 어떻게 동작할지 모르겠다

 

3. 보안 강화

WebSocket 보안을 좀 공부하고 찾아봐야 한다. 지금 그냥 ws://프로토콜에, 엔드포인트도 /ws처럼 단순하게 쓰고있고 http연결이라 탈취 위험이 존재한다.

Redis Pub/Sub 채널명도 암호화 해야겠다. 암호화 함으로써 생기는 오버헤드도 고려해보고..

 

-_- 할거많다!