Springboot에서 Docker RabbitMQ로 메세지 송수신하기

RabbitMQ는 AMQP 서버입니다. AMQP는 Advanced Message Queueing Protocol의 준말이며, 메시지 큐 기반의 프로토콜을 뜻합니다. 메시지 큐를 이용하면 메시지 발행-구독(Publish-Subscribe) 패턴을 사용할 수 있습니다. 그렇다면, 메시지 큐는 왜 사용하는 것일까요? 동기식으로 데이터를 주고받을 경우, Request 전송 후 Response가 올 때까지 대기해야 합니다. Response를 받는데 까지 걸리는 시간 동안 컴퓨터는 아무런 연산처리를 하지 못하므로, 비동기식에 비해 처리 성능이 떨어집니다. 비동기 방식은 요청을 보내고 다른 작업을 진행하다 응답이 왔을 때, 그때 처리하는 방식입니다. 메시지 큐는 비동기 통신을 지원하므로, 발행(Publish)된 데이터가 있을 경우에만 구독(Subscribe)하던 Topic의 데이터를 처리하면 되므로, 처리 성능의 이점을 확보할 수 있습니다.

이번 포스팅에서는 Springboot에서 메시지 큐 프로토콜 서버인 RabbitMQ와 연동하는 방법에 대해 알아보겠습니다.

Springboot RabbitMQ용 프로젝트 다운로드

https://start.spring.io/ 사이트에서 Spring for RabbitMQ 디펜던시를 추가하여 프로젝트를 다운로드합니다.

스프링 프로젝트 다운로드

build.gradle 파일을 열어보면, 아래와 같이 RabbitMQ에 관한 디펜던시가 추가된 것을 확인할 수 있습니다. RabbitMQ는 AMQP서버라서 디펜던시 이름에도 amqp가 붙어있네요.

implementation 'org.springframework.boot:spring-boot-starter-amqp'

도커를 이용한 RabbitMQ 실행하기

RabbitMQ는 AMQP 서버이므로, 별도의 서버를 실행해야 합니다. RabbitMQ 서버를 다운로드하여 설치 후 실행하셔도 되지만, 도커를 이용하면 손쉽게 RabbitMQ 서버를 컨테이너로 구동하므로, 도커로 RabbitMQ를 실행해보겠습니다. 도커 설치 및 실행방법은 아래 링크 참조해주세요. 

[OS] 윈도우(Windows)에서 도커(Docker) 설치하기

RabbitMQ 이미지는 Docker Hub에서 다운로드를 합니다. Docker Hub 사이트에 들어가면 도커에서 제공하는 다양한 이미지를 볼 수 있습니다. 여기서 이미지는 사진을 뜻하는 게 아니며, 이미지의 정의는 소프트웨어 패키지를 Snapshot 형태로 저장하여 관리하는 파일 단위라고 이해하시면 됩니다. 도커 허브 링크는 https://hub.docker.com/이고, 사이트에 접속하지 않아도 github처럼 명령어 만으로, 이미지를 다운로드 및 설치할 수 있습니다. 도커 허브에 명시된 RabbitMQ 다운로드 명령어는 아래와 같습니다.

docker pull rabbitmq

RabbitMQ 이미지 다운로드가 완료된 후, 컨테이너로 구동해보겠습니다. RabbitMQ 서버를 구동하는 도커 명령어는 어디서 볼 수 있을까요? 도커 이미지를 제공하는 도커 허브 사이트에서 RabbitMQ 이미지 사이트에 접속하면, 이미지에 대한 설명이 나와있습니다. 아래 링크는 RabbitMQ 이미지에 관해 설명이 적혀있는 링크입니다.

Rabbitmq - Official Image | Docker Hub

아래 명령어를 통해 Management Plugin이 설치된 RabbitMQ 이미지를 다운로드하고 실행해보겠습니다. Docker ps를 입력하면, 현재 도커에서 실행 중인 컨테이너 목록을 볼 수 있습니다.

docker run -d --hostname TestRabbitMQ --name ExecRabbitMQ -p 5672:5672 -p 8080:15672 rabbitmq:3-management
docker ps

RabbitMQ 실행 후, http://localhost:8080에 접속하면, RabbitMQ 관리 화면에 진입 가능합니다. ID는 guest, PASSWORD는 guest 입력하면 로그인 가능합니다.

RabbitMQ 관리용 화면

Docker Desktop에서 RabbitMQ 컨테이너 실행의 Inspect를 보면, Port 사용 여부를 확인할 수 있습니다.

RabbitMQ port binding 결과를 Docker Desktop에서 확인

RabbitMQ 서버를 구동하였으니, 다음으로 Springboot에서 RabbitMQ에 접근하는 코드를 작성해보겠습니다.

Springboot에서 RabbitMQ API 코드 작성하기

RabbitMQ서버는 메시지를 발행하고 구독할 때 중개자 역할을 하기 때문에 Broker라고 칭합니다. SpringBoot는 발행자(Publisher), 구독자(Subscriber)가 되고, RabbitMQ 서버는 Broker가 되는 것이죠. 메시지 구독을 위한 Receiver 클래스를 작성해보겠습니다.

package com.example.demo;
import java.util.concurrent.CountDownLatch;
import org.springframework.stereotype.Component;

@Component
public class Receiver {

    private CountDownLatch latch = new CountDownLatch(1);

    public void receiveMessage(String message) {
        System.out.println("Received <" + message + ">");
        latch.countDown();
    }

    public CountDownLatch getLatch() {
        return latch;
    }

}

Receiver 클래스는 단순히 수신받은 메시지를 콘솔 창에 출력하고 latch의 count를 감소합니다. 다음으로 메시지를 발행하는 Runner 클래스를 만들어보겠습니다.

package com.example.demo;

import java.util.concurrent.TimeUnit;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

@Component
public class Runner implements CommandLineRunner {

    private final RabbitTemplate rabbitTemplate;
    private final Receiver receiver;

    public Runner(Receiver receiver, RabbitTemplate rabbitTemplate) {
        this.receiver = receiver;
        this.rabbitTemplate = rabbitTemplate;
    }

    @Override
    public void run(String... args) throws Exception {
        System.out.println("Sending message...");
        rabbitTemplate.convertAndSend(DemoApplication.topicExchangeName, "foo.bar.baz", "Hello from RabbitMQ!");
        receiver.getLatch().await(10000, TimeUnit.MILLISECONDS);
    }

}

CommandLineRunner 인터페이스를 상속받아 run메서드를 구현하였는데, 이는 Spring Beans가 로딩된 후 run()을 실행합니다. RabbitTemplate의 convertAndSend()에서 Topic, Key, Message를 지정하여 메시지를 발행하고 있습니다. 그리고, Receiver 객체가 가진 latch가 활성화될 때까지, await 합니다.

다음으로 메시지 발행자, 구독자, 그리고 메시지 큐를 스프링 부트에 등록하는 소스코드를 작성해보겠습니다.

package com.example.demo;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;

@SpringBootApplication
public class DemoApplication {

    static final String topicExchangeName = "spring-boot-exchange";

    static final String queueName = "spring-boot";

    @Bean
    Queue queue() {
        return new Queue(queueName, false);
    }

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

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

    @Bean
    SimpleMessageListenerContainer container(ConnectionFactory connectionFactory,
                                             MessageListenerAdapter listenerAdapter) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.setQueueNames(queueName);
        container.setMessageListener(listenerAdapter);
        return container;
    }

    @Bean
    MessageListenerAdapter listenerAdapter(Receiver receiver) {
        return new MessageListenerAdapter(receiver, "receiveMessage");
    }

    public static void main(String[] args) {
        SpringApplication.run(DemoApplication.class, args).close();
    }

}

소스 설명에 앞서 Message Queue가 메시지를 생성 및 최종 수신하는 프로세스와 용어에 대해 알아보겠습니다. 메시지의 생성 및 수신 흐름은 Publisher ---> Exchange ---> Queue ---> Consumer입니다. 용어에 대한 설명은 아래와 같습니다.

• Publisher : 메시지를 발행하는 애플리케이션
• Exchange : 발생된 메시지를 Queue에 저장하는 모듈
• Queue : 메시지를 저장하는 버퍼
• Route : Exchange 모듈이 Queue로 메시지를 전달할 때 사용하는 Key
• Consumer : 발행된 메시지를 사용하는 애플리케이션
• Subscribe : 메시지 발행 여부를 감지하기 위해 리스닝하는 과정
• Topic : 라우팅 키 패턴과 일치하는 Queue에 모두 메시지를 전송함. Multicast 방식

다시 소스코드로 돌아오면, exchange() 함수로 Topic을 지정하고, queue()에서 Queue이름을 정해줍니다. 다음으로, binding()에서 queue와 topic 그리고 이를 연결할 Route Key인 "foo.bar.#"으로 연결해줍니다. listenerAdapter()에서 메세시 수신 시, 응답할 리스너 메서드를 등록해줍니다. 다음으로 container() 메서드가 호출되어, RabbitMQ서버와 연결, 그리고 메세시 수신할 큐 그리고 메시지 수신 시 실행할 리스너 메서드를 지정해줍니다. Queue, Exchange, Route Key 용어에 대한 설명은 아래와 같습니다.

DemoApplication을 실행하면, RabbitMQ에 메시지 큐를 생성하고 큐를 통해 메시지 송신, 그리고 큐에 쌓인 메시지를 받아 출력하는 로그를 확인할 수 있습니다.

Sending message...
Received <Hello from RabbitMQ!>

RabbitMQ 관리 화면에 접속하면, 생성된 Queue와 Topic 그리고 RouteKey값을 확인할 수 있습니다.

RabbitMQ 메세지 전송 이력 확인

이상으로 Springboot와 RabbitMQ를 연동하는 방법에 대해 알아보았습니다.