[Spring Boot, WebSocket, Kafka, MongoDB] 채팅 기능 구현#3. 기본 설정

 

채팅 기능 기본 설정 설명에 앞서, 시스템에서 채팅 기능이 어떻게 동작하는지 로직을 우선 설명하도록 한다.

 

웹소켓 생명주기 관리🥸

---

우리 시스템의 대략적인 웹소켓 연결 과정은 다음과 같다.

 

1. 웹소켓 연결 초기화
   
   • 사용자가 로그인을 완료하면, "ISLOGIN" 쿠키 값을 확인하여 로그인 상태(True)일 경우에 한 번만 웹소켓 연결을 요청한다.
2. 웹소켓 연결 유지 관리
   
   • 웹소켓 연결이 성공적으로 이루어지면, 서버는 사용자의 ID를 웹소켓 세션에 저장한다.
   • 리액트 클라이언트는 웹소켓을 통한 모든 통신 과정을 진행하기 전에 웹소켓 연결 상태를 확인한다. 연결이 끊어진 경우, 자동 재연결 시도와 함께 필요한 토픽 구독을 재요청한다.
3. 토픽 구독 및 메시지 처리
   
   • 사용자는 서버로부터 자신의 ID를 받고, 이를 사용하여 알림 및 채팅방 토픽에 구독 요청을 수행한다. 이는 사용자별 맞춤 알림 및 채팅 기능을 지원한다.
   • 사용자가 채팅방에 입장하면, 리액트 클라이언트는 서버에 채팅 메시지 목록을 요청하고, 메시지 전송 시 MongoDB에 저장한다.
   • 사용자는 구독한 채팅방으로부터 메시지를 받고, 화면에 출력한다.
4. 웹소켓 연결 종료 및 클린업
   
   • 사용자가 채팅을 마치고 채팅방에서 나올 때, 리액트 클라이언트에서는 구독 취소를 수행한다.
   • 로그아웃 시, 사용자는 알림 토픽 구독 취소와 웹소켓 연결 해제를 수행합니다. 이는 보안과 데이터 무결성을 유지하는 데 중요하다.
     
     

시스템에서는 사용자가 로그인을 수행하고 나서 바로 서버로 웹소켓 연결 요청을 수행한다. 그리고 로그아웃 시 웹소켓 연결을 해제한다.

우리 시스템에서 채팅 알림 기능을 웹소켓으로 구현하였다. 그래서 사용자는 자신의 개인 알림 토픽을 구독한 상태여야 알림을 받을 수 있다. 이렇듯 로그인 시 웹소켓을  연결하게 되면 채팅 기능을 사용하지 않더라도 서버에서 사용자 웹소켓 세션을 관리해야 하므로 오버헤드가 발생한다는 단점이 있다. 그럼에도 불구하고 이미 웹소켓 기술을 사용하고 있으며, 편의성으로 웹소켓 방식을 사용하여 구현하게 되었다. 추후 시간이 된다면 SSE 방식을 사용하여 웹소켓 세션 관리에 대한 오버헤드를 줄이고 SSE에 대해서도 학습하고 싶다.

 

---

기본 설정 코드👾

---

웹소켓 관련 설정 코드

---

WebSocketConfig

: 스프링의 웹소켓 지원과 STOMP 메시징 프로토콜을 사용하여, 실시간 양방향 통신을 가능하게 하는 웹소켓 서버를 설정하는 코드

• 어노테이션
  
  • @EnableWebSocketMessageBroker: 웹소켓 서버를 활성화하고, 메시지 브로커를 사용할 수 있도록 설정한다.
• 메서드
  
  • configureMessageBroker(MessageBrokerRegistry registry): 메시지 브로커를 구성한다.
    • '/sub'로 시작하는 경로를 구독 경로로, '/pub'로 시작하는 경로를 메시지 발행 경로로 설정한다.
  • registerStompEndpoints(StompEndpointRegistry registry): 클라이언트가 웹소켓 서버에 연결할 수 있는 엔드포인트를 등록한다.
    • '/stomp/chat' 경로로 엔드포인트를 추가하고, 모든 도메인에서의 접근을 허용하며, 'httpHandshakeInterceptor' 인터셉터를 추가한다. 또한, SockJS를 지원하도록 설정하여 웹소켓이 지원되지 않는 브라우저에서도 통신할 수 있게 한다.
    • 'httpHandshakeInterceptor'에서는 웹소켓 연결 시 사용자 인증 정보를 검증하고 사용자 정보를 웹소켓 세션에 속성으로 저장한다.
  • configureClientInboundChannel(ChannelRegistration registration): 클라이언트로부터 서버로의 인바운드 채널을 구성할 때 사용한다.
    • 'StompHandler' 인터셉터를 등록하여, 웹소켓 연결이 성립될 때와 끊길 때 추가적인 작업(예: 인증, 세션 관리 등)을 수행할 수 있다.
  • configureWebSocketTransport(WebSocketTransportRegistration registry): 웹소켓 통신의 성능과 관련된 설정을 한다.
    • 메시지의 최대 크기를 512KB로 제한하고, 메시지 전송 시간과 전송 버퍼 사이즈를 각각 10초, 512KB로 설정한다.

 

---

웹소켓 연결 및 통신 과정에서의 사용자 인증 방법 의사 결정

현재 우리 시스템에서 사용자 인증 방법으로 JWT를 사용한다. 그리고 해당 토큰은 사용자 웹 브라우저 내 http-only 쿠키에 저장된다.

 

이러한 구조는 리액트 클라이언트에서 직접적으로 토큰에 접근할 수 없게 만들며, 따라서 웹소켓을 통해 채팅 메시지와 같은 데이터를 전송할 때 서버로 토큰을 직접 전송하는 것이 불가능하다.

 

해결 방안 탐색

로컬 스토리지에 토큰을 저장하는 방식 대신, http-only 쿠키에 저장된 토큰을 이용해 인증을 처리하는 방식을 선호한다.

이에 대한 해결 방안으로 다음 세 가지 방안을 고려한다.

1. Spring Session: 세션 정보를 서버 밖의 저장소(예: Redis, JDBC)에 저장하여 여러 인스턴스가 동일한 세션 정보에 접근할 수 있도록 하는 Spring의 프로젝트
   
   • 장점
     • 분산 환경 지원: 여러 서버 인스턴스 간 세션 정보 공유로 부하 분산 및 고가용성 달성
     • 기술 독립성: 다양한 데이터 저장소를 지원하여, 특정 기술에 종속되지 않음
     • 보안 강화: Spring Security와의 통합을 통해 보안 기능 강화
   • 단점
     • 추가 인프라 요구: 외부 저장소(예: Redis) 설정 및 관리 필요
     • 성능 고려: 네트워크를 통한 세션 정보 접근이 필요하므로, 성능에 영향을 줄 수 있음
2. 토큰 기반 인증: 서버가 클라이언트에 인증 토큰(JWT 등)을 제공하고, 클라이언트가 해당 토큰을 사용해 요청마다 인증을 수행하는 방식
   • 장점
     • 상태 비저장: 서버는 상태 유지 필요 없이 토큰만으로 인증 수행
     • 확장성: 세션 정보를 서버에 저장할 필요가 없어, 서버 확장성 증가
   • 단점
     • 토큰 탈취 위험: 토큰이 탈취될 경우, 보안 문제 발생
     • 토큰 관리: 토큰의 생성, 전송, 만료 처리 등 관리 필요
3. 웹소켓 연결 과정에서의 사용자 인증 및 속성 저장: 웹소켓 연결 초기 단계에서 사용자 인증
   • 사용자가 웹소켓 연결 시도 시 인증 정보(토큰)를 서버에 전달하고, 서버는 해당 정보를 검증하여 연결 허용
   • 인증이 성공하면, 사용자의 정보를 웹소켓 세션의 속성으로 저장 - 연결이 유지되는 동안 참조 가능
   • 웹소켓 연결이 유지되는 동안 사용자 인증 상태를 지속적으로 유지할 수 있게 해주므로, 웹소켓 기반의 실시간 통신 어플리케이션에서 자주 사용되는 인증 방식이다.
   • 장점
     • 연결이 유지되는 동안 추가적인 인증 절차 없이 사용자 정보에 접근 가능
   • 단점
     • 인증 정보를 웹소켓 연결 초기에만 검증하기 때문에, 연결이 유지되는 동안 보안 위험 존재
     • 웹소켓 프로토콜 자체에는 인증 메커니즘이 내장되어 있지 않으므로, 별도의 인증 로직을 구현해야 한다.

 

일반적으로 채팅 메시지를 전송할 때마다 사용자 토큰을 검증하는 것은 비효율적일 수있다. 토큰 검증은 서버 자원을 사용하므로, 각 메시지마다 토큰을 검증하면 서버 부하가 증가한다. 이로인해 네트워크 지연을 증가시켜 사용자 경험에도 부정적인 영향을 미친다.

 

우리 시스템의 경우 사용자 간 빠른 응답 시간과 효율적인 서버 자원 관리를 통한 실시간 통신이 중요하다.

 

따라서, 별도의 추가 저장소를 필요로 하지 않고 웹소켓 연결이 처음 이루어질 때 한 번만 사용자를 인증하는 방식인 '웹소켓 연결 과정에서의 사용자 인증 및 속성 저장' 방식을 선택하였다. 해당 방식을 사용하면 매번 토큰을 검증하는 부하를 줄이고, 사용자 인증을 간단하게 처리할 수 있다. 하지만 웹소켓 연결이 유지되는 동안 사용자의 인증 상태가 변경되는 경우 처리하기 어렵다는 단점이 있다.

 

'웹소켓 연결 과정에서 사용자 인증 및 속성 저장 방식'을 다음과 같이 구현하였다.

 

HttpHandshakeInterceptor

: 웹소켓 핸드셰이크(handshake) 과정에서 사용자 인증 처리

• Spring Security와 Spring WebSocket을 함께 사용하면서, 사용자의 JWT(Json Web Token) 기반 인증 정보를 검증한다.
• 핸드셰이크 인터셉터 구현
  
  • beforeHandshake: 이 메서드는 핸드셰이크 과정이 완료되기 전에 호출된다.
    • 사용자의 JWT를 추출하고 검증하여, 인증이 성공하면 사용자의 ID를 attributes 맵에 저장한다.
    • 인증이 실패하면, 에러 메시지를 로깅하고, 토큰 검증 실패를 처리하는 메서드를 호출한 후, false를 반환하여 핸드셰이크 과정을 중단시킨다.
  • extractAccessToken: HttpServletRequest 객체에서 JWT를 추출
  • verifyTokenAndStoreMemberId: 추출한 JWT를 검증하고, 인증이 성공하면 사용자의 ID를 `attributes`에 저장

 

Map<String, Object> attributes
위에서 사용되는 attributes 맵은 웹소켓 핸드셰이크 과정에서 사용되는, 서버와 클라이언트 간의 웹소켓 세션을 구성하는 데 필요한 데이터를 임시로 저장하는 공간이다. 해당 맵은 HandshakeInterceptor 인터페이스의 beforeHandshake 메서드에 파라미터로 전달되며, 웹소켓 연결이 성립되는 동안 해당 세션에 대한 메타데이터나 사용자 정의 데이터를 저장하는 데 사용된다.

 

---

 

StompHandler

: WebSocket을 이용한 채팅 기능에서 메시지를 가공하고 처리하는 역할

• ChannelInterceptor 인터페이스를 구현하여, WebSocket 통신 과정에서 발생하는 다양한 이벤트를 처리한다.
• 어노테이션
  
  • @Order(Ordered.HIGHEST_PRECEDENCE + 99): StompHandler의 실행 우선순위를 설정한다.
    • Spring Security 필터보다 높은 우선순위를 가지지만, 최고 우선순위에서 조금 더 낮게 설정하여 특정 필터들의 실행 후에 동작하도록 한다.
• 메서드
  
  • preSend
    • StompHeaderAccessor를 사용해 메시지의 헤더에 접근하고, STOMP 명령어(예: CONNECT, SUBSCRIBE, UNSUBSCRIBE, DISCONNECT 등)를 확인하여 해당 명령어에 맞는 처리를 수행한다.
  • handleStompCommand: STOMP 명령어 처리
    • SUBSCRIBE, UNSUBSCRIBE, DISCONNECT 등의 명령어에 대한 로직을 처리한다.
    • 예를 들어 CONNECT 시에는 로그를 남기는 등의 간단한 작업을 수행한다.
  • handleSubscribe, handleUnsubscribe, handleDisconnect
    • 각각 구독, 구독 해지, 연결 해제 이벤트를 처리한다.
    • 특히, 채팅방 구독 시에는 채팅방 참여자인지 검증하고, 채팅방에 관련된 알림 메시지를 삭제하는 등의 작업을 수행한다.
    • 구독 해지나 연결 해제 시에는 참여자 목록에서 제거하는 등의 처리를 수행한다.

 

---

카프카 관련 설정 코드

---

본 시스템에서는 성능 향상을 위해 메시지 브로커로 Apache Kafka를 사용한다.

이때, 채팅 및 알림 토픽 각각을 2개의 파티션으로 분리하고 2개의 컨슈머를 할당해 사용한다. (참고: 채팅 기능 성능 개선)

 

ProducerConfig

: Kafka Producer 설정

• 어노테이션
  • @EnableKafka: Kafka 관련 설정을 활성화한다.
• 구성요소
  
  • chatProducerFactory 및 notificationProducerFactory: 각각 채팅 메시지와 알림 메시지를 위한 Kafka ProducerFactory 생성
    • ProducerFactory는 KafkaProducer 인스턴스를 생성하는 데 사용된다.
  • chatProducerConfigurations 및 notificationProducerConfigurations: Kafka Producer 설정을 담은 Map을 반환.
    • Kafka 서버 주소(`BOOTSTRAP_SERVERS_CONFIG`), 키와 값의 직렬화 방법(`KEY_SERIALIZER_CLASS_CONFIG`, `VALUE_SERIALIZER_CLASS_CONFIG`) 등이 포함된다.
    • StringSerializer는 메시지의 키를 문자열로 직렬화하고, JsonSerializer는 메시지의 값을 JSON 형식으로 직렬화한다.
  • chatKafkaTemplate 및 notificationKafkaTemplate
    • Kafka로 메시지를 보내는 작업을 쉽게 만들어 준다.
    • 각각 채팅 메시지와 알림 메시지를 보낼 때 사용되는 KafkaTemplate 인스턴스를 생성한다.

 

---

ListenerConfig

: Kafka 메시지 리스너 설정

• 구성요소
  • kafkaChatContainerFactory와 kafkaNotificationContainerFactory:
    • Message 타입과 Notification 타입 메시지를 처리하기 위한 ConcurrentKafkaListenerContainerFactory를 생성하고 구성한다.
    • 컨슈머 팩토리를 설정하고, 컨슈머의 동시성 수를 2로 설정하여, 동시에 두 개의 스레드가 메시지를 처리할 수 있게 한다.
  • kafkaChatConsumer와 kafkaNotificationConsumer
    • 각각 Message 타입과 Notification 타입 메시지를 처리하기 위한 ConsumerFactory를 생성하고 구성한다.
    • JsonDeserializer: Kafka에서 받은 메시지의 JSON 형식의 값을 Java 객체로 역직렬화를 수행한다.
    • addTrustedPackages("*") 메서드를 호출하여 모든 패키지에서 역직렬화할 클래스를 신뢰하도록 설정한다.
    • consumerConfigurations: Kafka 컨슈머 설정을 담은 Map
      • 부트스트랩 서버 주소, 컨슈머 그룹 ID, 키와 값의 역직렬화 클래스, 오프셋 리셋 정책, 최소 데이터 크기 및 최대 대기 시간 등을 설정한다.
    • DefaultKafkaConsumerFactory: 설정된 값을 바탕으로 Kafka 컨슈머 팩토리를 생성한다.
      • 이 팩토리는 Kafka로부터 메시지를 받아와 처리하는 데 사용된다. 

이 코드를 통해 Kafka 서버로부터 Message와 Notification 타입의 메시지를 효율적으로 받아와 처리할 수 있다.