[Web] 로컬 환경에서 SpringBoot와 Flask에 RabbitMQ 적용하기

📍 RabbitMQ 설치

1️⃣ RabbitMQ 및 Erlang 설치

설치는 아래 사이트를 참고했다.

Windows10에 RabbitMQ 빠르게 설치하기 (with Erlang), 그리고 RabbitMQ management 띄우기 (tistory.com)

Windows10에 RabbitMQ 빠르게 설치하기 (with Erlang), 그리고 RabbitMQ management 띄우기

 

2️⃣ 환경변수 설정

위 과정대로 진행한 후 환경 변수 설정이 필요하다.

[RabbitMQ] 무작정 시작하기 (1) - 설치 및 실행 (tistory.com)

[RabbitMQ] 무작정 시작하기 (1) - 설치 및 실행

 

3️⃣ 실행

아래와 같이 설치된 RabbitMQ의 sbin 경로로 들어가서 실행이 가능하다.

C:\Program Files\RabbitMQ Server\rabbitmq_server-3.11.11\sbin>rabbitmq-plugins enable rabbitmq_management

 

RabbitMQ의 기본 경로는 5672이며, RabbitMQ GUI의 기본경로는 15672이므로

http://localhost:15672 경로로 GUI에 접근이 가능하다.

성공적으로 설치되었다면 아래 화면을 볼 수 있고, Username과 Password 모두 guest로 접속할 수 있다.

 

4️⃣ 실행 실패 시

위의 1, 2, 3번 과정을 모두 진행하였는데 GUI가 실행되지 않는다면

RabbitMQ와 Erlang을 모두 삭제한 뒤 다시 설치하면 잘 실행된다.

(각각의 파일에 있는 Uninstall 파일을 사용할 것)

 

💥 삭제 후 다시 설치할 때의 주의사항

• RabbitMQ는 Uninstall 후 해당 폴더까지 삭제할 것
• Erlang은 Uninstall 후 남은 폴더를 삭제하려고 할 때 실행중이라면서 삭제되지 않을 것임. 이 때 그냥 냅두기
• Erlang은 다시 설치할 때 남은 폴더 때문에 경고 문구가 뜰텐데, ‘무시’ 선택하기

 

---

📍 SpringBoot RabbitMQ 적용

1️⃣ application.properties

# RabbitMQ
spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

 

2️⃣ build.gradle

// RabbitMQ
implementation 'io.github.fuquanming:fqm-mq-rabbit:1.0.5'
implementation 'jp.spring-boot-reference:functional-template:1.1'

 

3️⃣ config 파일 작성

Producer 및 Consumer 구현에 필요한 내용을 작성해주었다.

아래와 같이 작성해주면 별도로 queue와 exchange를 만들어 줄 필요 없이 자동으로 생성된다.

➕ 아래에서 queue는 java.util의 queue 아님 주의

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitmqConfig {

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

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

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

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

    /***
     * Producer
     ***/

    @Bean
    Queue queue() {
        return new Queue("test.queue", false);
    }

    @Bean
    DirectExchange directExchange() {
        return new DirectExchange("test.exchange");
    }

    @Bean
    Binding binding(DirectExchange directExchange, Queue queue) {
        return BindingBuilder.bind(queue).to(directExchange).with("test.key");
    }

    @Bean
    RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory, MessageConverter messageConverter) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(messageConverter);
        return rabbitTemplate;
    }

    /***
     * either Producer, Consumer
     ***/

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

    @Bean
    MessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    /***
     * Consumer
     ***/

    @Bean
    SimpleRabbitListenerContainerFactory simpleRabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
        final SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(messageConverter());
        return factory;
    }
}

 

4️⃣ Producer 구현

import lombok.RequiredArgsConstructor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Service;

@Service
@RequiredArgsConstructor
public class Producer {

    private final RabbitTemplate rabbitTemplate;

    // 메시지 전송
    public void sendMessage(MessageDto messageDto) {
        rabbitTemplate.convertAndSend("test.exchange", "test.key", messageDto);
    }
}

 

5️⃣ Consumer 구현

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class Consumer {

    // 메시지 받기
    @RabbitListener(queues = "test.queue")
    public void consume(MessageDto messageDto) {
        System.out.println("[ messageTitle ] : " + messageDto.getTitle());
        System.out.println("[ messageContent ] : " + messageDto.getContent());
    }
}

 

---

📍 테스트 및 결과 확인

 

1️⃣ 메시지 적재하기

위에서 작성한 Producer의 메소드를 원하는 서비스에서 호출하여 사용할 수 있지만,

우선 테스트를 위해 Controller 메소드를 하나 만들어주었다.

(Consumer의 경우 따로 호출해주지 않아도 알아서 작동한다.)

그리고 일단 메시지가 잘 적재되는지 확인하기 위해

위에서 작성했던 Consumer는 주석처리한 후 Spring을 실행시켰고,

아래와 같이 Postman을 활용하여 요청을 보냈다.

 

위와 같이 요청을 보낼 시 아래 화면처럼 Queue에 하나의 메시지가 쌓인 것을 볼 수 있다.

(Ready와 Total에 1씩 값이 추가된 것 확인)

 

2️⃣ 메시지 받기

위와 같이 메시지가 잘 적재되는 것을 확인했으니, 다시 Consumer의 주석을 풀어서 Spring을 실행시켰다.

Spring 실행 시 특별한 요청 없이도, 우리가 작성한 Consumer가

queue에 쌓인 메시지를 가져와 실행시키는 모습을 볼 수 있다.

(우리는 받은 메시지를 출력하는 문장을 작성했으므로 Console 창에서 확인 가능)

➕ Consumer를 주석처리 해놓고 Producer 요청을 여러 번 보낸 후, Consumer를 실행시켜보는 것도  좋다.

 

---

📍 Flask 서버와 RabbitMQ로 소통

위의 과정은 RabbitMQ를 적용하고 테스트하는 것을 목적으로 하기 때문에

같은 서버에서 Producer와 Consumer를 모두 구현했다.

 

그러나 현재 진행중인 프로젝트에서는 SpringBoot에서 Flask로 통신하는 것을 위해 RabbitMQ를 사용했다.

따라서 Spring API에 Producer를 구현하고, 두 개의 Flask에서 Consumer를 구현하여 통신했다.

(한 개의 Producer, 한 개의 exchange, 한 개의 queue, 두 개의 Consumer)

https://velog.io/@sdb016/RabbitMQ-%EA%B8%B0%EC%B4%88-%EA%B0%9C%EB%85%90

 

Flask에서 Consumer를 구현하는 부분은 아래 블로그를 참고했다.

Python에서 RabbitMq 사용하기 (1) - RabbitMQ 기본 동작 · snowdeer's Code Holic

Python에서 RabbitMq 사용하기 (1) - RabbitMQ 기본 동작 · snowdeer's Code Holic

 

💥 하나의 Queue에서 두 개의 Consumer가 메시지를 받을 때 주의사항

prefetch count를 1로 설정하기!

 

RabbitMQ의 prefetch count의 default 값은 20이다.

 

따라서 하나의 queue에 40개의 요청이 있다고 가정하면,

Consumer1이 queue에서 20개의 요청을 꺼내와서 Consumer1의 메모리에 쌓고,

Consumer2가 남은 queue의 20개의 요청을 꺼내와서 Consumer2의 메모리에 쌓은 뒤

각 Consumer는 메모리에 쌓아둔 요청을 차례대로 처리하는 식으로 동작한다.

 

하지만 이 때 Consumer가 3개라고 생각해보자.

Consumer1과 Consumer2는 사이좋게 20개씩 요청을 가져가지만,

Consumer3는 계속 일을 하지 않고, Consumer1과 Consumer2만 일을 하는 문제가 발생할 수 있다.

 

따라서 prefetch count를 1로 바꿔주면

Consumer들이 굳이 자신의 메모리에 메시지를 쌓아두고 처리하지 않고,

바로바로 Queue에서 하나씩 꺼내 요청을 처리함으로써 모든 Consumer에 메시지가 골고루 분배될 수 있다.

 

---

참고

• https://velog.io/@sdb016/RabbitMQ-%EA%B8%B0%EC%B4%88-%EA%B0%9C%EB%85%90
• Springboot + RabbitMQ 연동 및 초간단 샘플 프로젝트 만들기 (tistory.com)
• [Docker] Springboot RabbitMQ 초간단 연동 (velog.io)