RabbitMQ를 활용한 메시지 처리 방식 구현: 설치부터 적용까지

들어가기 전에

이번 F-lab 프로젝트에서 Feed 기능을 구현하게 됐다.

Prototype UI

각각의 이벤트가 발생했을 때 피드 메시지를 생성하고 전달해서 Feed 테이블에 저장했어야 하는데 직접 저장하는 메소드를 호출하느냐, 아니면 이벤트를 처리하기 위한 별도의 장치를 사용하느냐를 두고 생각하다가 각 서비스 간의 의존도와 결합도를 낮추기 위해서 후자를 선택하기로 했다.

 

그 다음으로 고민했던 것은 Spring의 @EventListener를 사용하느냐, Message broker를 사용하느냐였다.

Spring의 @EventListener로도 충분히 하고자 하는 기능 구현을 해낼 수 있었지만 굳이 Message broker를 사용한 이유는 scale out에 대한 대비였다. 지금 당장은 싱글 서버이지만 증설을 목표로 두고 하고 있었어서 따로 서버를 구축하기로 했다.

 

그리고 마지막으로 Apache Kafka와 RabbitMQ를 두고 고민을 해야했다. 아래는 내가 고민 끝에 Kafka보다는 RabbitMQ를 선택해야 한다는 결론에 도달하게 된 이유이다.

Kafka 고가용성의 핵심은 3개 이상의 카프카 브로커들로 이루어진 클러스터에서 진가를 발휘하게 된다.

출처 : <아파치 카프카 애플리케이션 프로그래밍 with 자바> 저자이신 최원영님 유튜브 영상에서

• kafka는 대용량 데이터를 다룰 때 훨씬 더 성능이 좋고 고가용성에 레퍼런스도 많다. 하지만 kafka를 제대로 쓴다고 하면 위처럼 카프카 클러스터를 최소로 구축하기 위해 3대 이상의 서버가 필요한데 우리 서비스는 그만큼의 비용을 쓸 만큼의 분산 처리가 필요한 기능을 구현하는 것 같지 않다는 판단이 든다.
• kafka가 제공하는 offset과 같은 개념들이 현재 기능 구현에 쓰일 것 같지 않다. 활용하지 않을 기능들을 장점으로 가져가기 보다는 필요한 기능만을 생각하자.

그래서 결국엔 원하는 기능도 구현할 수 있고 비용 대비 잘 쓸 수 있는 rabbitMQ를 사용하기로 했다. 더불어 모니터링 플러그인도 제공해 주기 때문에 활용하기에 더 좋을 것 같다고 생각했다.

 

 

시작하기

1. 서버 생성

NCP 서버 생성부터 ssh 접속까지

 

[ACG 설정]

출처 :&nbsp;https://guide.ncloud-docs.com/beta/docs/rabbitmq-rabbitmq-1-1

• TCP/5672 : AMQP 포트
• TCP/15672 : 모니터링 플러그인 콘솔에 접속하기 위한 포트

 

2. RabbitMQ 설치

1) RabbitMQ Repository 추가하기

curl -s https://packagecloud.io/install/repositories/rabbitmq/rabbitmq-server/script.deb.sh | sudo bash

 

2) 패키지 리스트 업데이트 및 설치하기

sudo apt update
sudo apt install rabbitmq-server

 

3) 상태 확인하기

sudo systemctl status rabbitmq-server.service

 

4) Management UI 플러그인 활성화하기

sudo rabbitmq-plugins enable rabbitmq-management

 

 

5) 사용자 추가하기

❶ 사용자 목록 조회

sudo rabbitmqctl list_users

 

❷ 사용자 추가

sudo rabbitmqctl add_user [username] [password]
sudo rabbitmqctl add_user admin admin1234

 

❸ 사용자 권한 부여

sudo rabbitmqctl set_user_tags [username] administrator
sudo rabbitmqctl set_user_tags admin administrator

 

 

6) 모니터링 콘솔 화면 접속하기

콘솔 화면에 접속해서 위에 추가한 Username과 Password를 입력한 후 로그인한다.

http://[PublicIP]:15672

 

만약, 연결 거부 화면이 뜬다면 플러그인이 실행되고 있는지 확인하자.

rabbitmq-plugins list

 

이때 제대로 실행이 되고 있지 않다면 Status([ ]) 칸이 모두 빈칸일 것이다.

 

다시 플러그인을 활성화하고 리스트를 확인하면 아래와 같이 Status가 변경된 것을 확인할 수 있다. 이때 다시 콘솔 화면으로 접속해서 정상적으로 화면이 뜬다면 로그인하면 된다.

sudo rabbitmq-plugins enable rabbitmq_management

 

 

프로젝트에 적용하기

1. 의존성 추가하기

# build.gradle
implementation 'org.springframework.boot:spring-boot-starter-amqp'
testImplementation 'org.springframework.amqp:spring-rabbit-test'

 

2. 연결 설정하기

사용자를 추가했을 때 설정했던 username과 password를 기입하면 된다.

# application.yml

spring:
  rabbitmq:
    host: [Public IP]
    port: 5672
    username: [username]
    password: [password]

 

3. 설정 클래스 생성하기

@Configuration
public class RabbitmqConfig {

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

    @Bean
    public Jackson2JsonMessageConverter jsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }
}

String이 아닌 Object를 전달하기 위해서는 반드시 Jackson2JsonMessageConverter를 @Bean으로 등록해야 한다.

 

 

4. Exchange / Queue / Routing key(binding) 설정하기

특정 Exchange로 메시지를 보냈을 때 어떤 routing key로 어떤 Queue에 전달할 것인지를 설정하면 된다.

출처 :&nbsp;https://www.rabbitmq.com/tutorials/amqp-concepts.html

 

위에 설정 클래스 내부에서 설정한다면 아래와 같이 코드를 작성하면 된다.

@Configuration
public class RabbitmqConfig {
	
    ...
    
    private static final String EXCHANGE = "exchange.name";
    private static final String QUEUE = "queue.name";
    private static final String ROUTING_KEY = "routing.name";

    @Bean
    TopicExchange exchange() {
        return new TopicExchange(EXCAHGE);
    }

    @Bean
    Queue queue() {
        return new Queue(QUEUE);
    }

    @Bean
    Binding binding(Queue queue, TopicExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY);
    }
    
    ...
    
}

 

나는 관리자용 콘솔 화면에서 직접 등록해줬다.

 

1) Exchange 추가하기

 

2) Queue 추가하기

 

3) Binding 추가하기 

 

 

5. Producer 생성하기

여기서는 원하는 방식으로 객체를 생성해서 전달하면 된다. 

@Component
@RequiredArgsConstructor
public class FeedMessageProducer {

    private final RabbitTemplate rabbitTemplate;

    private static final String EXCHANGE = "budsdom.events";
    private static final String SCHEDULE_ROUTING_KEY = "schedule.events";

    public void produceScheduleEvent(Long senderId, Long receiverId, Supplier<String> messageSupplier) {
        rabbitTemplate.convertAndSend(EXCHANGE, SCHEDULE_ROUTING_KEY, ScheduleEventDto.builder()
            .senderId(senderId)
            .receiverId(receiverId)
            .feedMessage(messageSupplier.get())
            .build()
        );
    }
}

 

6. Consumer 생성하기

메시지를 전달 받아 어떤 작업을 할지 로직을 작성하면 된다. 

@Component
@RequiredArgsConstructor
public class FeedMessageConsumer {

    private final FeedService feedService;

    private static final String SCHEDULE_QUEUE = "schedule.events";

    @RabbitListener(queues = SCHEDULE_QUEUE)
    public void consumeFromScheduleEvents(ScheduleEventDto scheduleEventDto) {
        feedService.saveFeed(
            scheduleEventDto.senderId(),
            scheduleEventDto.receiverId(),
            scheduleEventDto.feedMessage()
        );
    }
}

 

7. 활용하기

위의 작업이 모두 끝나게 되면 특정 이벤트가 발생했을 때 위에 생성한 Producer를 활용해서 보내고 싶은 정보를 전달할 수 있다. 나는 Producer에서 DTO 객체를 생성하기 때문에 아래와 같이 객체를 생성하기 위한 값들을 전달하도록 했다.

@Service
@RequiredArgsConstructor
public class MenteeScheduleService {

    private final FeedMessageProducer feedMessageProducer;

    @Transactional
    public MenteeScheduleResponseDto saveMenteeSchedule(long menteeId, long coachingScheduleId) {

        CoachSchedule coachSchedule = 
        	coachScheduleService.getCoachSchedule(coachingScheduleId,false);

        checkMenteeScheduleNotExist(coachingScheduleId);

        MenteeSchedule menteeSchedule = MenteeSchedule.of(coachingScheduleId, menteeId);
        menteeScheduleMapper.save(menteeSchedule);
        coachScheduleService.updateMatchYn(coachingScheduleId, true);

        // Producer를 활용해 원하는 정보 전달하기
        String dateTime = messageUtil.convertToString(coachSchedule.getPossibleDateTime());
        feedMessageProducer.produceScheduleEvent(menteeId, coachSchedule.getCoachId(), () ->
            MessageFormat.format(CREATE_SCHEDULE, dateTime));

        return MenteeScheduleResponseDto.from(menteeSchedule);
    }
    
}

 

 

 

마무리

사용해야 하는 이유를 찾고 선택하는 과정을 연속해서 경험하면서 여러 가지를 고민해 볼 수 있어서 재미있던 작업이었다. 이 포스트에는 담지 않았지만 Exchange 타입에도 여러 가지가 있어서 고려해야 했고 이번 구현에서는 Queue를 하나만 사용하지만 확장성과 유연성을 고려해서 여러 개 만들어서 사용해야 할지에 대한 것도 고민했었다. 실무에서는 훨씬 더 복잡하게 사용할 것 같은데 글또 반상회 때 관련 발표를 듣고 나서 그런지 더 흥미로움을 느끼고 궁금해진 것 같다. 

 

그리고 RabbitMQ와 별개로 이번 기능 구현을 하면서 책임과 역할을 나누는 것의 중요성이나 Functional Interface를 어떻게 활용하면 좋을지에 대한 것들을 배울 수 있었다. 이것도 따로 글을 작성해봐야겠다. 

 

 

 

참고

• Install RabbitMQ Server on Ubuntu on 22.04|20.04|18.04
• [RabbitMQ] Jackson2JsonMessageConverter
• Spring Boot와 RabbitMQ 초간단 설명서
• Message Broker - 왜 사용하는 것일까?