[Spring]메세지 브로커 - RabbitMQ

RabbitMQ란?

RabbitMQ는 분산 시스템에서 비동기 메시지 통신을 가능하게 해주는 메시지 브로커다.

쉽게 말해, A서비스에서 B서비스로 어떠한 데이터를 전달하는 구조에서 A서비스와 B서비스의 중간에 위치하여 데이터 전달을 조절하는 역할을 한다.

 

그러면 왜 굳이? 라는 생각이 들 수 있다. A서비스에서 B서비스로 데이터가 전달되는데 왜 중간에 끼어 컨트롤을 담당할까?

예를 들어, 좋아요 기능에 대해 생각해보자.

인스타나 유튜브 같은 사용자가 엄청나고 좋아요가 동시에 엄청나게 몰리는 서비스가 있다.

이때 수 천명의 사용자가 동시에 좋아요를 누르면 서버는 이를 받아 처리하고 데이터베이스에 저장 해야한다. 

그런데 이 좋아요가 동시에 쏟아지면 서버는 멀티 스레드로 어느정도 감당한다 치더라도 데이터베이스는

한번에 쏟아지는 데이터를 처리하느라 성능이 저하되고 또한 일부 데이터를 저장하지 못할 수도 있다.

이때, 래빗엠큐는 이 데이터베이스 쪽에 생긴 병목현상을 해결할 수 있다. 서버 A가 동시에 쏟아진 좋아요 데이터를

래빗엠큐에 전달하면 래빗엠큐는 그 데이터를 하나씩 큐에 넣고 대기한다.  그리고 다시 그 큐를 구독하고 있는 다른 서버B 메시지를 하나씩 전달받아 처리하고 db에 전달해 저장시킨다.

이렇게 실행 시점을 분리한 비동기 처리 구조로 데이터베이스는 한 번에 몰리는 트래픽에도 안정적으로 모든 좋아요 데이터를 저장할 수 있게 되는 것이다.

 

---

 

필요 개념

Queue

메시지가 실제로 저장되는 공간.

Consumer가 메시지를 가져가지 전까지는 메세지는 이 Queue에 쌓인다.

First In, First Out 즉 선입선출의 구조

producer

메세지를 보내는 쪽.

어떤 데이터 혹은 이벤트를 메시지로 변환해서 RabbitMQ로 발행(publish)하는 주체다.

이때, 바로 Queue로 메세지를 전달하는 것은 아니다. Exchange로 전달된다.

Exchange

메시지를 받아서 어떤 Queue에 보낼지 결정하는 배달원.

Producer가 보낸 메세지는 가장 먼저 이 Exchange로 들어오고, Exchange는 

Binding과 Routing Key에 따라 메세지를 Queue로 보낸다.

 

Binding

Exchange와 Queue를 연결하는 규칙.

이 연결에서 라우팅 키를 조건으로 건다.

(아래 코드를 보면 이해가 될 것이다)

 

Routing Key

Producer가 메시지를 보낼 때 지정하는 주소 라벨같은 개념.

Exchange는 바운딩에 묶여있는 라우팅 키를 가지고 메시지를 어떤 큐로 보낼지 결정한다.

Consumer

큐에서 메시지를 꺼내서 처리하는 주체.

Consumer는 큐를 구독(Subscribe)하고 있다가, 큐에 메세지가 들어오면 바로 전달받는다. Consumer가 큐에 쌓인 메시지를 처리한 후 ACK(확인 응답)을 RabbitMQ에 보내야 메시지가 큐에서 삭제된다.

 

 

메시지 처리 구조
Producer -> Exchange ->Queue ->Consumer

 

 

---

먼저 알아야 할 문법

converAndSend()

• convertAndSend(라우팅 키 이름, 전송할 메시지)
  • rabbitTemplate.convertAndSend("queueName", message);
• converAndSend(익스체인지 이름, 라우팅 키 이름, 전송할 메세지)
  • rabbitTemplate.convertAndSend("exchangeName", "routing.key", message);

왜? 라우팅 키 이름에 .key를 쓰는지

 

전체적인 설정 코드

@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("${spring.rabbitmq.virtual-host}")
    private String virtualHost;
 
    @Bean
    public ConnectionFactory connectionFactory(){
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setHost(host);
        connectionFactory.setPort(port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setVirtualHost(virtualHost);
        return connectionFactory;
    }
-->레디스에서 물리적인 연결은 LettuceConnetionFactory객체가 한 것처럼, 
래빗엠큐에서는 CachingConnectionFactory객체가 담당한다.     
-->레빗엠큐의 설정값들을 바탕으로 물리적인 연결 객체를 만든다.

    @Bean 
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory){
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
        -->객체를 자동으로 JSON으로 직렬화,역직렬화 한다는 뜻
        return rabbitTemplate;
    }
-->물리적인 연결을 바탕으로 래빗엠큐를 통해 메세지를 주고 받는 RabbitTemplate객체를 만든다.
   

	@Bean
	public Queue orderAddQueue() {
    	return new Queue("order.add.queue", true); // durable = true
	}

    //다이렉트 타입(라우팅 키가 정확히 일치하는 큐에 전달)의 Exchange 생성
    @Bean
   public DirectExchange directExchange(){
        return new DirectExchange("order.exchange");
    }
    
        //바인딩 규칙 설정. program.add.change와 program.add.queue를 program.add.key로 바인딩. 즉 program.add.queue에서는 라우팅 키가 program.add.key인 메시지만 받겠다는 의미
    @Bean
    public Binding binding(Queue orderAddQueue, DirectExchange directExchange){
        return BindingBuilder.bind(orderAddQueue)
                .to(directExchange)
                .with("order.add.key");
    }
    }

 

 

 

---

1.큐 만드는 코드

@Bean
public Queue orderAddQueue() {
    return new Queue("order.add.queue", true); // durable = true
}

 

 Queue 생성자 옵션

-name               ex."order.add.queue"

-durable            ex.true 

 

큐 이름을 정해 놓은 법칙은 없으나 실무에서는 일반적으로 영어소문자와 .(점)으로 큐 이름만 보고 어떤 역할인지 알 수 있도록 설정한다.

 

 

2.익스체인지 만드는 코드

    //다이렉트 타입(라우팅 키가 정확히 일치하는 큐에 전달)의 Exchange 생성
    @Bean
   public DirectExchange directExchange(){
        return new DirectExchange("order.exchange");
    }

 

 

3.바인딩 만드는 코드 여기에다가 라우팅 키를 설정한다.

    //바인딩 규칙 설정. program.add.change와 program.add.queue를 program.add.key로 바인딩. 즉 program.add.queue에서는 라우팅 키가 program.add.key인 메시지만 받겠다는 의미
    @Bean
    public Binding binding(Queue orderAddQueue, DirectExchange directExchange){
        return BindingBuilder.bind(orderAddQueue)
                .to(directExchange)
                .with("order.add.key");
    }

 

 

 

*추후 다른 이벤트에 대한 큐설정을 하려면 예를 들어, 주문 취소 이벤트에 대한 로직을 추가한다면

order.cancel.queue라는 새로운 큐를 추가하고, 지금 존재하는 order.exchange를 공유한다음 order.cancel.key라는 새로운 라우팅 키로 orderCancleQueue와 바인딩하는 코드를 추가하면 될 것이다. 

 

위와 같은 설정을 한 뒤, 실제 메시지를 보낼때는 RabbitTemplate.converAndSend()함수를 호출하면 된다.

우리는 아래의 2번째를 사용하면 된다. 첫번째는 따로 익스체인지와 바인딩을 정의하지 않을 때 사용된다.

converAndSend()

• convertAndSend(라우팅 키 이름, 전송할 메시지)
  • rabbitTemplate.convertAndSend("queueName", message);
• converAndSend(익스체인지 이름, 라우팅 키 이름, 전송할 메세지)
  • rabbitTemplate.convertAndSend("exchangeName", "routing.key", message);

 

---

위의 코드는  큐 뿐만 아니라 익스체인지와 바인딩까지 정의함으로써 확장성 있는 구조다.

사실 큐만 선언해도 작동은 한다. 왜냐하면 이는 래빗엠큐에서는 별도의 Exchange를 선언하지 않아도, 항상 이름이 " "(빈 문자열)인 Default Exchange가 존재하기 때문이다. 이런 상황에서는 Producer가 메세지를 보내면 Default Exchange로 가게 되고  Default Exchange는 라우팅 키를 보고 어떤 큐로 보낼지 판단하는데 이때는 라우팅 키를 또 별도로 설정하지 않아도 큐이름으로 보내게 되면 자동으로 해당 큐로 보낸다.(그러면 이때는 라우팅 키를 따로 설정하지 않아도 큐이름이 라우팅 키가 되는 것이다)

 

사실 Queue, Exchange, RoutingKey를  위의 예시처럼 코드로도 만들 수 있으나  localhost:15672 로 접속해서 UI상에서 만들어 놓을 수도 있다.

(그러나 프로젝트에서 다른 팀원들도 명확하게 확인할 수 있도록 되도록이면 코드로 생성해서 코드를 남기는 것을 권한다)

 

1.Exchange나 RoutingKey를 화면으로 만드는 경우

1.Name은 exchange 이름을 설정하는 곳이다. 

2.Type은 Exchange의 라우팅 방식을 지정하는 곳이다

• direct : 라우팅 키가 정확히 일치하는 큐에 전달
• fanout: 연결된 모든 큐에 브로드캐스트로 전달
• topic :  라우팅 키 패턴(*,#)으로 매칭
• headers: 메시지 헤더 조건으로 매칭

우리는 코드에서 direct 타입의 exchange를 만들었었다.

 

3.Durability

• Durable : 래빗엠큐 서버를 재시작해도 Exchange가 남음
• Transient(Non-durable) :서버 재시작하면 Exchange가 사라짐

4.Auto delete

Yes로 설정하면, 이 Exchange와 연결된 큐/바인딩이 하나도 없을 때 자동으로 삭제된다.

 

5.Internal

Yes 로 설정하면, 클라이언트(Producer)가 직접 메시지를 publish할 수 없으므로 특수한 경우를 제외하고는 보통 사용x

다른 Exchange에서 이 Exchange로만 메세지를 보낼 때 사용하는 듯 하다.

 

6.Arguments

추가 설정을 넣는 곳으로 대표적인 옵션이

 

---

 

 

리슨 설정

@RabbitListener(queues = "order.add.queue")
public void handleProgramAdd(OrderAddEvent event) {
    System.out.println("프로그램 추가 이벤트: " + event.getOrderId());
}

 

스프링에서는 @RabbitListener를 통해 리스너설정을 쉽게 할 수 있다.

이렇게 큐를 구독하고 있으면

이 큐에 메시지가 생기면 래빗엠큐가 다시 이 리스너(스프링)에게 전달하여 해당 로직이 실행된다.

 

좀 더 내부적으로 들어가자면 이 리스너 설정을 통해 해당 큐에 메시지가 들어오면 해당 큐를 구독하고 있는 이 리스너에서 메세지를 받는다. 이때 리스너 메서드가 로직을 다실행하고 메서드가 정상 종료되면,
큐에게 메시지를 잘 받았고, 처리까지 끝냈다는 신호인 ACK(acknowledgement)를 보낸다.

큐는 이 ACK를 받으면 해당 메세지를 큐에서 삭제시킨다.

 

ACK를 보내는 것은 우리가 따로 yml파일에 수동설정을 하지않는 이상 자동으로 ACK가 큐에 전송되기 때문에 우리가 따로 신경쓰지 않아도 된다.