Spring Cloud Stream으로 RabbitMQ 샘플 구축하기

🧨 RabbitMQ란

---

AMQP를 따르는 오픈 소스 메시지 브로커이다.

요청에 대한 처리 시간이 길 때, 해당 요청을 Queue에 위임하고 빠르게 응답할 때 사용한다.

또한, MQ를 사용하여 애플리케이션 간 결합도를 낮출 수 있다.

🧨 AMQP란

---

Advaced Message Queing Protocol의 약자로써, MQ의 표준 프로토콜이다.

이를 기반으로 나온 제품 중 하나가 RabbitMQ이다.

등장 배경

이전에는 각 플랫폼에 종속적인 제품들이었기 때문에 이기종 간 메시지 교환을 위해서

포맷 컨버전을 해야 했기 때문에 메시지 브릿지를 이용하거나(속도 저하 발생) 시스템 자체를 통일 시켜야 하는

불편함과 비효율성이 있었다.이러한 단점을 보완하기 위해 나온 표준 프로토컬이 AMQP이다.

즉, AMQP가 나온 목적은 이기종 간 시스템간에 최대한 효율적인 방법으로 메시지를 교환하기 위해서다.

AMQP 충족 조건

특정 벤더에 종속되는 것을 방지하기 위해 아래 조건을 충족한다.

• 모든 브로커들은 똑같은 방식으로 동작할 것
• 모든 클라이언트들은 똑같은 방식으로 동작할 것
• 네트워크 상으로 전송되는 명령어들의 표준화
• 프로그래밍 언어 중립

docker run

---

docker run -d --hostname cloud-stream-rabbit --name cloud-stream-rabbit -p 15672:15672 -p 5672:5672 rabbitmq:3-management

🧨 @Async가 아니고 왜 MQ인가

---

@Async로 할 경우는 문제점

• 서버가 다운될 경우 모든 쓰레드는 사라지기 때문에 데이터 유실
• 실패한 메시지를 DLQ와 같은 곳에 보존할 수 없다.
• 재시도가 번거롭다.

DB Queue가 아닌 이유

기존 php에서는 DB Queue를 사용하고 있었다.

하지만 스프링에서는 이를 지원하지 않기 때문에 굳이 구현할 수고를 덜기 위해

spring cloud stream에서 제공하는 RabbitMQ를 사용한다.

kafka는 비용이 발생하기 때문에 사용하지 않는다.

🧨 재시도 전략

---

재시도 전략이란?

재시도 전략은 consume이 실패했을 경우 몇변을 재시도 하고, 재시도 간격은 어떻게 할지 등의 전략이다.

우리 서비스에서 재시도 전략

크게 복잡하게 가져갈 상황은 아니라서 아주 간단하게 가져간다.

• 재시도는 일단 1번으로 한다.
• DLX로 전송한다.
• DLQ에 담긴 메시지를 수동으로 처리하거나 스크립트를 만든다.

🧨 에러 핸들링 전략

---

에러 핸들링 전략이란

메시지 consume이 실패할 경우 메시지를 버릴지, 보존할지 보존한다면 어떻게 보존할지 등의 전략이다.

우리의 전략

우리는 DLX를 활용한다.

DLX에 전송된 메시지는 수동으로 처리할 수도 있고, 자동화 스크립트를 개발할 수도 있다.

전략 종류

전략 종류에는 대략 크게 3가지가 있다.

• Message Drop: 실패한 메시지는 폐기한다.
• Re Queue Message: 실패한 메시지를 다시 queue에 넣는다.
• Dead Letter Queue: 실패한 메시지는 DLQ로 보낸다.

Message Drop

특별할 것 없이 메시지를 폐기하고 끝이다.

Re Queue Message

메시지를 다시 queue에 넣는데 무작정 계속 반복하면 해당 메시지는 평생 queue에 남아있게 된다.

그렇기 때문에 메시지 기간이나 시도 횟수에 따라서 핸들링하는 등의 행위가 필요하다.

Dead Letter Queue(사용할 전략)

실패한 메시지를 보관하는 별도의 queue이다.

DLQ로 메시지가 들어올 경우 개발자에게 알림을 가게 하는 등의 처리로 수동으로 재시도하는 등의

방식을 쓸 수 있다. 아니면 이를 자동화하는 코드를 작성해도 좋을 것 같다.

Error Handling

DLQ를 사용하지 않거나 Re Queue를 하지 않을 경우 별도의 핸들링이 필요할 것이다.

그럴 경우 ErrorMessage용 consumer를 만들어서 핸들링할 수 있다.

대략 살펴보면 아래와 같이 핸들링할 수 있고, 더 깊은 내용은 다루지 않는다.

java code

@Bean
public Consumer<Employee> sample() {
  return (employee) -> {
    System.out.println(employee.toString());
    if (employee.getAge() > 100) {
      throw new RuntimeException("occured consumer");
    }
  };
}

@Bean
public Consumer<ErrorMessage> errorHandler() {
  return e -> {
    errorOccur++;
    System.out.println("에러 발생: " + e);
  };
}

• errorHandler()
  • 이 역시 똑같은 Consumer이기 생성되는 exchange는 errorHandler-in-0이다.
  • 다만 T 타입만 ErrorMessage일 뿐이다. 이렇게 타입만 바꿔주면 알아서
    에러 전용 excahnge와 consumer가 반인딩 된다.
  • 재시도까지 모두 실패했을 경우 에러로 판단해서 그 때 queue에 담겨서 consume하게 된다.
  • 물론 이렇게 해도 아무 처리도 하지 않기 때문에 메시지가 버려지는 것은 똑같다.

🧨 용어 정리

---

Connection

단일 TCP 연결

• TCP 기반
• 효율성을 위해 수명이 긴 연결을 한다. 즉 프로토콜 작업당 새 연결이 열리지 않는다.
• 하나의 클라이언트는 단일 TCP 연결을 사용한다.
• 즉, 커넥션 풀처럼 커넥션을 유지한다.
• 예를 들어 a 서버에서 10개의 커넥션을 한 번 만들었다면 10개가 계속 유지된다.

AMQP 0-9-1

• 하나의 커넥션에 Channel이라고 하는 여러 경량 연결을 열 수 있다.
  이 프로토콜을 AMQP 0-9-1이라고 한다.
• AMQP 0-9-1는 하나 이상의 채널을 열고 채널에서 프로토콜 작업(구독, 소비)를 수행한다.

Channel

• channel은 connection없이는 존재할 수 없다.
• 단일 TCP 연결에 경량 연결을 Channel이라고 한다.
• 모든 작업은 채널에서 발생한다.
• 최대 채널 수를 자바에서 제어할 수 있다.
• 송신을 하면 아래와 같이 pulisher 전용 채널이 생긴다. publish channel은 consume channel과는
  다르다. 별개이다.
• 프로세스와 쓰레드와 비교하자면, connection은 프로세스이고, channel이 쓰레드와 비슷하게 생각할 수 있을 것 같다.

🧨 python으로 connection, channel 알아보기

---

1. consum 코드 작성

    #!/usr/bin/env python
    import pika, sys, os
   
    def main():
        def callback(ch, method, properties, body):
            print(" [x] Received %r" % body)
   
            # 커넥션 및 채널 생성
        connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
        channel = connection.channel()
   
            # 큐 선언, 익스체인지 선언, 큐 바인딩
        channel.queue_declare(queue='hello')
        channel.exchange_declare(exchange='defaultExchange',exchange_type='direct')
        channel.queue_bind(queue='hello', exchange='defaultExchange', routing_key='defaultRoutingKey')
   
            # 컨수머 센팅
        channel.basic_consume(queue='hello', on_message_callback=callback, auto_ack=True)
   
            # 컨숨 시작
        print(' [*] Waiting for messages. To exit press CTRL+C')
        channel.start_consuming()
   
    if __name__ == '__main__':
        try:
            main()
        except KeyboardInterrupt:
            print('Interrupted')
            try:
                sys.exit(0)
            except SystemExit:
                os._exit(0)

2. receiver.py

3. consume 실행

   • 실행하게 되면 컨숨을 하기 위해 대기하게 된다.
     
   • 브로커에 defaultExchange, queue, binding이 모두 생성된다.(매니지먼트 툴에서 확인 해보면 된다.)

4. python receiver.py

5. publisher 코드 작성

    #!/usr/bin/env python
    import pika
   
    if __name__ == "__main__":
   
        connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
        channel = connection.channel()
   
        channel.queue_declare(queue='hello')
        channel.exchange_declare(exchange='defaultExchange', exchange_type='direct')
        channel.queue_bind(queue='hello', exchange='defaultExchange', routing_key='defaultRoutingKey')
   
        channel.basic_publish(exchange='defaultExchange', routing_key='defaultRoutingKey', body='Hello World!')
        print(" [x] Sent 'Hello World!'")
        connection.close()

   receiver코드와 거의 동일하고 큐, 익스체인, 바인딩 코드가 중복된다.그런데 send.py가 먼저 실행될지 receiver.py가 먼저 실행될지는 알 수 없기 때문에 동일한 셋팅이 들어가게 된다.

6. 해당 코드는 브로커에 큐, 익스체인지, 바인딩이 있는지 없는지 판단하고 있으면 그대로 사용하고 없으면 새로 만든다.

7. send.py

8. publisher 실행

   • publishing을 했으므로 대기 중이던 consumer는 consume을 하게 된다.
     

     

9. python send.py

10. connections 및 channels 확인하기

위 사진을 보면 connection이 2개이다.

60370이 consume connection이고, 58306이 publish connection이다.

send.py, [receiver.py](http://receiver.py) 코드에서 각각 connection을 생성했기 때문에 그렇다.

publishing, consume 서버가 분리 되어 있다면 당연히 이런 구조가 나오는 것이다.

6. publishing, consume이 같은 connection의 같은 channel 사용하기

7. 위 사진과 같이 하나의 channel에서 publising, consume을 모두 하게 된다.

8. #!/usr/bin/env python import pika if __name__ == "__main__": connection = pika.BlockingConnection( pika.ConnectionParameters(host='localhost')) channel = connection.channel() channel.queue_declare(queue='hello') channel.exchange_declare(exchange='defaultExchange', exchange_type='direct') channel.queue_bind(queue='hello', exchange='defaultExchange', routing_key='defaultRoutingKey') # publisher channel.basic_publish(exchange='defaultExchange', routing_key='defaultRoutingKey', body='Hello World!') print(" [x] Sent 'Hello World!'") # consumer def callback(ch, method, properties, body): print(" [x] Received %r" % body) channel.basic_consume(queue='hello', on_message_callback=callback, auto_ack=True) channel.start_consuming() print('a') print('b') print('c') connection.close()

9. rabbitMQ가 추구하는 경량 connection 만들기

    #!/usr/bin/env python
    import pika
   
    if __name__ == "__main__":
   
        connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
        channel1 = connection.channel()
        channel2 = connection.channel()
   
        channel1.queue_declare(queue='hello1')
        channel1.exchange_declare(exchange='defaultExchange1', exchange_type='direct')
        channel1.queue_bind(queue='hello1', exchange='defaultExchange1', routing_key='defaultRoutingKey1')
   
        channel2.queue_declare(queue='hello2')
        channel2.exchange_declare(exchange='defaultExchange2', exchange_type='direct')
        channel2.queue_bind(queue='hello2', exchange='defaultExchange2', routing_key='defaultRoutingKey2')
   
        # publisher
        channel1.basic_publish(exchange='defaultExchange1', routing_key='defaultRoutingKey1', body='Hello World1!')
        channel2.basic_publish(exchange='defaultExchange2', routing_key='defaultRoutingKey2', body='Hello World2!')
        print(" [x] Sent 'Hello World!'")
        connection.close()

   위 코드와 같이 connection에서 channel만 새로 만드는 것이다. 그러면 connection 1개에 2개의 publishing channel이 만들어진다.
   

10. rabbitMQ는 경량 connection은 connection 하나에 여러개의 channel을 운영하는 것이다.

channel 상세하게 알아보기

• 구조
  • spring boot server 하나에서 pruducer, rabbitMQ, consume 모두 함.
  • exchage: defaultExchange, customExchange
  • queue: defaultExchange.defaultQueue, customExchange.customQueue

1. spring boot server 실행
   • connection 정보
     
   • channels 정보
     
   • queue, consumer 정보
     • channel: 62740(1)
       
     • channel: 62740(2)
       
   • connection: 62740
2. defaultExchange, customExchange에 모두 전송
   • connection 정보
   • server 실행 시 생겼던 connection(62740)에 channel(consume)이 1개 추가되고,
     62868 connection이 새로 생기면서 channel(publisher)이 1개 추가되어 총 4개의 channels가 된다.
     
   • channel 정보
     • 62868(1): publisher 라고 나와있음.
       
     • 62740(1), 62740(2), 62740(3)(1), (2)에 기존에 exchange가 연결되어 있었기 때문에 이 채널로 consume한다.
       
     • (1), (2)는 기존에 있던 것이고 (3)이 추가된 것인데 어떤 exchange도 연결되지 않았다. ???!@

channel 정리

• 62868(1) 채널은 publisher 라고 나와 있는데 왜 channel이 1개일까? exchange가 2개인데 2개 생겨야 하는거 아닌가?
• 62740(3)는 뭘까? 아무런 역할이 없어 보이는데?
• consumer는 각각의 채널을 이용한다라는게 맞을까?

exchange와 prefetch count

https://blog.dudaji.com/general/2020/05/25/rabbitmq.html

🧨 구조

---

모든 잡을 하나의 queue, exchange로 관리한다. 여거 큐를 관리하는 것 보다는 하나의 큐에서 관리하는 것이 편할 것으로 판단했고

분리가 필요하다면 그 때 가서 분리하는 방향을 생각해 보자.

🧨 구현 레벨

---

위의 도식과 같이 모든 작업을 하나의 exchange, queue에서 관리하기 위해서 리플렉션을 사용하는 것이 핵심히다.

원래는 어떤 잡을 실행할지 케이스 별로 if 문을 consumer에서 사용해서 코드가 굉장히 지저분 했는데

if 문이 필요한 부분을 producer가 push할 때 리플렉션을 payload로 넘겨 consumer를 라이브러리화 하여

심플하게 만들었다.

추가로 exchange, queue를 만들지 않을 경우 사용 방법

1. 클라이언트는 DefaultProducer를 주입받아 push를 한다.
   • push메서드 파라미터로는 JobSpringBean 클래스의 Method 타입과 method의 파라미터가 필요하다.
   • 파라미터는 기본 데이터 타입만 가능하다.
   • private final DefaultProducer defaultProducer; // bean public void client() throws NoSuchMethodException { defaultProducer.push( JobSpringBean.class.getMethod("jobMethod", Long.class), List.of(1L) ); }
2. Job을 실행할 JobSpringBean 클래스를 생성해서 로직을 작성한다.
3. @Component public class JobSpringBean { public void jobMethod(Long id) { Member member = findById(id); ... } }

리플렉션을 사용해서 개선 된 점을 정리 해보면 아래와 같다.

• consumer에서의 분기 처리 복잡도 단순화 되었다.

• 애초에 관리의 용이성을 위해서 하나의 exchange, queue로 모두 관리하는 것을 원했는데
  이를 리플렉션 없이 구현할 경우 consumer에 분기 처리가 너무 많아지게 된다.
  이를 개선하기 위해 프록시 패턴도 사용하고 했지만 그래도 결국 복잡한 것은 비슷하다.

• 개선 후에는 클라이언트에서 Job 클래스의 Method만 넘겨주면 되기 때문에 Consumer가 consume하는 역할만 하도록 개선 되었다.

• 전체 코드

  DefaultConsumer.java

    @Slf4j
    @RequiredArgsConstructor
    @Service
    public class DefaultConsumer {
  
      private final ApplicationContext applicationContext;
  
      @Bean
      public Consumer<DefaultPayload> defaultConsume() {
        return (message) -> {
          Object bean = applicationContext.getBean(message.getJobBeanClass());
          String methodName = message.getMethodName();
          List<Serializable> args = message.getNormalizedArgs();
  
          try {
            message.getJobBeanClass()
                .getMethod(methodName, message.getParameterTypes())
                .invoke(bean, args.toArray());
          } catch (IllegalAccessException | InvocationTargetException | NoSuchMethodException e) {
            throw new RuntimeException(e);
          }
        };
      }
    }

  DefaultProducer.java

    @RequiredArgsConstructor
    @Component
    public class DefaultProducer {
  
      private final StreamBridge streamBridge;
  
      public void push(Method method, List<Serializable> args) {
        Class<?> clazz = method.getDeclaringClass();
        String methodName = method.getName();
  
        DefaultPayload payload = new DefaultPayload(clazz, methodName, args);
            streamBridge.send("defaultExchange", MessageBuilder.withPayload(payload).build());
      }
    }

DefaultPayload.java

@NoArgsConstructor(access = AccessLevel.PUBLIC) // StreamBridge가 직렬화를 하기 위해서 기본 생성자가 필요함.
@Getter
public class DefaultPayload {

  private Class<?> jobBeanClass;
  private String methodName;
  private List<Serializable> args;
  private final List<Class<?>> argTypes = new ArrayList<>();

  public DefaultPayload(Class<?> jobBeanClass, String methodName, List<Serializable> args) {
    this.jobBeanClass = jobBeanClass;
    this.methodName = methodName;
    this.args = args;
    args.forEach(argValue -> this.argTypes.add(argValue.getClass()));
  }

  /**
   * consumer에서 역직렬화 시 Long type을 Integer로 받기 때문에 Integer를 Long으로 변환
   * args 필드를 깊은 복사해서 반환한다.
   */
  public List<Serializable> getNormalizedArgs() {
    int size = this.argTypes.size();
    return IntStream.range(0, size)
        .mapToObj(i -> convertArgIntegerToLong(this.argTypes.get(i), this.args.get(i)))
        .toList();
  }

  /**
   * @return Class<?>[]
   * reflection invoke를 위해서 배열이 필요하기 때문에 제공하는 메서드. consumer에서 사용
   */
  public Class<?>[] getParameterTypes() {
    return argTypes.toArray(Class[]::new);
  }

  private Serializable convertArgIntegerToLong(Class<?> argType, Serializable arg) {
    if (argType == Long.class && arg.getClass() == Integer.class) {
      return Long.valueOf(String.valueOf(arg));
    }

    return arg;
  }
}

application.yml 예제 코드

---

spring:
  cloud:
    function:
      definition: defaultConsume;customConsume
    stream:
      bindings:
        defaultConsume-in-0:
          destination: defaultExchange
          group: defaultQueue
          consumer:
            max-attempts: 1
        customConsume-in-0:
          destination: customExchange
          group: customQueue
          consumer:
            max-attempts: 1
      rabbit:
        bindings:
          consumeDefault-in-0:
            consumer:
              dead-letter-queue-name: defaultDlxQueue
              auto-bind-dlq: true
          customExchange-in-0:
            consumer:
              dead-letter-queue-name: customtDlxQueue
              auto-bind-dlq: true
  rabbitmq:
    addresses: localhost
    username: guest
    password: guest
    port: 5672