Rabbit MQ 메시지 브로커

steve·2023년 12월 12일

NestJS rabbitmq

Backend

목록 보기

6/17

목표

RabbitMQ와 다른 메시지 브로커의 차이점을 이해하고, 적합한 메시지 브로커를 선택할 수 있다.
RabbitMQ의 개요와 장점을 이해하고, 기능을 활용하여 분산 시스템을 구축할 수 있다.

메시지 브로커 비교

다운로드 수 비교
- 2023년 10월 기준 RabbitMQ (108만) > Kafka (86만) > MQTT (79만)
기능별 장단점 비교

RabbitMQ 선택 이유 : 범용성이 좋고 대규모 데이터 처리 목적이 아닌 일반 규모의 메시지 전송 처리에 적합

RabbitMQ 개요

RabbitMQ는 AMQP 프로토콜을 지원하는 오픈소스 메시지 브로커이다.
클라이언트로부터 메시지를 받아서, 이를 Queue에 저장하거나 다른 Queue로 전달하는 역할을 한다.
RabbitMQ는 다음과 같은 개념으로 구성된다.
- Producer: 메시지를 생성하는 클라이언트
- Consumer: 메시지를 소비하는 클라이언트
- Exchange: 메시지를 Queue에 전달하는 중계기
- Queue: 메시지를 저장하는 큐
- Binding: Exchange와 Queue를 연결하는 규칙

대표 기능

조건부 전달: 메시지를 특정 조건을 만족하는 Queue로 전달할 수 있다.
TTL: 메시지의 TTL(Time To Live)을 설정하여, 특정 시간 동안 소비되지 않은 메시지를 삭제할 수 있다.
ACK: Consumer가 메시지를 소비했음을 브로커에 알리기 위한 ACK 기능을 제공한다.
Dead Letter Queue: Consumer가 메시지를 처리하지 못하는 경우, 이를 Dead Letter Queue로 전달할 수 있다.

Queue 동작

Round-robin dispatching
- 프로세스들 사이에 우선순위를 두지 않고, 순서대로 시간단위(Time Quantum/Slice)로 CPU를 할당하는 방식의 CPU 스케줄링 알고리즘
- 위와 같은 Producer - Queue - 다중 Consumer 구조에서, 순서대로 메시지를 보내게 됨
- 평균적으로 모든 Consumer는 동일한 수의 메시지를 받음
Message acknowledgement
- 메시지가 손실되지 않음을 보장하기 위해 RabbitMQ는 ack (Consumer로부터 특정 메시지를 받았는지에 대한 여부를 RabbitMQ에 말해주는 것) 기능을 제공
- ack 기능을 활성화 시, Consumer가 어떤 ack도 전송하지 않고 연결이 끊길 경우 정상적으로 Queue가 처리되지 않음을 확인하고 다시 Queue로 보냄
- ack는 기본 타임 아웃 30분
Message Durability
- 메시지 브로커가 연결이 끊길 경우, 메시지 손실 방지를 보장하기 위해 durable 옵션을 제공
- 브로커 서버가 재시작되어도 Queue를 보존함
Fair dispatch
- round-robin 방식은 각 메시지 처리 시간을 고려하지 않기 때문에 Consumer가 둘 인 경우, 한 쪽으로 부하가 몰릴 경우가 발생
- prefetch 옵션으로 Consumer가 메시지 처리를 완료하지 않은 경우 메시지를 전달하지 않을 수 있음

옵션 설정 코드 예시 (noAck, durable, Fair dispatch)

const USER_SERVICE_PROXY =  {
  provide: 'USER_SERVICE',
  useFactory: (config: ConfigService) => {
    return ClientProxyFactory.create({
      transport: Transport.RMQ,
      options: {
        urls: [
          `amqp://${config.get('RABBITMQ_USER')}:${config.get(
            'RABBITMQ_PW'
          )}@${config.get('BROKER_HOST')}:${config.get('BROKER_PORT')}`,
        ],
        queue: 'user',
        noAck: true, // true인 경우, Consumer의 메시지 수신응답을 받지 않음
        queueOptions: {
          durable: true, // true인 경우, 브로커 서버가 재시작되어도 기존 Queue를 보존
        },
		prefetchCount: 1 // 1인 경우, 연결된 Consumer는 동시에 1개의 작업만 처리할 수 있음
      },
    });
  },
  inject: [ConfigService],
};

Pub/Sub

개요
- 이전 사례에서 보았던 일반적인 메시지 전달 패턴은 하나의 메시지는 하나의 Consumer로 전달됨
- 하나의 메시지를 여러 Consumer로 보내기 위해서는 Publisher / Subscriber 방식으로 구성 필요
Exchanges
- Pub/Sub 구조의 핵심 개념은 Producer가 Queue로 직접적으로 메시지를 전달하지 않는 것
- 중간에 exchange가 Producer로부터 메시지를 받고, 받은 메시지를 Queue로 푸시
- 기본 exchange 옵션은 fanout을 사용 (전체 Queue 대상으로 연결된 모든 Queue에 전달)
  - 활용 예: 스포츠 뉴스 사이트에서 클라이언트들에게 실시간 점수 업데이트
- 결과

Routing

개요
- Pub/Sub 구조에서 좀 더 복잡한 라우팅을 지원
Direct Exchange
- Routing Key를 기반으로 메시지를 Queue에 전달
- bindQueue의 세 번째 파라미터에 key를 지정하여 사용
결과
- 왼쪽부터 순서대로 gateway, microservice 1(info+warning), microservice 2(warning)이며, "warning" key에 양쪽 microservice로 메시지가 전달 됨을 확인할 수 있음

Topic

Routing과 유사하며, 지정된 Topic으로 메시지를 전달하게 됨
bindQueue의 세번째 파라미터로 topic을 전달하여 사용

RPC

개요
- Remote Procedure Call
- 클라이언트의 요청에 correlationId를 포함시켜, 요청에 대한 응답을 Queue를 통해 비동기적으로 처리
Callback Queue와 Correlation Id
- RPC 요청에 대한 Callback Queue를 생성하여 처리
결과
- gateway → microservice로 요청했을 때 microservice 측에서 5초간 delay후 응답을 처리하도록 코드를 작성함