실시간 채팅 구현기(3) 코드 분석

지난 시간에 이어 코드를 분석해보는 시간을 갖겠다.

우선 spring에서 웹소켓을 사용하기 위한 설정부터 알아보자.

웹소켓 통신 구현을 위한 필수적인 코드만 살펴보겠다.

 

WebSocketConfig

@Configuration
@EnableWebSocketMessageBroker // STOMP 메시징 활성화
@RequiredArgsConstructor
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {

• @EnableWebSocketMessageBroker : STOMP 프로토콜을 이용한 메시징 인프라를 활성화하고 웹소켓 연결을 허용한다. 이를 통해 @MessageMapping 어노테이션을 컨트롤러에서 사용할 수 있게된다.

    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry.addEndpoint("/ws")
                .setAllowedOriginPatterns("*")
                .withSockJS();

        // 엔드포인트에 핸들러 연결
        registry.setErrorHandler(stompErrorHandler);
    }

• addEndpoint("/ws") : 핸드쉐이크를 시도할 주소 "ws://{url}/ws"와 같은 주소로 접속하면 웹소켓 연결을 시도한다.
• setAllowedOriginPatterns("*") : CORS 설정. 나는 원활한 개발을 위해 모든 도메인에서 접속할 수 있도록 열어두었다.
• withSockJS() : SockJS 기술을 적용하여 웹소켓을 지원하지 않는 브라우저에서도 polling 방식으로 수행할 수 있도록 설정

    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
        registry.enableSimpleBroker("/sub");
        registry.setApplicationDestinationPrefixes("/pub");
    }

• enableSimpleBroker("/sub") : 메시지를 구독하는 경로의 접두사 지정
• setApplicationDestinationPrefixes("/pub") : 클라이언트가 메시지를 발행할 때 사용하는 접두사. /pub/chat/message 처럼 /pub으로 시작하는 메시지는 컨트롤러의 @MessageMapping으로 라우팅된다.

 

StompHandler

@Component
@RequiredArgsConstructor
@Slf4j
public class StompHandler implements ChannelInterceptor {

• ChannelInterceptor : Spring WebSocket(STOMP) 메시지가 오갈 때 중간에서 가로채는 인터셉터

    @Override
    public Message<?> preSend(Message<?> message, MessageChannel channel) {
        StompHeaderAccessor accessor = StompHeaderAccessor.wrap(message);
        StompCommand command = accessor.getCommand();

        if (StompCommand.CONNECT.equals(command)) {
            // CONNECT 로직
        }

        if (StompCommand.SUBSCRIBE.equals(command)) {
            // SUBSCRIBE 로직
        }

        if(StompCommand.SEND.equals(command)) {
            // SEND 로직
        }

        return MessageBuilder.createMessage(message.getPayload(), accessor.getMessageHeaders());
    }

• 메시지를 받기 전 가로채서 선취적으로 실행되어야하는 로직 구현
• 나는 웹소켓 요청이 들어왔을 때도 인증된 사용자만 사용할 수 있도록 JWT 인증을 넣어줬다.

 

StompErrorHandler

@Component
public class StompErrorHandler extends StompSubProtocolErrorHandler {

• StompSubProtocolErrorHandler : HTTP의 경우 @ControllerAdvice, @ExceptionHandler로 예외 처리가 가능하지만 STOMP는 메시지 기반으로 에러 프레임을 별도로 구현해줘야 한다. StompSubProtocolErrorHandler가 그 역할을 한다. Spring이 내부적으로 WebSocket 에러 발생 시 이 클래스를 호출한다.

@Override
public Message<byte[]> handleClientMessageProcessingError(Message<byte[]> clientMessage, Throwable ex) {}

• handleClientMessageProcessingError : 클라이언트 메시지 처리 중 에러 발생하면 호출된다.

다음과 같은 구조다.

클라이언트 SEND
   ↓
Controller / Handler
   ↓ (예외 발생)
StompErrorHandler
   ↓
ERROR 프레임 생성
   ↓
클라이언트로 전달

 

StompController

@Controller
public class StompChatController {
    
    @MessageMapping("/chat/message")
    public void message(MessageRequest request, Message<?> message) {
        StompHeaderAccessor accessor = StompHeaderAccessor.wrap(message);
        // service 메서드 호출
    }
}

• HTTP 요청은 @GetMapping과 같이 요청을 매핑하는 것처럼 STOMP 요청은 @MessageMapping으로 요청을 매핑한다.

 

RedisMessageBrokerService

public void publish(String topic, Object message) {
	// message 파싱
    redisTemplate.opsForStream().add(topic, message); // 레디스 스트림에 해당 레코드 발행
}

public void subscribe(String topic, String consumerGroup, MessageHandler handler) {
    redisTemplate.opsForStream().createGroup(topic, consumerGroup); // 토픽을 구독하는 컨슈머그룹 등록
	
    // 비동기 options 로직

    var container = StreamMessageListenerContainer.create(connectionFactory, options);
    container.receive(
            Consumer.from(consumerGroup, "my-instance"),
            StreamOffset.create(topic, ReadOffset.lastConsumed()), // 마지막으로 소비한 메시지부터 받아오기
            message -> {
                handler.handle(message.getValue());
            }
    );

    container.start();
}

• publish : redisTemplate.opsForStream().add()를 통해 레코드 발행
• subscribe : redisTemplate.opsForStream().createGroup()을 통해 컨슈머 그룹을 만든다. 이미 존재하는 이름이라면 예외가 발생한다. 
• container : Listener Container에 대한 옵션을 설정한다. 아래의 공식문서를 참고하자. 중요한 설정은 Offset 설정이다. ReadOffset.lastConsumed()로 설정해뒀는데 이는 마지막으로 소비한 메시지 다음부터 받아오는 설정이다.
  https://docs.spring.io/spring-data/redis/docs/current/api/org/springframework/data/redis/stream/StreamMessageListenerContainer.html

StreamMessageListenerContainer (Spring Data Redis 4.0.4 API)

 

MessageBrokerChatService

@PostConstruct
public void init() {
    // 메시지 브로커에서 STREAM_TOPIC, CONSUMER_GROUP를 기준으로 메시지 구독
    messageBrokerService.subscribe(STREAM_TOPIC, CONSUMER_GROUP, this::dispatchStreamMessageToClients);
}

private void dispatchStreamMessageToClients(Map<String, String> message) {
    String roomId = message.get(ROOM_ID);
    messagingTemplate.convertAndSend(SUBSCRIPTION_URL + roomId, message);
    chatService.saveMessageToMongo(message);
}
    
public void sendChatMessageToRedisStreams(Long roomId, Long senderId, String nickname, String inputMessage) {
    // message 객체 파싱

    messageBrokerService.publish(STREAM_TOPIC, message);
}

• init() : 실시간 채팅서비스에서 채팅 메시지를 메시지 브로커로 처리할 예정이기 때문에 서버를 실행할때 처음 메시지 구독을 신청해주는 과정이 필요하다.
• dispatchStreamMessageToClients() : 구독과 동시에 메시지를 받았을 때 실행되어야할 콜백함수.
• sendChatMessageToRedisStreams() : 발행 메시지 파싱

현재는 단일 토픽, 단일 컨슈머 그룹을 설정했는데 향후 스케일을 키운다면 방의 숫자에 따라 topic을 구분하고, 컨슈머 그룹을 각 토픽마다 하나씩 두는 방식으로 구현할 것이다. 같은 방에서는 한 서버의 topic을 구독하고 있어야한다. 그리고 컨슈머 그룹은 각 컨슈머가 경쟁 소비 방식으로 가져가기 때문에 모두가 같은 메시지를 받아봐야하는 채팅 서비스에선 컨슈머 그룹은 토픽당 하나다.