RabbitMQ란?

Overview

오늘은 RabbitMQ에 대해서 공부해보려고 한다.

 

---

 

RabbitMQ란?

 

RabbitMQ는 AMQP를 따르는 오픈소스 메세지 브로커인데, 메세지를 많은 사용자에게 전달하거나, 요청에 대한 처리 시간이 길 때, 해당 요청을 다른 API에게 위임하고 빠른 응답을 할 때 많이 사용한다.

 

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

 

---

AMQP란?

AMQP란 Advanced Message Queueing Protocol의 줄임말로 MQ의 오픈소스에 기반한 표준 프로토콜을 의미한다. AMQP는 마지막 P(rotocol)에서 보는 것과 같이 프로토콜을 의미하기 때문에 이 것을 사용한 가장 유명한 소프트웨어는 RabbitMQ라 볼 수 있다.

 

---

 

RabbitMQ 개념

 

RabbitMQ에서 중요한 개념으로는 Producer, Consumer, Queue, Exchange, Binding이 있다.

 

 

먼저 Producer는 이름에서 알 수 있듯이, 메세지를 생성하고 발송하는 주체이다. 이 메세지는 Queue에 저장이 되는데, 주의할 점은 Producer는 Queue에 직접 접근하지 않고, 항상 Exchange를 통해 접근하게 된다.

 

다음으로는, Consumer가 있으며, 메세지를 수신하는 주체이다. Consumer는 Queue에 직접 접근하여 메세지를 가져온다.

 

Queue는 Producer들이 발송한 메세지들이 Consumer가 소비하기 전까지 보관되는 장소이다. Queue는 이름으로 구분되는데, 같은 이름과 같은 설정으로 Queue를 생성하면 에러 없이 기존 Queue에 연결되지만, 같은 이름과 다른 설정으로 Queue를 생성하려고 시도하면 에러가 발생한다.

 

Exchange는 Producer들에게서 전달받은 메세지들을 어떤 Queue들에게 발송할지를 결정하는 객체이다. Exchange는 네 가지 타입이 있으며, 일종의 라우터 개념이다.

 

Binding은 Exchange에게 메세지를 라우팅 할 규칙을 지정하는 행위이다. 특정 조건에 맞는 메세지를 특정 큐에 전송하도록 설정할 수 있는데, 이는 해당 Exchange 타입에 맞게 설정되어야 한다. (Exchange와 Queue는 m:n binding이 가능하다.

​

 

Exchange에는 네 가지 타입이 있는데, 각 타입별 특징은 아래와 같다.

타입	설명	특징
Direct	Routing key가 정확히 일치하는 Queue에 메세지 전송	Unicast
Topic	Routing key 패턴이 일치하는 Queue에 메세지 전송	Multicast
Headers	[key:value]로 이루어진 header 값을 기준으로 일치하는 Queue에 메세지 전송	Multicast
Fanout	해당 Exchange에 등록된 모든 Queue에 메세지 전송	Broadcast

​ * Headers 타입의 경우, 모든 header가 일치할 때, 하나라도 일치할 때 메세지 전송 등의 옵션을 줄 수 있습니다.

 

---

Direct Exchange

Direct Exchange는 라우팅 키를 이용하여 메세지를 라우팅 하는데, 하나의 큐에 여러 개의 라우팅 키를 지정할 수 있다.

위 그림은 error 메세지만 저장소에 기록하고, info와 warning을 포함한 모든 정보를 디스플레이에 출력할 때를 나타내는 모식도이다. (C1 : 저장소, C2 : 디스플레이)

Direct Exchange에 여러 큐에 같은 라우팅 키를 지정하여 Fanout처럼 동작하게 할 수도 있다.

​
* RabbitMQ에는 Default Exchange(Unnamed Exchange)가 있는데, 이 Exchange는 Direct타입이며, RabbitMQ에서 생성되는 모든 Queue가 자동으로 binding된다. 이 때, 각 Queue의 이름이 라우팅 키로 지정된다.

 

---

Topic Exchange

Topic Exchange는 라우팅 키 패턴을 이용하여 메세지를 라우팅한다.

라우팅 키가 example.orange.rabbit 인 경우 메세지가 Q1, Q2에 모두 전달되고,

라우팅 키가 example.orange.turtle 인 경우 메세지가 Q1에만 전달이 된다.

라우팅 키가 lazy.grape.rabbit인 경우엔 메세지가 Q2에 한 번만 전달이된다. (라우팅 패턴이 여러 개 일치하더라도 하나의 큐에는 메세지가 한 번만 전달됩니다.)

 

---

Headers Exchange

Headers Exchange는 Topic Exchange와 유사하지만 라우팅을 위해 header를 쓴다는 차이점이 있다.

​

---

​

​

Fanout Exchange

Fanout Exchange는 exchange에 등록된 모든 queue에 메세지를 전송한다.

​

---

Message Queue 및 Message 보존

RabbitMQ server가 종료(어떤 이유에서든지)후 재기동하면, 기본적으로 Queue는 모두 제거됩니다. 이를 막기 위해서는 Queue를 생성할 때, Durable 옵션에 true를 주고 생성해야 하며, Producer가 메세지를 발송할 때, PERSISTENT_TEXT_PLAIN 옵션을 주어야 메세지가 보존된다.

​

Queue 생성 예시

rabbitmqChannel.queueDeclare(rabbitmqQueueName, true, false, false, null); //QueueName 다음의 true가 durable option

​

Message 발송 예시

rabbitmqChannel.basicPublish(exchangeName, routingKey, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());

​

---

dispatching

만약 여러 소비자가 1개의 Queue를 바라보고 있다면 RabbitMQ에서는 Round-Robin을 사용해 메시지를 균등하게 분배한다. 즉, 중복 처리를 방지하기 위해 첫 번째 메시지는 소비자 1에게 전달하고 두 번째 메시지는 소비자 1이 아닌 소비자 2에게 전달한다. 소비자 뿐만 아니라 여러 생산자도 같은 Queue에 메시지를 전달 할 수 있습니다. 이러한 특성으로 메시지를 받아 처리하는 프로그램의 수평 확장이 가능하다. 

 

---

Fair dispatching

만약 소비자는 2개만 존재하고 홀수 번째의 메시지의 크기는 항상 크고 짝수 번째 메시지는 항상 작다면 Round-Robin 알고리즘을 사용해서 메시지 분배를 해도 공평하게 소비자에게 전달되지 않는다. 이러한 이유로 지연이 발생한 소비자에는 메시지를 전달하지 않도록 prefetch count라는 개념을 사용한다. prefetch count가 1로 설정되어 있다면 소비자로부터 act을 받지 못한 메시지가 1개라도 있으면 해당 소비자에 메시지를 전달하지 않는다. 즉, prefetch count는 소비자에게 동시에 전달되는 메시지의 양이다.

 

• 라운드 로빈 스케줄링(Round Robin Scheduling, RR)은 시분할 시스템을 위해 설계된 선점형 스케줄링의 하나로서, 프로세스들 사이에 우선순위를 두지 않고, 순서대로 시간단위(Time Quantum/Slice)로 CPU를 할당하는 방식의 CPU 스케줄링 알고리즘이다.

---

 

소비자 서버가 죽었을 경우

Queue는 소비자에게 메시지를 전달한 후 ACK을 받았을 때, 해당 메시지를 dequeue한다. 소비자가 ACK을 Queue에 전달하지 못하는 경우는 받은 메시지가 너무 커 아직 처리 중이거나 소비자 서버가 죽었을 때이다. RabbitMQ에서는 ACK을 받지 못한 메시지의 경우, 대기를 하고 있다가 전달한 소비자 서버의 상태를 확인한 후, Disconnected와 같은 신호를 받았을 경우 해당 소비자를 제외하고 다른 소비자에게 동일한 메시지를 전달한다.

 

---

Message Durability

만약 메시지를 Queue에 넣은 다음 소비자에게 전달하기 전에 RabbitMQ 서버가 죽는다면 Queue는 메모리에 데이터를 쓰는 형식이므로 모든 데이터가 소멸하게 된다. 이러한 문제를 해결하기 위해 영속성이란 개념을 가지고 있다. 

message durability는 메시지가 Queue에 저장될 때, disk의 파일에도 동시에 저장하는 방법이다. 해당 방법을 사용하면 서버가 죽었을 때, Queue의 데이터가 어느 정도 복구가 되지만 메시지가 disk의 파일에 쓰는 도중에 서버가 죽는 경우도 있어서 일부 데이터의 소실이 발생할 수 있다.

 

---

 

RabbitMQ Cluster 명령어

 

rabbitmq pod 접속 후 rabbitmq cluster 상태 확인

$ kubectl exec -ti -n openstack rabbitmq-rabbitmq-0 /bin/bash

$ rabbitmqctl cluster_status
Cluster status of node rabbit@rabbitmq-rabbitmq-0.rabbitmq.openstack.svc.cluster.local ...
[{nodes,
     [{disc,
          ['rabbit@rabbitmq-rabbitmq-0.rabbitmq.openstack.svc.cluster.local',
           'rabbit@rabbitmq-rabbitmq-1.rabbitmq.openstack.svc.cluster.local',
           'rabbit@rabbitmq-rabbitmq-2.rabbitmq.openstack.svc.cluster.local']}]},
 {running_nodes,
     ['rabbit@rabbitmq-rabbitmq-1.rabbitmq.openstack.svc.cluster.local',
      'rabbit@rabbitmq-rabbitmq-2.rabbitmq.openstack.svc.cluster.local',
      'rabbit@rabbitmq-rabbitmq-0.rabbitmq.openstack.svc.cluster.local']},
 {cluster_name,
     <<"rabbit@rabbitmq-rabbitmq-0.rabbitmq.openstack.svc.cluster.local">>},
 {partitions,[]},
 {alarms,
     [{'rabbit@rabbitmq-rabbitmq-1.rabbitmq.openstack.svc.cluster.local',
          []},
      {'rabbit@rabbitmq-rabbitmq-2.rabbitmq.openstack.svc.cluster.local',
          []},
      {'rabbit@rabbitmq-rabbitmq-0.rabbitmq.openstack.svc.cluster.local',
          []}]}]

 

---

 

rabbitmq 노드 중지, 재설정, 클러스터링, 노드 시작

$ kubectl exec -ti -n openstack rabbitmq-rabbitmq-2 /bin/bash

rabbitmq@rabbitmq-rabbitmq-2:/$ rabbitmqctl stop_app
# rabbitmqrabbitmq-rabbitmq-2 노드 중지
   
rabbitmq@rabbitmq-rabbitmq-2:/$ rabbitmqctl reset
# rabbitmq@rabbitmq-rabbitmq-2 노드 재설정 

rabbitmq@rabbitmq-rabbitmq-2:/$ rabbitmqctl join_cluster rabbitmq@rabbitmq-rabbitmq-1
# rabbitmq@rabbitmq-rabbitmq-2 노드를 rabbitmq@rabbitmq-rabbitmq-1  로 클러스터링합니다.

rabbitmq@rabbitmq-rabbitmq-2:/$ rabbitmqctl start_app
# rabbitmq@rabbitmq-rabbitmq-2 노드 시작

rabbitmq@rabbitmq-rabbitmq-2:/$ rabbitmqctl cluster_status
# 클러스터 상태 확인

 

---

 

접속 계정 생성 및 확인

 

1. 전체 계정 확인

$ sudo rabbitmqctl list_users

 

 

2. RabbitMQ 사용자 추가(<사용자> <비번>)

$ sudo rabbitmqctl add_user somaz somaz!1234

 

 

3. 사용자 태그 설정

$ sudo rabbitmqctl set_user_tags somaz administrator

 

 

4. 사용자 접속 퍼미션 설정

$ sudo rabbitmqctl list_permissions sudo rabbitmqctl set_permissions -p / adteck "." "." ".*"

 

 

5. 비밀번호 변경

$ rabbitmqctl change_password <사용자> <신규비번>

 

---

​

클러스터 노드 타입 변경

RabbitMQ 노드는 RAM과 disc, 2가지 타입을 갖는다. RAM 타입은 모든 메타 정보를 메모리에 저장한다. disc 보다 성능이 좋지만 해당 노드에 문제가 생겼을 때 데이터가 유실될 수 있다. 반대로 disc 타입은 데이터를 디스크에 저장한다.

RabbitMQ는 클러스터 내 모든 노드를 RAM 타입을 구성하는 것을 제한한다. 반드시 1개 이상의 disc 타입의 노드를 포함해야 한다.

위에서 이미 클러스터 구성을 마친 노드에 대해 RAM 타입으로 변환하는 과정이다.

 

# dev-rabbitmq001 기준
# 노드 중지
$ rabbitmqctl stop_app

# 노드 타입 변경
$ rabbitmqctl change_cluster_node_type ram

# 노드 시작
$ rabbitmqctl start_app

 

클러스터에 들어갈 때 노드 타입을 지정할 수도 있다.

# dev-rabbitmq002 가 dev-rabbit001 로 클러스터 구성을 요청
# 노드 중지
$ rabbitmqctl stop_app

# 클러스터 조인
$ rabbitmqctl join_cluster --ram rabbit@dev-rabbit001

# 노드 시작
$ rabbitmqctl start_app

 

---

미러링 정책 설정

 

• 클러스터 모든 노드의 Queue를 미러링하기

$ rabbitmqctl set_policy ha-all "^ha\\." '{"ha-mode":"all"}'

• 지정한 개수의 리플리카만 생성

$ rabbitmqctl set_policy ha-two "^two\\." '{"ha-mode":"exactly", "ha-params": 2, "ha-sync-mode": "automatic"}'

• 지정한 이름의 노드만 미러링

$ rabbitmqctl set_policy ha-nodes "^nodes\\." '{"ha-mode": "nodes", "ha-params": ["rabbit@nodeA", "rabbit@nodeB"]}'