[재능교환소] Spring Boot와 RabbitMQ로 확장 가능한 1:1 채팅 구축하기

10000JI·2024년 6월 26일
rabbitmqstompwebsocket채팅
0

프로젝트

목록 보기
14/14
post-thumbnail

🍭 상황

직전 포스팅 : [재능교환소] Spring Boot로 Stomp 기반 1:1 채팅 구현 (with React)

프로젝트를 개발하는 과정에서 실시간 소통을 위한 1:1채팅 기능이 필요하여 만들었었다.

초기에는 빠른 구현과 간단한 설정을 위해 Spring Boot의 내장 메시지 브로커를 활용하여 STOMP 프로토콜 기반의 1:1 실시간 채팅 시스템을 구현했다.

이 방식은 별도의 외부 메시징 시스템 없이 채팅 기능을 구현할 수 있어 개발 초기 단계에 적합했다.

하지만 시스템의 확장성과 안정성에 대한 문제가 남아있었다.

특히 Nginx를 사용한 무중단 배포를 구현하려 했을 때, Spring Boot의 내장 브로커로는 배포 과정에서의 메시지 연속성 보장에 한계가 있었다.

내장 브로커는 단일 서버 환경에서는 잘 작동하지만, 무중단 배포 과정에서 메시지의 일관성을 보장하기 어렵다.

이러한 문제를 해결하고 시스템의 안정성을 개선하기 위해 외부 메시지 브로커인 RabbitMQ를 도입하기로 결정했다.

🍫 RabbitMQ에 대해 알아보자

  1. Producer:
  • 메시지를 생성하고 RabbitMQ로 보내는 주체이다.
  • 메시지를 Exchange에 발행(publish)한다.
  • 메시지 생성 후 어떤 Exchange로 보낼지 결정한다.
  1. Exchange:
  • Producer로부터 받은 메시지를 Queue로 라우팅하는 역할을 한다.
  • 여러 타입의 Exchange가 있으며, 각각 다른 라우팅 로직을 가진다.
    (Direct, Fanout, Topic, Headers).
  • 메시지의 라우팅 키와 Exchange에 설정된 규칙에 따라 하나 이상의 Queue로 메시지를 전달한다.
  1. Queue:
  • 메시지가 저장되는 버퍼이다.
  • Consumer가 메시지를 받아갈 때까지 메시지를 보관한다.
  • 하나의 Queue에 여러 Consumer가 연결될 수 있으며, 이 경우 라운드 로빈 방식으로 메시지를 분배한다.
  • Exchange와 바인딩(binding)되어 있어, 특정 조건의 메시지만 받을 수 있다.
  1. Consumer:
  • Queue로부터 메시지를 받아 처리하는 응용 프로그램이다.
  • 하나 이상의 Queue를 구독(subscribe)하여 메시지를 소비한다.
  • 메시지를 받은 후 처리하고, 처리 완료를 RabbitMQ에 알린다 (ack).

작동과정

  1. Producer가 메시지를 생성하여 Exchange에 발행
  2. Exchange는 라우팅 규칙에 따라 메시지를 적절한 Queue(들)로 전달
  3. 메시지는 Consumer가 가져갈 때까지 Queue에 저장
  4. Consumer는 구독 중인 Queue에서 메시지를 받아 처리

앞서 말했듯 RabbitMQ에서는 메시지를 다양한 방식으로 라우팅하기 위해 네 가지 타입의 Exchange를 제공한다.

각 Exchange 타입의 특징과 작동 원리를 살펴보자.

🍡 Exchange 타입

RabbitMQ Direct Exchange

Direct Exchange는 메시지의 라우팅 키와 동일한 바인딩 키를 가진 큐에 메시지를 전달한다.

작동 과정:

  1. Producer가 메시지를 Direct Exchange에 전송한다.
  2. Exchange는 메시지의 라우팅 키를 확인한다.
  3. Exchange는 자신에게 바인딩된 큐들 중 바인딩 키가 메시지의 라우팅 키와 정확히 일치하는 큐를 찾는다.
  4. 일치하는 큐에 메시지를 전달한다.

RabbitMQ Fanout Exchange

Fanout Exchange는 자신에게 바인딩된 모든 큐에 메시지를 라우팅한다.
라우팅 키는 무시된다.

작동 과정:

  1. Producer가 메시지를 Fanout Exchange에 전송한다.
  2. Exchange는 자신에게 바인딩된 모든 큐를 확인한다.
  3. 바인딩된 모든 큐에 메시지의 복사본을 전송한다.

RabbitMQ Topic Exchange

Topic Exchange는 라우팅 키와 바인딩 패턴 사이의 와일드카드(*) 매칭을 통해 메시지를 큐에 전달한다.

작동 과정:

  1. Producer가 메시지를 Topic Exchange에 전송한다.
  2. Exchange는 메시지의 라우팅 키를 확인한다.
  3. Exchange는 자신에게 바인딩된 큐들의 바인딩 패턴과 메시지의 라우팅 키를 비교한다.
  4. 매칭되는 패턴을 가진 모든 큐에 메시지를 전달한다.

RabbitMQ Headers Exchange

Headers Exchange는 메시지의 헤더 속성을 사용하여 라우팅을 수행한다.

작동 과정:

  1. Producer가 메시지를 Headers Exchange에 전송한다.
  2. Exchange는 메시지의 헤더를 확인한다.
  3. Exchange는 자신에게 바인딩된 큐들의 바인딩 속성과 메시지의 헤더를 비교한다.
  4. 매칭되는 속성을 가진 큐에 메시지를 전달한다.

설치

윈도우 pc 로컬에 설치한다면 다음 링크를 참고하길 바란다.

Windows10에 RabbitMQ 빠르게 설치하기 (with Erlang)

🍮 RabbitMQ를 사용한 1:1 채팅 구현

필자는 Maven을 사용하였기 때문에 pom.xml에 의존성을 추가해주었다.

pom.xml

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-reactor-netty</artifactId>
    <version>3.0.0</version>
</dependency>

RabbitMQ 의 정상적인 동작을 위해 설정 클래스 만들어 보자.

RabbitConfig

@Configuration
@EnableRabbit
@RequiredArgsConstructor
public class RabbitConfig {

    private static final String CHAT_QUEUE_NAME = "chat.queue";
    private static final String CHAT_EXCHANGE_NAME = "chat.exchange";
    private static final String ROUTING_KEY = "*.room.*";

    //Queue 등록
    @Bean
    public Queue queue() {
        return new Queue(CHAT_QUEUE_NAME, true);
    }

    //Exchange 등록
    @Bean
    public TopicExchange exchange() {
        return new TopicExchange(CHAT_EXCHANGE_NAME);
    }

    // Exchange와 Queue바인딩
    @Bean
    public Binding binding(Queue queue, TopicExchange exchange){
        return BindingBuilder
                .bind(queue)
                .to(exchange)
                .with(ROUTING_KEY);
    }

    // RabbitMQ와의 메시지 통신을 담당하는 클래스
    @Bean
    public RabbitTemplate rabbitTemplate() {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory());
        rabbitTemplate.setMessageConverter(jsonMessageConverter());
        return rabbitTemplate;
    }

    // RabbitMQ와의 연결을 관리하는 클래스
    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory factory = new CachingConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("guest");
        factory.setPassword("guest");
        return factory;
    }

    // 메시지를 JSON형식으로 직렬화하고 역직렬화하는데 사용되는 변환기
    // RabbitMQ 메시지를 JSON형식으로 보내고 받을 수 있음
    @Bean
    public Jackson2JsonMessageConverter jsonMessageConverter() {
        //LocalDateTime serializable을 위해
        ObjectMapper objectMapper = new ObjectMapper();
        objectMapper.configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, true);
        objectMapper.registerModule(dateTimeModule());

        Jackson2JsonMessageConverter converter = new Jackson2JsonMessageConverter(objectMapper);

        return converter;
    }

    @Bean
    public Module dateTimeModule() {
        return new JavaTimeModule();
    }
}

위 설정 내용을 살펴보면 다음과 같다.

  • chat.queue라는 이름의 Queue 를 생성
  • chat.exchange라는 이름의 Topic Exchange를 생성
  • Queue와 Exchange를 *.room.* 라우팅 키로 Binding
  • JSON형태로 주고받을 수 있도록 RabbitTemplate 설정
  • RabbitMQ 서버와의 연결을 관리하는 ConnectionFactory를 설정
  • 로컬호스트의 5672 포트로 연결하며, 기본 사용자 이름과 비밀번호인 guest를 사용
  • Jackson 라이브러리를 사용하여 메시지를 JSON 형식으로 직렬화/역직렬화
  • LocalDateTime 등의 날짜/시간 타입을 처리하기 위한 설정

여기서 주의 깊게 봐야되는 점은 Exchange 타입 중 Topic Exchange을 사용했다는 점이다.

특정 패턴의 라우팅 키에 따라 메세지를 라우팅 할 수 있는데, 여기서 특정 패턴을 방 번호ID로 주었다.

(채팅방 생성 시, UUID로 고유한 방 번호를 생성하여 채팅방 ID로 사용한다. 이 ID와 함께 채팅방에 참여한 사용자들의 userId를 데이터베이스에 저장하고 있다.)

기존에 만들었던 WebSocketConfig도 수정해주어야 한다.

WebSocketConfig

@Configuration
@EnableWebSocketMessageBroker
@RequiredArgsConstructor
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {

    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {

        // socketJs 클라이언트가 WebSocket 핸드셰이크를 하기 위해 연결할 endpoint를 지정할 수 있다.
        registry.addEndpoint("/chat/inbox")
                .setAllowedOriginPatterns("*"); // cors 허용을 위해 꼭 설정해주어야 함. setCredential() 설정시에 AllowedOrigin 과 같이 사용될 경우 오류가 날 수 있으므로 OriginPatterns 설정으로 사용하였음
    }

    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
    
    	// 메시지 브로커 설정
        registry.setPathMatcher(new AntPathMatcher(".")); // url을 chat/room/3 -> chat.room.3으로 참조하기 위한 설정
        registry.enableStompBrokerRelay("/queue", "/topic", "/exchange", "/amq/queue")
                .setRelayHost("localhost")
                .setRelayPort(61613)
                .setSystemLogin("guest")
                .setSystemPasscode("guest")
                .setClientLogin("guest")
                .setClientPasscode("guest");

        // 클라이언트로부터 메시지를 받을 api의 prefix를 설정함
        // publish
        registry.setApplicationDestinationPrefixes("/pub");


    }


}

보면 나머지는 다 동일하나, configureMessageBroker()에 메시지 브로커 설정이 변경되었다.

메시지 브로커 설정

이전에는 registry.enableSimpleBroker("/sub")로 적어 내장 메모리 기반 브로커를 사용하였다.

이를 registry.enableStompBrokerRelay(...)로 작성해주며 RabbitMQ와 같은 외부 메시지 브로커 사용할 수 있도록 변경해주었다.

registry.setPathMatcher(new AntPathMatcher("."))은 URL 패턴을 점(.)으로 구분하도록 설정한 것이다.

브로커 릴레이 호스트를 localhost, RabbitMQ의 STOMP 프로토콜 기본 포트인 61613으로 설정해주었다.

이후 setSystemLogin("guest") 와 setSystemPasscode("guest")은 STOMP 브로커 릴레이가 RabbitMQ 서버에 연결할 때 사용하는 시스템 레벨의 인증 정보를 추가해준 것

setClientLogin("guest") 와 setClientPasscode("guest")는 개별 클라이언트 연결에 사용되는 인증 정보를 추가해주었다.

다음은 채팅 메시지를 처리하는 컨트롤러 및 서비스를 수정해보자.

ChatMessageController

@Controller
@RequiredArgsConstructor
@Slf4j
public class ChatMessageController {

    private final static String CHAT_EXCHANGE_NAME = "chat.exchange";

    private final ChatMessageService chatMessageService;
    private final RabbitTemplate template;

     @MessageMapping("chat.message")
    public void sendMessage(ChatDto.ChatMessageDto message) {
        try {
            ChatMessage newChat = chatMessageService.createChatMessage(message);
            if (newChat != null) {
                template.convertAndSend(CHAT_EXCHANGE_NAME, "room." + message.getRoomId(), newChat);
                log.info("Message sent to RabbitMQ: {}", newChat);
            } else {
                log.error("Failed to create chat message. User might not be in the chat room. User: {}, Room: {}",
                        message.getAuthorId(), message.getRoomId());
            }
        } catch (Exception e) {
            log.error("Error processing message: ", e);
        }
    }
}

ChatMessageServiceImpl

@Service
@RequiredArgsConstructor
public class ChatMessageServiceImpl implements ChatMessageService{

    private final ChatRoomRepository chatRoomRepository;

    @Override
    public ChatMessage createChatMessage(ChatDto.ChatMessageDto chatMessageDto) {

        ChatRoom chatRoom = chatRoomRepository.findById(chatMessageDto.getRoomId())
                .orElseThrow(() -> new RuntimeException("Chat room not found"));

        boolean exists = chatRoom.getChatRoomMembers().stream()
                .anyMatch(member -> member.getUsername().equals(chatMessageDto.getAuthorId()));

        if (!exists) {
            log.error("User not found in chat room: {}", chatMessageDto.getAuthorId());
            return null;
        }

        ChatMessage chatMessage = chatMessageDto.toEntity();
        chatRoom.setLastChatMesg(chatMessage);
        chatRoomRepository.save(chatRoom);

        return chatMessage;
    }
}

ChatMessageDto

/**
     * 웹소켓 접속시 요청 Dto
     */
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public static class ChatMessageDto {
    private String roomId;
    private String authorId;
    private String message;

    /* Dto -> Entity */
    public ChatMessage toEntity() {
        ChatMessage chatMessage = ChatMessage.builder()
                .roomId(roomId)
                .authorId(authorId)
                .message(message)
                .createdAt(LocalDateTime.now())
                .build();
        return chatMessage;
    }
}

채팅 메시지 처리하는 컨트롤러의 주요 기능

@MessageMapping("chat.message")로 설정하여 클라이언트로부터 /pub/chat.message 목적지로 전송된 STOMP 메시지를 처리한다.

template.convertAndSend() 메소드를 사용하여 메시지를 RabbitMQ로 전송한다.

메시지는 chat.exchange로 전송되며, 라우팅 키는 room. + 메시지의 방 ID로 구성된다.

기존 [재능교환소] Spring Boot로 Stomp 기반 1:1 채팅 구현 (with React) 포스팅에서는 Spring의 내장 STOMP 브로커를 사용하였다.

이때는 헤더의 AccessToken을 통해 사용자 ID를 추출하고, 이를 STOMP 세션에 저장하여 사용했다. 그러나 RabbitMQ로 전환하면서 이 방식이 더 이상 유효하지 않다는 것을 발견했다.

RabbitMQ는 외부 브로커로서 Spring의 STOMP 세션을 직접 관리하지 않기 때문에, 연결 시 헤더에서 검증한 토큰으로부터 추출한 사용자 ID를 세션에 저장하는 것이 불가능하며 메시지를 보낼 때마다 저장된 세션 정보에서 사용자 ID를 가져오는 것도 할 수 없었다.

이러한 제약을 해결하기 위해, 접근 방식을 변경하였다.

클라이언트가 메시지를 보낼 때 authorId를 JSON 페이로드에 직접 포함시켰다.

서버에서는 이 authorId를 사용해 해당 사용자가 채팅방에 존재하는지, 그리고 메시지를 보낼 권한이 있는지 확인하는 유효성 검사를 수행한다.

이 방식으로 매 메시지마다 사용자의 권한을 확인할 수 있어, 세션 저장의 한계를 우회하면서도 보안성을 유지할 수 있다.

이 새로운 접근 방식은 외부 브로커 사용에 따른 세션 관리의 복잡성을 피하고, 처리 로직을 단순화하면서도 필요한 보안 검증을 수행할 수 있게 해준다.

🧁 APIC로 STOMP 테스트 및 RabbitMQ Management 확인

APIC로 테스트 하러 가기

Request URL에는 websocket을 연결하기 위해 ws://localhost/chat/inbox을 적어준다.

그리고 Connection type에 WebSocket 대신 Stomp를 클릭해주고, Stomp Subscription URL에는 /exchange/chat.exchange/room.{roomId}을 적어준다.

/exchange/chat.exchange/room.{roomId}

/exchange: 메시지가 RabbitMQ exchange로 라우팅되어야 함
chat.exchange: RabbitConfig에서 정의한 exchange의 이름
room.{roomId}: 라우팅 키의 패턴으로 ChatMessageController에서 메시지를 보낼 때 사용한 라우팅 키와 일치
"."을 구분자로 사용: WebSocketConfig에서 AntPathMatcher를 .을 구분자로 설정

이로 인해 클라이언트는 특정 채팅방(roomId)에 대한 메시지만 구독이 가능하다.

동일하게 APIC 테스터를 두 개를 열어 놓고 한 곳에서 메세지를 보내보자.

Send의 Destination Queue에는 메세지 발행 요청을 위해 /pub/chat.message라고 적어준다.

옆에 json형태로 메세지를 보내줄 건데, 위에서 본 DTO 형식 그대로 적어주면 된다.

{
    "roomId" : "10ec8870-64b6-40cc-869f-17b1df2bc21e",
    "authorId": "kakao_sksk436@nate.com",
    "message" : "소셜로그인 계정입니다"
}

그럼 상대방에게도 메세지가 정상 수신되는 것을 확인할 수 있다.

메세지가 정상 수신되면 Overview에선 메세지가 방금 전송되었다고 Message rates에 표시가 되는 것을 확인할 수 있다.

참고

RabbitMQ를 실행한 후 터미널에서 rabbitmq-plugins enable rabbitmq_management를 친 후에 브라우저에 http://localhost:15672로 접속하면 RabbitMQ Managemnt 창에 들어갈 수 있다.

어플리케이션을 실행하면 RabbitMQ와 커넥션이 활성화되고, 소켓을 2개 연결했기 때문에 이름이 127.0.0.1:61826과 127.0.0.1:61829가 생겼다.

Queues and Streams에도 두 개의 Queue가 생겼다.

이처럼 RabbitMQ 웹 관리 콘솔창에서 실시간으로 다양한 정보를 확인할 수 있다!

출처

RabbitMQ Exchanges

profile
10000JI
Velog에 기록 중
이전 포스트

[재능교환소] Spring Boot: 소셜 로그인 - 로그아웃 & 회원탈퇴(Kakao, Google)

0개의 댓글