분산환경에 RabbitMQ 적용하기

RabbitMQ란 AMQP(MQ 오픈소스 표준 프로토콜)의 구현체 중 하나이자 미들웨어 브로커다.

MSA 구조에서 서버와 서버 간 데이터 교환이 필요할 때 주로 사용하며 직접 통신이 아닌 MQ를 거쳐 통신함으로써 서비스 간 결합도를 낮추고 신뢰성 있는 통신을 하고자 할 때 주로 사용한다. 

kafka에 비해 비교적 적용이 쉬운 편이며 회사에서도 대규모의 이벤트 처리를 해야하는 구조가 아니어서 rabbitMQ를 사용중이다.

 

AMQP의 구성 요소

• Exchange : Publisher로부터 수신한 메시지를 큐에 분배하는 라우터 역할 (쉽게 말하면 배달원)
• Queue : 메시지를 메모리나 디스크에 저장했다가 컨슈머에게 메시지를 전달하는 역할
• Binding : Exchange와 Queue의 관계를 정의

 

구현체인 RabbitMQ도 당연히 위의 구성요소로 이루어져있다. Exchange 종류와 큐 바인딩 방식에 대해 알아보자.

 

Exchange Type

1) Direct Exchange (1:1)

- Exchange 에 바인딩된 큐 중에서 보내는 메시지의 routing key 와 매핑되어있는 Queue로 메시지를 전달한다.

- 특정 메시지가 특정 큐에만 들어가길 원할 때 사용하면 된다. 

 

2) Fanout Exchange (1:N)

- 라우팅 키를 무시하고 Exchange에 바인딩 된 모든 Queue 에 메시지를 전달한다.

- 현재는 해당 방식을 사용해 특정 이벤트를 구독한 서버라면 바인딩 큐를 생성하고 서로 다른 서버들이 같은 메시지를 소비하고 있다.

 

3) Topic Exchange (1:N)

- Exchange에 바인딩 된 Queue 중에서 메시지의 routing key pattern에 맞는 Queue에게 모두 메시지를 전달한다.

- Direct Exchange 와 비슷해보이지만 정확한 routing key가 아니어도 패턴에 맞는 routing key를 가진 Queue라면 모두 전달하기 때문에 좀 더 유연하다.

 

4) Headers Exchange 

- key-value로 정의된 헤더를 기반으로 메세지를 매칭되는 queue에 전달하는 방법이다.

 

 

Model 

rabbitMQ의 사용가능한 모델 종류이다. 

 

1. work Queues

- 제일 간단한 방식.

- 하나의 큐를 만들고 메시지를 보내는 publisher와 그 메시지를 소비하는 consumer를 전부 가져다 붙인다.

- exchange와 라우팅 키는 신경 쓸 필요가 없다.

- round-robin 방식으로 메세지들을 알아서 분산 처리한다. (즉 하나의 메시지는 하나의 소비자만 받을 수 있다.)

 

2. pub/sub

- 주로 fanout 방식을 사용하며, publisher가 어떤 메시지를 exchange로 보내면 binding 된 모든 큐로 메시지를 보내주는 방식이다.

- publisher는 그저 이벤트를 발행하며 목적지를 알 필요가 없다.

- 하나의 이벤트를 여러개의 서비스가 함께 구독할 때 사용하면 유용하다.

 

 

3. Routing

- direct exchange type을 사용하는 방식이며 메시지를 선택적으로 받고자 할 때 사용한다. 

- routingKey에 따라 특정 메시지는 특정 큐에만 들어갔으면 할 때 사용한다. 

 

 

4. Topics

- topic exchange type을 사용하는 방식이며 routingKey 패턴에 따라 특정 큐에 들어간다.

 

 

그 외에 다른 모델도 아래 공식 문서에서 확인 가능하다.

https://www.rabbitmq.com/getstarted.html

 

 

신뢰성(Reliability)과 내구성(Durability)

rabbitMQ에서 메시지를 유실을 막기 위한 방법에는 뭐가 있을까?

 

1) Subscriber가 메시지를 정상적으로 받았지만 처리 중 비정상적으로 Subscriber 서버가 종료된 경우

- rabbitMQ는 자체적으로 ACK, NACK 처리를 받기 전까지는 큐에서 메시지를 지우지 않는다.

- 즉, 서버가 재시작되면서 다시 큐로부터 메시지를 소비해 재처리하게 된다. (멱등성있는 소비 로직을 작성하는 것이 중요한 이유다)

 

2) RabbitMQ 브로커 서버 자체가 컨슈머의 메시지 처리를 기다리던 중 비정상 종료된 경우

- Queue 유지하기 : 바인딩 된 큐를 계속해서 기억할 수 있도록 큐 선언시 durable = true 설정을 통해 유지할 수 있다. (기본적으로 브로커가 종료되면 재시작시 큐가 사라져있을 것이다)

- Queue 안의 메시지 유지하기 : 발행된 메시지도 지속성을 유지하고 싶다면 publisher가 publish 할 때의 persistent 속성을 부여해야 한다. 마찬가지로 ACK를 받지 못한 경우이기 때문에 큐에서 메시지를 유지할 수 있다. persistent 속성은 deliveryMode를 2 (persist to disk)로 설정해주면 된다.

 

자세한 설정은 아래 링크 참고

https://teragoon.wordpress.com/2012/01/26/message-durability

 

스케일 아웃

어떤 이벤트를 소비하는 하나의 마이크로서비스에 대해 스케일아웃을 하게 되어 서버가 여러대가 된다면 어떻게 이벤트를 중복없이 처리할 수 있을까?

특정 이벤트에 바인딩된 큐는 서버별로 하나만 생겨야 하며 레플리카 중 하나의 서버만 해당 이벤트를 가져가 처리해야 한다.

서버가 늘어날 때마다 큐가 추가되면 같은 이벤트에 대해 각 서비스들이 중복으로 이벤트를 처리하게 될 수 있기 때문이다.

이럴 경우를 대비해 각 마이크로서비스당 큐의 이름은 똑같이 만들어지도록 구성해야 한다.

만약 큐를 생성했을 때 동일한 이름의 큐가 존재할 경우 해당 큐를 공유하게 된다. 즉 하나의 서비스는 스케일 아웃이 되었더라도 하나의 큐만 사용해 이벤트를 처리할 수 있다. 이렇게 되어야 라운드 로빈방식으로 돌아가며 서비스가 이벤트를 중복 없이 처리할 수 있다. 

 

 

순서 보장과 동시 처리 (prefetch)

메시지 큐의 특성 상 선입선출 방식이다. 큐에 들어온 순서대로 컨슈머가 소비한다. 

즉, 바인딩 큐에 대한 컨슈머가 하나라면, 메시지 순서를 보장한다.

컨슈머가 여럿이라면 메시지 순서가 보장되지 않는다. 라운드 로빈 방식으로 하나씩 컨슈머에게 번갈아 가며 보낸다.

 

rabbitMQ는 기본적으로 큐로부터 메시지를 동시에 전부 가져와 처리한다. 즉, 큐에 이벤트가 많이 쌓여있을 수록 메모리 사용량이 증가해 서비스 성능이 저하될 수 있다. 따라서 이벤트를 대용량으로 받아야 하는 경우 prefetch count 설정을 통해 동시에 처리할 메시지 개수를 정해줄 수 있다. 단, prefetch count를 사용할 경우 반드시 ACK, NACK 처리를 확실히 해주어야 한다. 

혹시라도 prefetch 를 1로 설정한 후 가져온 이벤트에 대해 ACK,NACK 처리를 해주지 않을 경우 큐에 남아있는 메시지들은 계속해서 ready 상태로 대기에 빠지게 되며 소비중인 이벤트는 unacked 상태로 큐에 계속 남아있게 된다. 즉 1개는 unacked, 나머지는 ready 상태로 메시지큐는 어떤 작업 진행도 할 수 없다.

 

+) prefetch 값은 어떻게 설정하는게 좋을까?

- 처리 시간이 매우 느릴 경우, prefetch 값을 작게 가져가는 것이 좋다. (ex: 1) 미리 컨슈머가 데이터를 가지고 있을 이유가 없다.

- 해당 이벤트를 처리하는 컨슈머의 개수를 고려한다. 보통 보내는 메시지 양/컨슈머 개수 만큼 설정하여 유휴 상태를 가지는 consumer가 없도록 한다.

- 클라이언트 처리 시간 대비 너무 값이 적을 경우에도 conumer의 유휴 상태가 길어지고 너무 높아도 하나의 consumer가 전부 가져가고 나머지 consumer가 유휴 상태로 유지될 수 있어 적정 값 설정이 중요하다.

 

 

트러블 슈팅

rabbitMQ 타임아웃 발생

- 데이터를 디스크에 저장 도중 강제 종료되어 데이터가 꼬인 경우 아래와 같은 에러 발생

- 클라우드 엔지니어 분이 도와주셨다!

로그:  command time out:  '/usr/bin/df -kP ..' 

해결 방법 : 

1. rebuilding 하는 도중 타임아웃이 나지 않도록 rabbitMQ 최신 버전 업데이트 (고쳐졌다고 한다.)

2. rebuilding 되는 동안 파드가 죽지 않도록 livenessprobe 꺼두기

3. rebuilding 완료될 때까지 기다린다. (강제 종료하지 않고!)

+) 다른 대안: 데이터를 초기화하면 해결되긴 한다.