[Spring] 소켓통신 이용한 채팅 구현하기 3 - RabbitMQ 연동

 

기본적으로 스프링에서 내장브로커를 제공하지만,

1. 인메모리 형식으로 데이터의 유실 위험도 있고

2. spring boot서버 내에서 함께 처리하기 때문에 서버의 부담도 커진다.

3. 메시지 큐를 모니터링하기 어렵고

4. 또한 현재 프로젝트 규모에서는 아직 필요없지만, 추후 서버를 여러개 둔다면 메시지를 함께 처리할 수 없어 확장성이 떨어진다.

이러한 여러 이유들로 외부 메시지 브로커인 RabbitMQ를 도입하고자 한다.

 

• RabbitMQ 의 메시지 전달 과정

1. 송신자가 메시지를 보내면 브로커가 처리과정을 위임받는다.

2. 일종의 우체통 역할을 하는 Exchange로 먼저 전달해서 메시지를 분류한다.

3. exchange에서는 몇가지 종류가 있지만 여기서는 넘어간다. ( Topic은 라우팅 키를 패턴으로 검사하고 전달한다 )

4. exchange와 메시지 queue는 라우팅 키, 혹은 패턴으로 바인딩 되어있어 일치하는 큐로 메시지를 전달한다.

5. 수신자는 queue에서 메시지를 가져온다

---

 

• 의존성 추가

implementation group: 'org.springframework.boot', name: 'spring-boot-starter-reactor-netty', version: '2.4.6'
implementation 'org.springframework.boot:spring-boot-starter-amqp'

 

 

• 프로퍼티 추가

spring.rabbitmq.username=guest //default ID
spring.rabbitmq.password=guest //default Password
spring.rabbitmq.host=localhost //default host
spring.rabbitmq.port=61613

 

• webSocketConfig

    @Override
    public void configureMessageBroker(MessageBrokerRegistry config) {
        //메시지 구독 url
        config.enableStompBrokerRelay("/exchange")
                .setClientLogin(rabbitUser)
                .setClientPasscode(rabbitPwd)
                .setSystemLogin(rabbitUser)
                .setSystemPasscode(rabbitPwd)
                .setRelayHost(rabbitHost)
                .setRelayPort(rabbitPort)
                .setVirtualHost(rabbitVHost);
        //메시지 발행 url
        config.setPathMatcher(new AntPathMatcher("."));
        config.setApplicationDestinationPrefixes("/pub");
    }

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

 

rabbitmq에서는 기본적으로 queue,exhange의 이름이나 라우팅 키, 패턴 등을 작성할때 .을 구분자로 사용한다.

특히 요청 경로로 사용될때 url에서 계층 구분자로 사용되는 / 와 혼동이 생길수 있어 그런듯 하다.

때문에 요청경로에서 .을 사용할 수 있도록 config.setPathMatcher(new AntPathMatcher("."));를 적어주었다.

 

 

 

 

chat-gpt의 대답

 

 

• RabbitConfig

@Configuration
@EnableRabbit
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.*";

    @Value("${spring.rabbitmq.username}")
    private String rabbitUser;
    @Value("${spring.rabbitmq.password}")
    private String rabbitPw;
    @Value("${spring.rabbitmq.host}")
    private String rabbitHost;
    @Value("${spring.rabbitmq.virtual-host}")
    private String rabbitVh;
    @Value("${RABBITMQ_PORT}")
    private int rabbitPort;

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

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

    //Exchange와 Queue 바인딩
    @Bean
    public Binding binding(Queue queue, TopicExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY);
    }
    @Bean
    SimpleRabbitListenerContainerFactory simpleRabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
        final SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(jsonMessageConverter());
        return factory;
    }

    /* messageConverter를 커스터마이징 하기 위해 Bean 새로 등록 */
    @Bean
    public RabbitTemplate rabbitTemplate(){
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory());
        rabbitTemplate.setMessageConverter(jsonMessageConverter());
        rabbitTemplate.setRoutingKey(ROUTING_KEY);
        return rabbitTemplate;
    }

    @Bean
    public ConnectionFactory connectionFactory(){
        CachingConnectionFactory factory = new CachingConnectionFactory();
        factory.setHost(rabbitHost);
        factory.setVirtualHost(rabbitVh);
        factory.setUsername(rabbitUser);
        factory.setPassword(rabbitPw);
        factory.setPort(5672);
        return factory;
    }

    @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();
    }
}

 

채팅을 보여줄 대상을 채팅방으로 구분하기때문에 채팅방id를 라우팅키로 사용할 것이다.

 

• chatController 수정

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

    // /pub/chat.message.{roomId} 로 요청하면 브로커를 통해 처리
    // /exchange/chat.exchange/room.{roomId} 를 구독한 클라이언트에 메시지가 전송된다.
    @MessageMapping("chat.enter.{chatRoomId}")
    public void enterUser(@Payload ChatDTO chat, @DestinationVariable String chatRoomId) {
        chat.setTime(LocalDateTime.now());
        chat.setMessage(chat.getSender() + " 님 입장!!");
        rabbitTemplate.convertAndSend(CHAT_EXCHANGE_NAME, "room." + chatRoomId, chat);

    }

    @MessageMapping("chat.message.{chatRoomId}")
    public void sendMessage(@Payload ChatDTO chat, @DestinationVariable String chatRoomId) {
        chat.setTime(LocalDateTime.now());
        chat.setMessage(chat.getMessage());
        rabbitTemplate.convertAndSend(CHAT_EXCHANGE_NAME, "room." + chatRoomId, chat);

    }

    //기본적으로 chat.queue가 exchange에 바인딩 되어있기 때문에 모든 메시지 처리
    @RabbitListener(queues = CHAT_QUEUE_NAME)
    public void receive(ChatDTO chatDTO){
        System.out.println("received : " + chatDTO.getMessage());
    }

.
.
}

 

RabbicConfig에서 미리 chat.queue를 만들어두고 root.*을 라우팅키로 사용하여 exchange에 연결시켜 놓았기 때문에 exchange로 들어오는 모든 채팅방의 메시지를 receive()를 통해서 처리할수있다.

 

 

apic에서 위와같이 테스트해보면 정상적으로 메시지가 전달되고

 

receiver()를 통해 위와같이 처리되는걸 확인할수있다.

 

rabbitmq페이지에서도 위와 같이 클라이언트와 연결된 queue를 확인할 수 있다.