Rabbitmq 적용

Posted Apr 5, 2024 Updated May 17, 2026

By haedal-uni

9 min read

Rabbitmq 적용

RabbitMQ 적용

이 전에 어떻게 적용을 해야하는지 파악했으니 이제 본 코드에 적용해보기로 했다.

적용할 부분

1:1 채팅
현재 나는 1:n 채팅은 존재하지 않는다.
따라서 “topic”은 지운다.
topic과 queue에 대한 stackoverflow 글

MessageType에 따른 Queue 설정
Connect, Send, Disconnect에 대한 별도의 Queue를 설정하여 처리를 분산시킨다.
사용자마다 Queue를 생성하는 것보다 효율적으로 처리할 수 있을 것 같았다.

Code 수정

application.properties

  
rabbitmq.connect.queue = connect.queue
rabbitmq.send.queue = send.queue
rabbitmq.disconnect.queue = disconnect.queue

rabbitmq.connect.exchange = connect.exchange
rabbitmq.send.exchange = send.exchange
rabbitmq.disconnect.exchange = disconnect.exchange

rabbitmq.routing.key = room.*

Stomp Config

  
@Configuration
@EnableWebSocketMessageBroker
@RequiredArgsConstructor
public class StompWebSocketConfig implements WebSocketMessageBrokerConfigurer {
    private final StompHandler stompHandler;
    
    @Override
    public void configureMessageBroker(MessageBrokerRegistry config) {
        config.setApplicationDestinationPrefixes("/app");
        config.setPathMatcher(new AntPathMatcher("."));
        config.enableStompBrokerRelay("/exchange"); // custom exchange 사용 
    }

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

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

custom exchange만을 사용하여 message routing을 처리하기 때문에 “/exchange”만 추가했다.

RabbitMQConfig

MessageType(connect, send, disconnect)에 따라 exchange, queue 분리

  
@Configuration
public class RabbitMQConfig {

    @Value("${spring.rabbitmq.host}")
    private String host;

    @Value("${spring.rabbitmq.port}")
    private int port;

    @Value("${spring.rabbitmq.username}")
    private String username;

    @Value("${spring.rabbitmq.password}")
    private String password;

    @Value("${rabbitmq.connect.queue}")
    private String connectQueue;

    @Value("${rabbitmq.send.queue}")
    private String sendQueue;

    @Value("${rabbitmq.disconnect.queue}")
    private String disconnectQueue;

    @Value("${rabbitmq.connect.exchange}")
    private String connectExchange;

    @Value("${rabbitmq.send.exchange}")
    private String sendExchange;

    @Value("${rabbitmq.disconnect.exchange}")
    private String disconnectExchange;

    @Value("${rabbitmq.routing.key}")
    private String routingKey;

    @Bean
    public Queue connectQueue() {
        return new Queue(connectQueue, true);
    }

    @Bean
    public Queue sendQueue() {
        return new Queue(sendQueue, true);
    }

    @Bean
    public Queue disconnectQueue() {
        return new Queue(disconnectQueue, true);
    }

    @Bean
    public TopicExchange connectExchange() {
        return new TopicExchange(connectExchange);
    }

    @Bean
    public TopicExchange sendExchange() {
        return new TopicExchange(sendExchange);
    }

    @Bean
    public TopicExchange disconnectExchange() {
        return new TopicExchange(disconnectExchange);
    }

    @Bean
    public Binding connectBinding() {
        return BindingBuilder
                .bind(connectQueue())
                .to(connectExchange())
                .with(routingKey);
    }

    @Bean
    public Binding sendBinding() {
        return BindingBuilder
                .bind(sendQueue())
                .to(sendExchange())
                .with(routingKey);
    }

    @Bean
    public Binding disconnectBinding() {
        return BindingBuilder
                .bind(disconnectQueue())
                .to(disconnectExchange())
                .with(routingKey);
    }

    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host);
        connectionFactory.setHost(host);
        connectionFactory.setPort(port);
        connectionFactory.setVirtualHost("/");
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        return connectionFactory;
    }

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(jackson2JsonMessageConverter());
        rabbitTemplate.setRoutingKey(routingKey);
        return rabbitTemplate;
    }

    @Bean
    public MessageConverter jackson2JsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    @Bean
    public SimpleMessageListenerContainer container(ConnectionFactory connectionFactory) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        return container;
    }
}

connectExchange, sendExchange, disconnectExchange: 각각 연결, 전송, 연결 끊기와 관련된 Exchange를 정의했다.

connectQueue, sendQueue, disconnectQueue: 위와 동일하게 연결, 전송, 연결 끊기와 관련된 Queue를 정의했다.

그리고 Exchange와 Queue를 연결하는 Binding을 설정했다:

각각의 RabbitTemplate과 container 빈에 Exchange와 Queue 이름을 설정하지 않은 이유는

Exchange와 Queue가 각각 1개씩 있는 것이 아니라 여러 개 있기 때문이다.

RabbitTemplate과 container는 모든 Exchange와 Queue에 대해

메시지 전송 및 수신을 담당하는 역할을 하기 때문에 공통적인 설정만을 추가했다.

Controller

  
@Controller
@RequiredArgsConstructor
@Slf4j
public class ChatController {
    private final ChatServiceImpl chatService;
    private final JwtTokenProvider jwtTokenProvider;
    private final RedisService redisService;
    private final RabbitTemplate rabbitTemplate;

    @Value("${rabbitmq.connect.exchange}")
    private String connectExchange;

    @Value("${rabbitmq.send.exchange}")
    private String sendExchange;

    @MessageMapping("chat.sendMessage")
    public void sendMessage(@Payload ChatMessage chatMessage) {
        String roomId = chatMessage.getRoomId();
        rabbitTemplate.convertAndSend(sendExchange, "room." + roomId, chatMessage);
    }

    @MessageMapping("chat.addUser")
    public void addUser(@Payload ChatMessage chatMessage, SimpMessageHeaderAccessor headerAccessor) {
        String token = headerAccessor.getFirstNativeHeader("Authorization");
        User user = jwtTokenProvider.getUserFromToken(token);
        String roomId = chatMessage.getRoomId();

        chatMessage.setSender(user.getNickname());
        chatMessage.setType(MessageType.JOIN);
        chatMessage.setAuth(user.getRole().name());

        rabbitTemplate.convertAndSend(connectExchange, "room." + roomId, chatMessage);
    }

    @EventListener
    public void handleWebSocketDisconnectListener(SessionDisconnectEvent event) {
        StompHeaderAccessor headerAccessor = StompHeaderAccessor.wrap(event.getMessage());
        String sessionId = (String) headerAccessor.getHeader("simpSessionId");
        String token = redisService.getSession(sessionId);
        User user = jwtProvider.getUserFromToken(token);
        String nickname = user.getNickname();
        String roomId = redisService.getRoomId(nickname);

        ChatMessage chatMessage = new ChatMessage();
        chatMessage.setType(MessageType.LEAVE);
        chatMessage.setSender(nickname);
        chatMessage.setRoomId(roomId);

        rabbitTemplate.convertAndSend(disconnectExchange, "room." + roomId, chatMessage);
    }
}

client

  
stompClient.subscribe('/exchange/' + connectEx +'/room.' + roomId, message);
stompClient.subscribe('/exchange/' + sendEx +'/room.' + roomId, message);
stompClient.subscribe('/exchange/' + disconEx +'/room.' + roomId, message);