📲 실시간 채팅을 위한 웹소켓과 STOMP

웹소켓(WebSocket) 은 클라이언트와 서버 간의 전이중(양방향) 통신을 제공한다. HTTP 통신은 기본적으로 비연결성 통신으로, 클라이언트에게 한번 보내고 나면 연결이 끊겨 지속적으로 데이터를 주고 받을 수 없다. 그러나 WebSocket 은 HTTP Handshake 를 통해 처음에 연결을 맺으면 그 연결을 계속 유지하는 연결지향성을 보인다.

웹소켓만을 사용해서 채팅 서버를 구현하다면, 메시지 포맷 형식이나 메시지 통신 과정 등을 관리해야 하는 번거로움이 있다. 따라서 이러한 관리를 대신해줄 수 있는 STOMP 프로토콜을 서브 프로토콜로 사용할 수 있다.

STOMP 는 Simple Text Oriented Messaging Protocol 의 약자로, 메시지의 형식이나 내용 등을 정의하여 메시징 전송을 효율적으로 도와주는 프로토콜이다. 기본적으로 Pub/Sub 구조로 되어있어 메시지 송/수신 처리는 정의가 되어 있다. 따라서 개발자는 STOMP 규칙에 맞춰 메시징 처리를 재정의하여 사용할 수 있다는 이점이 있다.

💡 여기서 Pub/Sub 이란 메시지를 공급하는 객체(Publisher) 와 소비하는 객체(Subscriber) 를 말한다. Publisher 이 특정 Topic 에 메시지를 보내면 해당 topic 을 구독한 모든 Subscriber 에게 메시지가 전송되는 방식이다.

Pub/Sub 은 서로 직접 통신하지 않고 메시지 브로커(Message Broker) 를 통해 메시지를 전달한다. Message Broker 는 Publisher 가 보낸 메시지를 Subscriber 로 전달해주는 중간 다리 역할을 한다.

직접 통신하지 않고 Message Broker 를 두는 이유는 다음과 같다.
1. Publisher 가 메시지를 발신할 때 다른 서비스들에 대해 알 필요가 없기 때문에 scale-out이 용이하다.
2. Subscriber 에서 원하는 시점에 메시지를 처리할 수 있다.

구현

1. WebSocket

• 의존성 주입 (build.gradle)

dependencies { 	
	...

	// WebSocket
	implementation 'org.springframework.boot:spring-boot-starter-websocket'
}

• WebSocketConfig.java

@Configuration
@EnableWebSocket    // 웹소켓 서버 사용
@EnableWebSocketMessageBroker   // STOMP 사용
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {

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

    @Override
    public void registerStompEndpoints(final StompEndpointRegistry registry) {
        registry
                .addEndpoint("/ws")     // 엔드포인트: /ws
                .setAllowedOrigins("*");
    }

}

• MessageController.java

@Controller
@RequiredArgsConstructor
public class MessageController {

    private final SimpMessageSendingOperations simpMessageSendingOperations;

    /**
     *  클라이언트가 /pub/hello 로 메시지를 발행한다.
     *
     *      /sub/channel/channelID     - 구독
     *      /pub/hello                 - 메시지 발생
     */
    @MessageMapping("/hello")
    public void message(final Message message) {

        // - 해당 채널ID 에 메시지를 보낸다.
        // - 그리고 "/sub/channel/channelID" 에 구독 중인 클라이언트에게 메시지를 보낸다.
        simpMessageSendingOperations.convertAndSend("/sub/channel/" + message.getChannelId(), message);
    }
}

😮‍💨 In-Memory 기반 메시지 유실 가능성이 있다.

Spring 에 내장된 Simple Message Broker 는 SpringBoot 서버 내부 메모리에서 동작한다. 만약 메시지 발행 시 서버가 다운 되어 메시지 전송을 실패하게 된다면, 인메모리 기반으로 동작하는 메시지 큐(Message Broker)로 인해 메시지를 유실할 가능성이 매우 높다. 더불어 인메모리 기반 시스템은 메시지 모니터링이 쉽지 않다.

이러한 문제들을 해결하기 위해 외부 브로커를 사용하는 방법이 있다. 외부 메시지 브로커를 사용한다면 인프라 비용이 들겠지만, 여러 상황을 고려했을 때 외부 브로커를 연동하는 방법이 효율적일 수 있다.

💡 외부 브로커는 RabbitMQ 나 Redis 의 Pub/Sub, Kafka 등이 있다.

• RabbitMQ 는 다양한 비즈니스에 의한 복잡한 라우팅 설계에 적합한 메시지 브로커이다. 신뢰성 있는 메시지를 전송할 수 있다.
• Redis의 Pub/Sub 은 다른 메시지 브로커와 다르게 메시지 지속성이 없다. 즉, 메시지를 전송한 후 해당 메시지는 삭제되며 Redis 어디에도 저장되지 않는다. 실시간 데이터 처리에 매우 적합하지만, 메시지가 저장되지 않는다. 또한, 메시지 전송 신뢰성을 보장하지 않기 때문에 단점을 보완할 별도의 추가 구현이 필요할 수 있다.
• Kafka 는 대량의 데이터를 저장하면서 높은 처리량이 필요에 적합한 메시지 브로커이다.
  
  

✅ 채팅 정보에 대한 세션을 관리할 key-value 데이터베이스를 사용하고자 Redis 를 선택했다. Redis 는 STOMP 프로토콜을 지원하지 않지만, Redis 가 제공하는 Pub/Sub 기능을 통해 메시지 브로커를 사용할 수 있다.

2. Redis Pub/Sub

💡 참고로 Redis 는 Docker 를 사용했다.
각 모듈마다 docker-compose.yml 파일을 생성해 포트와 컨테이너를 지정해주었다.

• 의존성 주입 (build.gradle)

dependencies { 	
	...

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

• docker-compose.yml

# 파일 규격 버전
version: "3.1"

# 실행하려는 컨테이너들 정의
services:
  # 서비스명
  redis_container:
    # 사용할 이미지
    image: redis:latest
    # 컨테이너명
    container_name: redis-pub-container
    # 접근 포트 설정(컨테이너 외부:컨테이너 내부)
    hostname: test
    ports:
      - 6380:6379
    # 스토리지 마운트(볼륨) 설정
    volumes:
      - ./redis/data:/data
      - ./redis/conf/redis.conf:/usr/local/conf/redis.conf
    # 컨테이너에 docker label을 이용해서 메타데이터 추가
    labels:
      - "name=redis"
      - "mode=standalone"
    # 컨테이너 종료시 재시작 여부 설정
    restart: always
    command: redis-server /usr/local/conf/redis.conf

Publisher

• application.yml

server:
  port: 8082

spring:
  redis:
    host: localhost
    port: 6380

• RedisConfig.java

@Configuration
public class RedisConfig {

    @Bean
    public RedisConnectionFactory redisConnectionFactory() {
        final LettuceConnectionFactory lettuceConnectionFactory = new LettuceConnectionFactory();
        return lettuceConnectionFactory;
    }

    @Bean
    public RedisTemplate<String, String> stringValueRedisTemplate() {
        final RedisTemplate<String, String> redisTemplate = new RedisTemplate<>();

        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(new StringRedisSerializer());
        redisTemplate.setConnectionFactory(redisConnectionFactory());

        return redisTemplate;
    }
}

• RedisPubApplication.java

@SpringBootApplication
@RequiredArgsConstructor
public class RedisPubApplication implements CommandLineRunner {

    private final RedisTemplate<String, String> stringValueRedisTemplate;

    public static void main(String[] args) {
        SpringApplication.run(RedisPubApplication.class);
    }

    @Override
    public void run(final String... args) {

        // TEST: String 메시지 전송
        stringValueRedisTemplate.convertAndSend("ch01", "Apple, Orange");
    }
}

Subscriber

• application.yml

server:
  port: 8081

spring:
  redis:
    host: localhost
    port: 6379

• RedisConfig.java

@Configuration
public class RedisConfig {

    @Bean
    public RedisConnectionFactory redisConnectionFactory() {
        final LettuceConnectionFactory lettuceConnectionFactory = new LettuceConnectionFactory();
        return lettuceConnectionFactory;
    }

    /**
     * 메시지 발송을 위한 RedisTemplate
     */
    @Bean
    public RedisTemplate<String, String> stringValueRedisTemplte() {
        final RedisTemplate<String, String> redisTemplate = new RedisTemplate<>();

        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(new StringRedisSerializer());
        redisTemplate.setConnectionFactory(redisConnectionFactory());

        return redisTemplate;
    }

    /**
     * 메시지 수신을 위한 설정: RedisMessageListenerContainer
     *
     * Redis 의 channel 로부터 메시지를 수신받아 해당 MessageListenerAdapter 에게 디스패치
     */
    @Bean
    public RedisMessageListenerContainer redisContainer() {
        final RedisMessageListenerContainer container = new RedisMessageListenerContainer();

        container.setConnectionFactory(redisConnectionFactory());
        container.addMessageListener(messageStringListener(), channelTopic());

        return container;
    }

    /**
     * 메시지 수신을 위한 설정: MessageListenerAdapter
     *
     * RedisMessageListenerContainer 로부터 메시지를 받아서 등록된 리스너(구독자)에게 전달
     */
    @Bean
    public MessageListenerAdapter messageStringListener() {
        return new MessageListenerAdapter(new RedisMessageStringSubscriber());
    }

    /**
     * Redis 에서 주고 받을 채널 설정
     */
    @Bean
    public ChannelTopic channelTopic() {
        return new ChannelTopic("ch01");
    }

}

• RedisMessageStringSubscriber.java

/**
 * 메시지 수신 처리 로직
 */

@Slf4j
@Service
public class RedisMessageStringSubscriber implements MessageListener {

    public void onMessage(final Message message, final byte[] pattern) {
        log.info("String Message received: " + message.toString());
    }
}

• RedisSubApplication.java

@SpringBootApplication
public class RedisSubApplication {
    public static void main(String[] args) {
        SpringApplication.run(RedisSubApplication.class);
    }
}

실행은 Subscriber Application → Publisher Application 순으로 진행한다. Publisher 실행 후 Subscriber 로 넘어가면 터미널창에서 다음과 같이 메시지가 잘 전달된 것을 확인할 수 있다.

📝 참고한 자료

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