[WebSocket] Spring Boot + STOMP + Redis Pub/Sub 이용한 채팅 서버 구현

WebSocket?

WebSocket(웹 소켓)이란 HTTP 환경을 기반으로 하여 TCP/IP 연결을 통해 전이중 통신(양방향 송수신) 채널을 제공하는 컴퓨터 통신 프로토콜이다. WebSocket의 접속 과정은 TCP/IP 접속, 웹 소켓을 열기 위한 HandShake 과정으로 나뉠 수 있다.

WebSocket 최초 연결 과정을 나타내면 아래와 같다.

1. 서버와 클라이언트 간 TCP/IP 연결을 수립한다.
   
   
2. HTTP 요청 기반의 HandShake 과정을 거친다. (HTTP Request 헤더에 Upgrade, Connection를 추가하여 WebSocket 요청임을 표시)
   
   
3. HandShake 연결이 끝나면, HTTP 프로토콜을 WebSocket 프로토콜로 변환하여 웹 소켓 통신을 시작한다.

웹 소켓의 최초 접속은 HTTP Request를 통해 HandShake 과정을 거치므로 기존 HTTP 규격이나 인증을 그대로 가져올 수 있고, 기존 80, 443 포트로 접속을 하므로 추가 방화벽에 대한 설정이 필요 없다는 것이 장점이다.

아래 링크에 들어가보면 웹 소켓 탄생 배경, HandShake 과정에 대해 정말 자세하고 쉽게 설명해주신다.
https://tecoble.techcourse.co.kr/post/2021-08-14-web-socket/

STOMP?

우리가 WebSocket 프로토콜만을 사용해서 채팅 서버를 구현한다 가정해보면, 메시지 포맷 형식이나 메시지 통신 과정, 세션 등을 일일이 관리해야하는 번거러움이 있다. 따라서 메시징 처리에 최적화 시키기 위해 STOMP(Simple Text Oriented Message Protocol) 프로토콜을 서브 프로토콜로 사용하게 된다.

STOMP는 메시지 송수신을 효율적으로 하기 위해 나온 프로토콜이며 WebSocket 프로토콜 위에서 동작한다. 이렇게 보면 WebSocket이 하위 프로토콜, STOMP가 상위 프로토콜로 생각해도 될 것 같다.

기본적으로 pub/sub 구조로 되어있어 메시지 송신이나 수신 처리하는 부분이 확실히 정의되어 있기 때문에 개발자 입장에선 메시징 처리 할 때 STOMP 스펙의 규칙만 잘 지키면 된다는 이점이 있다.

STOMP Frame

STOMP 메시지 형식(Frame)은 아래와 같다. (stomp-specification 참고)
즉, 아래의 Frame 구조로 Client-Server 간의 메시지 송수신이 이루어진다.

COMMAND
header1:value1
header2:value2

Body^@

• 클라이언트는 메시지를 발송할 때 COMMAND로 SEND 또는 SUBSCRIBE 등의 명령을 사용할 수 있다.
  
  
• 그리고 header:value 형식으로 누가 받을지에 대한 정보와 메시지 정보 등이 포함된다.

Frame에 대한 예시는 위 stomp-specification 링크에 나와있다.

즉, 위 형식에 메시지 송수신에 필요한 정보를 담게되고, STOMP 플로토콜은 pub/sub 방식을 사용하여 Message Broker를 통해 특정 작업을 수행하거나 메시지를 보내게 된다.

Pub/Sub 방식과 Message Broker

pub/sub는 메시지를 공급하는 객체(publisher)와 소비하는 객체(subscriber)를 분리해 제공하는 비동기식 메시징 방법이다. publisher가 특정 topic에 메시지를 보내면 해당 topic을 구독해놓은 모든 subscriber에게 메시지가 전송되는 방식이다.

간단하게 유튜브 채널이나 뉴스 구독을 생각하면 금방 이해할 수 있을 것이다.

Pub/Sub 구조는 정말 다양하게 사용되는데 간단하게 이번에 구현할 채팅부터 시작해서 MSA(한 API에서 다른 API로의 전달)에서 까지 폭넓게 사용되며, 아래 링크에서 자세한 내용을 확인할 수 있다.
https://www.bmc.com/blogs/pub-sub-publish-subscribe/

pub/sub는 위 그림에서 보듯이 서로 직접 통신하는 것이 아닌 Message Broker를 통해 메시지를 전달하는 구조이다. Message Broker는 Publisher가 보낸 메시지를 Subscriber로 전달해주는 중간 다리(미들웨어) 역할을 한다.

왜 중간에 Message Broker가 필요한지에 대한 이유는 다음과 같다.

• 느슨한 결합 (Publisher는 메시지를 발신할 때 다른 서비스들에 대해 알 필요가 전혀 없음, scale-out 용이)
  
  
• 메시지 버퍼링 (Subscriber 측에서 원하는 시점에 메시지를 처리할 수 있음)

아무래도 브로커를 통해 통신을 하기 때문에 이 과정에서 부하가 발생하거나 브로커리스 시스템에 비해 속도가 느릴 수 있다는 단점이 존재한다!

Spring Boot - STOMP 구현 해보기

이제 개념을 파악했으니 구현에 들어가보도록 하자!

#1 build.gradle

	implementation 'org.springframework.boot:spring-boot-starter-websocket'

• build.gradle에 소켓 관련 의존성을 추가시킨다.

#2 WebSocketConfig.java

WebSocketConfig 클래스에서는 Message Broker에 대한 설정을 한다.

@Configuration
@EnableWebSocketMessageBroker 
@RequiredArgsConstructor
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer { // (1)

    private final StompHandler stompHandler;

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

    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) { // (5)
        registry.addEndpoint("/stomp/chat") // ex ) ws://localhost:9000/stomp/chat
                .setAllowedOriginPatterns("*").withSockJS(); 
    }

    @Override // (6)
    public void configureClientInboundChannel (ChannelRegistration registration){
        registration.interceptors(stompHandler);
    }
}

(1) WebSocketMessageBrokerConfigurer를 상속받아 STOMP로 메시지 처리 방법을 구성한다.

(2) configureMessageBroker에서는 메시지를 중간에서 라우팅할 때 사용하는 메시지 브로커를 구성한다.

(3) enableSimpleBroker에서는 해당 주소를 구독하는 클라이언트에게 메시지를 보낸다. 즉, 인자에는 구독 요청의 prefix를 넣고, 클라이언트에서 1번 채널을 구독하고자 할 때는 /sub/1 형식과 같은 규칙을 따라야 한다.

(4) setApplicationDestinationPrefixes에는 메시지 발행 요청의 prefix를 넣는다. 즉, /pub로 시작하는 메시지만 해당 Broker에서 받아서 처리한다.

(5) 클라이언트에서 WebSocket에 접속할 수 있는 endpoint를 지정한다.

(6) (무시하셔도 됩니다) 사용자가 웹 소켓 연결에 연결 될 때와 끊길 때 추가 기능(인증, 세션 관리 등)을 위해 인터셉터를 걸어주었다. 인자에는 추가 기능을 구현한 StompHandler를 빈으로 등록하여 넣어주었다.

In-Memory 기반 Message Broker 문제점

사실 Spring에서 제공하는 STOMP를 활용하고도, 내장된 Simple Message Broker를 사용해 채팅 서버를 구현할 수 있다. 하지만 Simple Message Broker 같은 경우 스프링 부트 서버의 내부 메모리에서 동작하게 된다.

인메모리 기반 브로커를 사용했을 때의 문제점은 다음과 같다.

• 서버가 down되거나 재시작을 하게되면 Message Broker(메시지 큐)에 있는 데이터들은 유실될 수 있다.
  
  
• 다수의 서버일 경우 서버간 채팅방을 공유할 수 없게 되면서 다른 서버간에 있는 사용자와의 채팅이 불가능 해진다.

이러한 문제들을 해결하기 위해서는 Message Broker가 여러 서버에서 접근할 수 있도록 개선이 필요하다. 즉, 외부 메시지 브로커를 연동한다면 인프라 비용은 좀 더 증가하겠지만 위 문제들을 해결할 수 있게 된다!

Redis Pub/Sub 적용하기

Redis는 STOMP 프로토콜을 지원하지 않지만, Redis가 제공하는 Pub/Sub 기능을 통해 메시지 브로커로 사용할 수 있다.

몰론 STOMP 프로토콜을 지원하는 RabbitMQ와 같은 전용 메시지 브로커를 사용하면 더 고도화된 기능(메시지 전달 보장, SSL 지원)을 사용할 수 있다.

하지만 채팅 정보에 대한 세션을 관리할 key-value 데이터베이스가 필요했고, 더 빠르고 쉽게 구현할 수 있는 Redis를 선택하게 되었다.

다중 서버에서 Redis Pub/Sub를 이용하면 아래 그림과 같을 것이다.
(제 생각대로 나타낸 것이라 틀린 부분이 있을 수 있습니다)

먼저 구현에 들어가기 앞서, 본인의 PC에 Redis가 설치되있거나 docker를 통해 redis container가 구동되어 있어야만 한다.

#1 build.gradle

implementation 'org.springframework.boot:spring-boot-starter-data-redis'

• build.gradle에 Redis 관련 의존성을 추가한다.

#2 RedisConfig.java

@Configuration
public class RedisConfig {
    @Bean
    public RedisMessageListenerContainer redisMessageListenerContainer( // (1)
            RedisConnectionFactory connectionFactory,
            MessageListenerAdapter listenerAdapter,
            ChannelTopic channelTopic
    ) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.addMessageListener(listenerAdapter, channelTopic);
        return container;
    }
    
    @Bean
    public MessageListenerAdapter listenerAdapter(RedisSubscriber subscriber) { // (2)
        return new MessageListenerAdapter(subscriber, "onMessage");
    }

    @Bean
    public RedisTemplate<String, Object> redisTemplate
    (RedisConnectionFactory connectionFactory) { // (3)
        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(connectionFactory);
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(new Jackson2JsonRedisSerializer<>(String.class));
        return redisTemplate;
    }

    @Bean
    public ChannelTopic channelTopic() { // (4)
        return new ChannelTopic("chatroom");
    }
}

(1) RedisMessageListenerContainer는 Redis Channel(Topic)로 부터 메시지를 받고, 주입된 리스너들에게 비동기적으로 dispatch 하는 역할을 수행하는 컨테이너이다. 즉, 발행된 메시지 처리를 위한 리스너들을 설정할 수 있다.

(2) MessageListenerAdaper에서는 RedisMessageListenerContainer로부터 메시지를 dispatch 받고, 실제 메시지를 처리하는 비즈니스 로직이 담긴 Subscriber Bean을 추가해준다.

(3) Redis서버와 상호작용하기 위한 RedisTemplate 관련 설정을 해준다. Redis 서버에는 bytes 코드만이 저장되므로 key와 value에 Serializer를 설정해준다. Json 포맷 형식으로 메시지를 교환하기 위해 ValueSerializer에 Jackson2JsonRedisSerializer로 설정해준다.

(4) Topic 공유를 위해 Channel Topic을 빈으로 등록해 단일화 시켜준다.

#3 RedisSubscriber.java

@Slf4j
@RequiredArgsConstructor
@Service
public class RedisSubscriber implements MessageListener {
    private final ObjectMapper objectMapper;
    private final RedisTemplate redisTemplate;
    private final SimpMessageSendingOperations messagingTemplate;
 
    @Override
    public void onMessage(Message message, byte[] pattern) {
        try {
            String publishMessage = (String) redisTemplate.getStringSerializer().deserialize(message.getBody());

            ChatMessageRequest roomMessage = objectMapper.readValue(publishMessage, ChatMessageRequest.class);

            if (roomMessage.getType().equals(MessageType.TALK)) {
                GetChatMessageResponse chatMessageResponse = new GetChatMessageResponse(roomMessage);
                messagingTemplate.convertAndSend("/sub/chat/room/" + roomMessage.getRoomId(), chatMessageResponse);
            }
            
            ...
            
        } catch (Exception e) {
            throw new ChatMessageNotFoundException();
        }
    }
}

• onMessage 메소드는 리스너에 수신된 메시지를 각 비즈니스 로직을 거쳐 messagingTemplate을 이용해 WebSocket 구독자들에게 메시지를 전달하는 메소드이다.

• Redis로부터 온 메시지를 역직렬화하여 ChatMessage DTO로 전환뒤 필요한 정보와 함께 메시지를 전달한다.

#4 ChatController.java & ChatService.java

• ChatController

@RequiredArgsConstructor
@Controller
public class ChatMessageController {

    private final ChatMessageService chatMessageService;
    private final UserRepository userRepository;

    @MessageMapping("/chat/message")
    public void message(
            ChatMessageRequest chatMessageRequest
    ) {
        User user = userRepository.findById(chatMessageRequest.getUserId()).orElseThrow(UserNotFoundException::new);

        chatMessageService.sendMessage(chatMessageRequest, user);
    }
}

• ChatService.sendMessage()

    @Transactional
    public void sendMessage(ChatMessageRequest chatMessageRequest, User user) {
        ChatRoom chatRoom = chatRoomRepository.findById(chatMessageRequest.getRoomId()).orElseThrow(ChatRoomNotFoundException::new);

        //채팅 생성 및 저장
        ChatMessage chatMessage = ChatMessage.builder()
                .chatRoom(chatRoom)
                .user(user)
                .message(chatMessageRequest.getMessage())
                .build();

        chatMessageRepository.save(chatMessage);
        String topic = channelTopic.getTopic();

        // ChatMessageRequest에 유저정보, 현재시간 저장
        chatMessageRequest.setNickName(user.getNickname());
        chatMessageRequest.setUserId(user.getId());

        if (chatMessageRequest.getType() == ChatMessageRequest.MessageType.TALK) {
            // 그륩 채팅일 경우
            redisTemplate.convertAndSend(topic, chatMessageRequest);
        }
    }