RabbitMQ IN DEPTH

RabbitMQ IN DEPTH 책을 공부하며 작성한 글.

 

이 책은 RabbitMQ 3.6.3 을 기준으로 저술되었다. 

 

예제 코드

이 책에 나오는 모든 예제 코드는 매닝 웹사이트 (https://www.manning.com/books/rabbitmq-in-depth) 와 깃허브 저장소 (https://github.com/gmr/RabbitMQ-in-Depth) 에서 무료로 다운로드할 수 있다. 

에이콘출판사 도서정보 페이지 http://www.acornpub.co.kr/book/rabbitmq-depth 에서도 예제 코드를 다운로드할 수 있다. 

 

1부 RabbitMQ 와 애플리케이션 아키텍처

1. RabbitMQ 살펴보기

RabbitMQ 의 기능과 장점

• 오픈소스
• 플랫폼과 업체 중립성
• 경량성
• 다양한 클라이언트 라이브러리 
• 유연한 성능과 안정성 절충 제어 
• 대기 시간이 긴 네트워크 환경을 위한 플로그인 
• 서드파티 플러그인 
• 보안 계층 

1.1.1 RabbitMQ 와 얼랭 

RabbitMQ 는 얼랭으로 구현됨

 

1.1.2 RabbitMQ와 AMQP

2007년 처음 출시했을 때 AMQP 스펙을 구현한 최초의 메시지 브로커 중 하나였다. AMQP 자체가 RabbitMQ 에 영향을 많이 줬다. 

 

AMQP에 대한 자세한 내용은 https://www.amqp.org/ 에서 확인할 수 있다. 

 

메시지 프로토콜과 메시지 브로커는 다양하므로, 애플리케이션에 미치는 영향을 고려해 메시지 프로토콜과 메시지 브로커를 선택해야 한다. RabbitMQ 는 AMQP 를 기반으로 구현됐지만 MQTT, STOMP, XMPP 등 다양한 프로토콜도 제공한다. 멀티 프로토콜 애플리케이션 아키텍처에는 좋은 선택이다. 

 

1.2 RabbitMQ 를 사용하는 곳들 

• Reddit
• NASA 의 Nebula
• 아구라 게임즈
• Ocean Observations Initiative
• 래포티브
• 메르카도리브레
• 구글의 AdMob
• Andhaar

1.3 느슨하게 결합된 아키텍처의 장점

로그인 서비스에서 마지막으로 로그인한 시간을 비동기로 작업하므로, 애플리케이션은 즉시 인증된 회원 페이지로 이동한다.

 

1.3.1 애플리케이션의 의존성 제거 

애플리케이션 아키텍처는 더 이상 데이터베이스 쓰기 성능에 영향을 받지 않는다. 핵심 애플리케이션 코드를 수정하지 않고도 데이터를 처리하는 새로운 애플리케이션을 쉽게 추가할 수 있다. 

 

1.3.4 데이터와 이벤트 복제 

Federation 플러그인은 WAN 허용 오차 및 네트워크 단절을 고려해서 원격 RabbitMQ 인스턴스에 메시지를 전달한다. 

 

1.3.5 다중 마스터 Federation 

 

노트

메시지 지향 미들웨어를 사용하면 어느 정도의 운영상 복잡성이 발생한다. 메시지 브로커는 구조상 중심점으로 애플리케이션 설계상 새로운 단일 장애 지점 (Single point of failure) 이 된다. 

 

1.3.6 AMQ 모델 

HTTPj, SMTP 와 같은 프로토콜과 달리 AMQP 스펙은 네트워크 프로토콜의 정의뿐 아니라 서버 측 서비스와 동작 방식도 정의하는데, AMQ (Advanced Message Queuing) 모델을 살펴보면 확인할 수 있다. AMQ 모델은 메시지 라우팅 동작을 정의하는 메시지 브로커의 세 가지 추상 컴포넌트를 다음과 같이 논리적으로 정의한다. 

• 익스체인지: 메시지 브로커에서 큐에 메시지를 전달하는 컴포넌트 
• 큐: 메시지를 저장하는 디스크상이나 메모리상의 자료 구조 
• 바인딩: 익스체인지에 전달된 메시지가 어떤 큐에 저장돼야 하는지 정의하느 컴포넌트 

익스체인지

익스체인지 Exchange 는 RabbitMQ 에서 메시지를 적절한 목적지로 전달하기 위해 필요한 첫 번째 입력 값으로 AMQ 모델이 정의하는 세 컴포넌트 중 하나다. 여러 유형의 익스체인지가 있다. 특히 플러그인을 사용해서 직접 커스텀 익스체인지도 정의할 수 있다. 

 

큐

큐는 수신한 메시지를 저장하는 역할

 

바인딩

바인딩을 이용해 큐와 익스체인지의 관계를 정의한다. 

바인딩 키는 익스체인지가 어떤 큐에 메시지를 전달해야 하는지를 의미한다. 

익스체인지에 메시지를 발행할 때 애플리케이션은 라우팅 키 (routing-key) 속성을 사용한다. 라우팅 키는 때로는 큐의 이름이거나 의미적으로 메시지를 설명하는 문자열이다. 익스체인지는 라우팅 키를 바인딩 키에 맞춰서 평가한다. 즉, 바인딩 키는 큐를 익스체인지에 연결하고 라우팅 키를 평가하는 기준이다. 

 

RabbitMQ 에서느 AMQP 스펙을 확장해 특정 익스체인지를 다른 익스체인지에 연결할 수 있는데, 이는 메시지를 라우팅하는 데 상당한 유연성을 제공한다. 

 

2. AMQP와 RabbitMQ 코드 작성하기

2.1 RPC 전송으로서의 AMQP

RabbitMQ 는 거의 모든 부분에서 RPC (Remote Procedure Call) 패턴으로 엄격하게 통신한다. RPC 는 한 컴퓨터에서 다른 컴퓨터의 프로그램이나 프로그램의 메서드를 원격에서 실행할 수 있게 해주는 컴퓨터 간의 통신 유형 중 하나다. 원격 API 와 통신하는 웹 프로그램이 있다면, 이는 일반적인 RPC 패턴을 사용했다고 볼 수 있다. 

AMQP 스펙은 서버와 클라이언트 모두 명령을 실행할 수 있다. 

 

2.1.1 대화 시작하기 

AMQP 로 통신을 시작할 때, 인사말은 프로토콜 헤더 (Protocol header) 에 해당되는데 클라이언트가 서버로 전송한다. RabbitMQ 는 Connection.start 명령으로 인사말에 응답해 명령/응답 흐름을 시작하고 클라이언트는 Connection.StartOk 응답 프레임으로 RPC 요청에 응답한다. 

 

2.1.2 올바른 채널로 튜닝 

RabbitMQ 를 이용해서 클라이언트 애플리케이션을 구현할 때는 복잡하게 너무 많은 채널을 사용하지 않는 것이 좋다. 

 

2.2 AMQP 의 RPC 프레임 구조 

Connection.Start (클래스.메소드) 

 

2.2.1 AMQP 프레임 컴포넌트 

저수준 AMQP 프레임은 다섯 개의 별개 구성 요소로 구성된다. 

1. 프레임 유형 
2. 채널 번호
3. 프레임 크기 (바이트) 
4. 프레임 페이로드 (Payload) 
5. 끝 바이트 표식 (ASCII 값 206) 

2.2.4 메소드 프레임 해부하기 

노트

메시지를 발행할 때 mandatory 플래그를 사용하는 경우 애플리케이션은 RabbitMQ 에서 응답한 Basic.Return 명령을 수신해야 한다. RabbitMQ 가 mandatory 플래그에 설정한 요구 사항을 충족하지 못하면 Basic.Return 명령을 동일한 채널의 클라이언트에 전송한다. 자세한건 4장에서 다룬다. 

 

2.3 프로토콜 사용하기 

메시지를 큐에 발행하기 전에 몇 가지 설정 단계를 거쳐야 하는데, 최소한 익스체인지와 큐를 설정한 후 둘을 연결해야 한다. 

 

2.3.1 익스체인지 선언하기 

AMQ 모델에서 익스체인지는 큐와 같이 '1급 시민' 으로 AMQP 스펙에 해당 클래스가 존재한다. 

Exchange.Declare 명령을 전송하면 RabbitMQ 는 익스체인지를 생성한 후 Exchange.DeclarOk 메소드 프레임을 응답으로 전송한다. Declare 명령이 실패하면 RabbitMQ sms Exchange.Declare 가 실패하고 채널이 닫힌 이유를 나타내는 숫자 응답 코드와 텍스트 값을 Channel.Close 명령에 포함시켜 전송하고 Exchange.Declare 명령이 전송된 채널을 닫는다. 

 

2.3.2 큐 선언하기 

Queue.Declare 명령이 실패하면 채널이 닫힌다. 

큐를 선언할 때 동일한 Queue.Declare 명령을 두 번 이상 전송해도 문제는 발생하지 않는다. RabbitMQ 는 중복된 큐 선언을 감지해 큐에 대기 중인 메시지의 수와 구독 중인 구독자의 수와 같이 큐에 대한 유용한 상태를 반환한다. 

 

정상적으로 에러 처리하기

이미 생성한 큐와 같은 이름이지만, 속성이 다른 큐를 선언하려고 시도하면 RabbitMQ 는 RPC 요청을 발행한 채널을 닫는다. 예를 들어 가상 호스트 (virtual host) 의 설정에 대해 접근 권한이 없는 사용자가 Queue.Declare 명령을 실행하면 403 에러가 반환되고 채널은 닫힌다. 

 

클라이언트 애플리케이션이 에러를 정상적으로 처리하려면 RabbitMQ 로부터 Channel.Close 명령을 전달받아 적절하게 응답해야 한다. 특정 클라이언트 라이브러리는 에러 응답을 애플리케이션이 처리할 수 있는 예외로 변환해 처리하며, 다른 유형의 라이브러리는 사용자가 메소드를 등록할 때 콜백을 추가하도록 하고 Channel.Close 명령을 보낼 때 콜백을 호출하는 식으로 처리하기도 한다. 

 

클라이언트 애플리케이션이 서버에서 전송하는 이벤트를 수신하지 않거나 적절하게 처리하지 않으면 메시지가 손실될 수 있다. 존재하지 않거나 이미 닫힌 채널에 메시지를 발행하는 경우 RabbitMQ 는 연결을 종료한다. 메시지를 소비하는 애플리케이션이 RabbitMQ 가 채널을 닫은 사실을 모르는 경우 RabbitMQ 가 메시지를 더는 전송하지 않지만, 클라이언트는 빈 큐를 구독하고 있다고 간주하는 문제가 발생한다. 

 

2.3.3 큐와 익스체인지 연결하기 

Queue.Declare 와 유사하게 큐를 익스체인지에 연결하는 명령인 Queue.Bind 는 한번에 하나의 큐만 지정한다. 성공적으로 처리된 경우 클라이언트 애플리케이션에 Queue.BindOk 메소드 프레임을 전송한다. 

 

위는 모두 동기이며, 비동기 명령들도 있다. 

 

2.3.4 RabbitMQ 에 메시지 발행하기 

Basic.Publish 메소드 프레임에는 익스체인지의 이름과 라우팅 키가 들어있는데, 이를 RabbitMQ 는 익스체인지의 이름을 저장한 데이터베이스와 비교한다. 

 

팁

RabbitMQ 에 존재하지 않는 익스체인지에 메시지를 발행하는 경우, 기본적으로 메시지는 자동으로 버려진다. 메시지가 제대로 발행됐는지 확인하려면, 메시지 발행 시에 mandatory 플래그를 true 로 설정하거나 발행자 확인 (publisher confirmations) 을 사용해야 하는데, 이 옵션에 대해서는 4장에서 자세히 알아본다. 이 중 하나를 사용하면 애플리케이션의 메시지 발행 속도가 저하될 수 있으니 주의해야 한다. 

 

큐를 구독하는 소비자가 없어서 메시지를 소비하지 않는다면 메시지는 큐에 계속 저장되고 메시지를 더 추가할수록 큐의 크기도 커진다. RabbitMQ 는 메시지의 Basic.Properties 에 지정된 배달 모드 (delivery mode) 에 따라 메시지를 메모리에 보관하거나 디스크에 기록한다. 배달 모드는 매우 중요하므로 다음 4장에 걸쳐 자세히 알아본다. 

 

2.3.5 RabbitMQ 에서 메시지 소비하기 

소비자 애플리케이션은 Basic.Consume 명령을 실행해서 RabbitMQ 의 큐를 구독한다. 

Basic.Cunsume 이 발급되면 특정 상황이 발생하기 전까지 활성 상태를 유지한다. 소비자가 메시지 수신을 중지하려면 Basic.Cancel 명령을 발행해야 한다. 

소비자는 Basic.CancelOk 응답 프레임을 받기 전에 RabbitMQ 가 미리 할당된 메시지 수만큼 메시지를 받을 수 있다. 

메시지를 소비할 때 RabbitMQ 에는 소비자의 수신 방식을 알 수 있는 몇 가지 설정이 있는데, 그 중 하나는 Basic.Consume 명령의 no_ack 인수다. no_ack 를 true 로 설정하면 RabbitMQ 는 소비자가 Basic.Cancel 명령을 보내거나 연결을 끊을 때 까지 계속 메시지를 보낸다. no_ack 플래그를 false 로 설정하면 소비자는 Basic.Ack RPC 요청을 전송해 수신한 각 메시지를 확인해야 한다. 

 

2.4 파이썬으로 메시지 발행자 작성하기 

일단 도커로 RabbitMQ 를 실행해 보자. 

docker run -d -p 15672:15672 -p 5672:5672 --name rabbitmq rabbitmq:3.8-management
docker exec rabbitmq rabbitmq-plugins enable rabbitmq_management

 

그리고 

http://localhost:15672

로 접속을 해서 ID PASSWORD 를 guest/guest 로 접속해보자. 

 

exchange 를 선언해보자. 

python example/Examples/2.4\ Publisher\ Example.py
# exchange = rabbitpy.Exchange(channel, 'chapter2-example')

 

chapter2-example exchange 가 생성되었다.

 

 

example queue 도 생성되었다.

 

메시지 발행방법 요약

1. connection.channel 로 채널 생성 
2. exchange 생성 후 declare 
3. queue declare 
4. queue 에 exchange 와 bind 
5. 메시지에 채널 명시해 생성 후, exchange 에 routing key 와 함께 publish 

팁

운영환경에서 발행자 애플리케이션을 작성하는 경우 JSON 또는 XML 과 같은 데이터로 직렬화하면 소비자가 메시지를 쉽게 디코딩할 수 있으므로 문제가 발생하는 경우 더 쉽게 원인을 찾을 수 있다. 

 

큐에 10개의 메시지가 들어갔다.

 

Get messages 를 하면 

10개의 메시지를 볼 수 있다.

 

 

3. 메시지 속성 심층 탐사

메시지를 설명하기 위한 일관된 방법을 찾기 위해 RabbitMQ 에 발행된 모든 메시지와 함께 전달되는 데이터 구조인 AMQP 스펙의 Basic.Properties 를 살펴봤다. Basic.Properties 를 활용하면 메시지가 자동으로 제거되거나, 처리하기 전에 메시지의 출처와 유형을 검증할 수 있는 더욱 지능적인 소비자 애플리케이션의 작성이 가능했다. 3장에서는 메시지의 각 속성과 다양한 용도로 사용할 수 있는 Basic.Properties 에 대해 자세히 살펴본다. 

 

3.1 메시지 속성 적절히 사용하기 

발행한 메시지는 세 가지 유형의 프레임인 Basic.Publish 메소드 프레임, 콘텐츠 헤더 프레임, 바디 프레임으로 구성된다. 

콘텐츠 헤더 프레임에 있는 메시지 속성은 Basic.Properties 데이터 구조로 사전에 정의한 값이 있는 집합이다. delivery-mode 와 같은 일부 속성은 AMQP 스펙에서 정의한 의미를 갖지만, type 과 같이 정확한 스펙이 없는 속성들도 있다. 

delivery-mode 속성은 메시지가 큐에 있을 때 메시지를 메모리에 보관할지, 디스크에 먼저 저장해야 할지 RabbitMQ 에 알리는 데 사용된다. 

 

팁

RabbitMQ 에서 MQTT 와 같은 프로토콜을 사용하는 경우, AMQP 에 특정된 속성은 사용할 수 없게 되므로 특정 속성이 손실되지 않도록 유의해야 한다. 

 

각 기본 속성을 살펴보자.

• content-type: 메시지 본문 해석하는 방법 
• content-encoding: 메시지 본문이 어떤 방법으로 압축되거나 인코딩됐는지
• message-id, correlation-id: 메시지와 메시지 응답을 고유하게 식별해 메시지 추적 
• timestamp: 메시지 생성 시점에 대한 표준시간 전달 
• expriation: 메시지 만료 전달
• delivery-mode: 큐에 메시지 추가할 때, 디스크 또는 메모리에 저장할지를 전달 
• app-id, user-id: 문제가 발생한 애플리케이션을 추적 
• type: 발행자와 소비자 사이에 계약 (contract) 정의 
• reply-to: 패턴을 값으로 전달해 응답 메시지를 라우팅할 때 사용 
• headers: RabbitMQ 에 메시지를 라우팅할 때, 사용자 정의 형식의 속성을 정의하는 데 사용 

3.2 content-type 으로 명시적 메시지 계약 작성하기 

HTTP 의 content-type 과 같다. application/json 같은 것 

파이썬에서 라이브러리는 content-type 헤더에서 메시지가 어떤 유형으로 직렬화됐는지 감지하고, 이를 사용해 메시지 본문을 자동으로 디코딩해 dict, list 또는 다른 원시 데이터 유형으로 변환할 수 있다. 이를 통해 소비자 애플리케이션의 코드 복잡성을 현저하게 줄일 수 있다. 

 

3.3 gzip, content-encoding 으로 메시지 크기 줄이기 

AMQP 를 이용해서 전달한 메시지는 기본적으로 압축되지 않는다. 이는 XML 과 같이 지나치게 자세한 마크업 문법이나 큰 메시지에는 JSON, YAML 과 같이 마크업을 사용하지 않는 경우에도 문제가 될 수 있다. 서버에서 웹 페이지를 gzip 으로 압축하고 브라우저가 렌더링하기 전에 압축을 푸는 것과 마찬가지로 발행자는 메시지를 발행하기 전에 압축하고 소비자로부터 메시지를 전달받아 압축을 풀 수 있다. 

content-encoding 속성으로 메시지 본문이 base64 혹은 gzip 과 같은 특수한 형식으로 인코딩됐는지 알 수 있다. 

 

운영 환경에서는 발행자와 소비자의 메시지 계약을 운영 중에 변경하지 않는 것이 바람직하며 기존 코드에 대한 잠재적 영향을 최소화하는 것이 좋다. 그러나 메시지 크기가 애플리케이션의 전체 성능이나 안정성에 영향을 미쳐서 본문 인코딩의 변경이 불가피한 경우, content-encoding 헤더를 사용하면 소비자가 메시지의 형식을 사전에 확인할 수 있으므로 메시지 본문을 적절하게 디코딩할 수 있다. 

 

노트

AMQP 클라이언트는 자동으로 content-encoding 값을 UTF-8 로 설정하지만, 이는 잘못된 동작이다. AMQP 스펙에는 content-encoding 이 MIME 콘텐츠 인코딩을 저장하기 위한 것이라고 명시돼 있다. 

 

MIME 이메일 마크업은 병렬 처리가 가능하도록 content-encoding 필드를 사용해 이메일의 각 파트에 대한 인코딩을 표현한다. 이메일에서 가장 일반적인 인코딩 유형은 Base64 와 QP (Quoted-Printable) 인코딩이다. Base64 인코딩은 메시지에서 전송할 바이너리 데이터가 텍스트 전용인 SMTP 프로토콜의 범위를 넘지 않도록 사용된다. 예를 들어 이미지가 포함된 HTML 의 이메일 본문을 만드는 경우 포함된 이미지는 Base64 로 인코딩된다. 

하지만 SMTP 와 달리 AMQP 는 바이너리 프로토콜이다. 

 

클라이언트 라이브러리 활용하기 

클라이언트 라이브러리를 사용해 소비자 코드를 작성하는 경우, content-encoding 속성을 사용해 수신 시 메시지를 자동으로 디코딩할 수 있다. 보통 라이브러리가 메시지의 전처리, 디코딩, 압축 해제를 처리하므로 직접 작성해야 하는 소비자 애플리케이션의 로직과 코드는 단순하다. 따라서 개발자는 메시지 본문을 처리하는 작업에 집중할 수 있다. 

 

애플리케이션의 운영 중에 메시지 본문을 bzip2 로 압축하는 것이 더 적합하다고 판단돼 변경하더라도 content-encoding 속성을 검사하도록 소비자 애플리케이션을 구현하면 디코딩할 수 없는 메시지를 거부할 수 있다. zlib 압축만 해제할 수 있는 소비자 애플리케이션은 bzip2 로 압축된 메시지를 거부해 bzip2 압축을 풀 수 있는 다른 소비자 애플리케이션이 처리할 수 있도록 큐에 메시지를 남겨둔다. 

 

3.4 message-id 와 correlation-id 를 이용한 메시지 참조

AMQP 스펙에서 message-id 와 correlation-id 는 '애플리케이션 용도' 로 지정됐으며, 공식적으로 정의된 동작은 없다. 두 필드는 최대 255 바이트의 UTF-8 로 인코딩된 값을 가지며 Basic.Properties 데이터 구조에 포함된 압축되지 않은 값으로 저장된다. 

• message-id: 메시지를 고유하게 식별
• correlation-id: 메시지가 다른 메시지에 대한 응답임을 나타내는 데 사용, 이 경우 이전 메시지의 meesage-id 를 값으로 포함 

3.4.1 message-id

판매 주문이나 지원 요청과 같은 경우 메시지를 쉽게 파악하는 데 도움이 된다. 

 

3.4.2 correlation-id 

다른 메시지에 대한 응답임을 표시. 또 다른 사용 예는 트랜잭션 ID 나 메시지가 참조하는 다른 데이터를 전달하는 데 이 속성을 사용하는 것이다. 

 

3.5 timestamp

timestamp 속성을 사용해 메시지 생성 시점을 기록하면 메시지를 발행할 때 성능을 추적할 수 있다. 

프로세스가 시행해야하는 서비스 수준 계약 (SLA, Service Level Agreement) 이 있다면, 소비자 애플리케이션에서 메시지의 timestamp 속성을 평가해 메시지를 처리할 지 여부를 결정하거나 메시지의 수명이 지정한 값을 초과한 경우 모니터링 애플리케이션에 경고 메시지를 발행해서 누군게에게 알릴 수 있다. 

 

timestamp 는 유닉스 시간 (Unix epoch) 또는 1970년 1월 1일 자정 이래로 경과된 초를 나타내는 정수로 전송된다. 불행히도 timestamp 는 시간대 (time zone) 정보가 없으므로 UTC 혹은 다른 일관된 시간대를 약속해 사용하는 것이 좋다. 

 

3.6 자동으로 메시지 만료하기 

expiration 속성은 RabbitMQ 에서 소비하지 않은 메시지를 버려야 할 때를 파악하는데 사용한다. expiration 속성은 AMQP 스펙 0-8 과 0-9-1 버전에 모두 존재하지만, RabbitMQ 3.0 버전 이전에는 지원되지 않았다. '구현할 때 사용할 수 있지만 공식적인 동작은 없음' 으로 정의됐다. timestamp 와 동일하게 유닉스 시간을 값으로 갖지만, 타입은 255자의 짧은 문자열이라는 것이다. 

RabbitMQ 에서 expiration 속성을 사용해 메시지를 자동으로 만료 처리하려면 유닉스 시간 또는 정수 기반 timestamp 를 값으로 가져야 하지만, 타입은 문자열로 저장돼야 한다. "2002-02-20T00:00:00-00" 과 같이 ISO-8601 형식의 타임스탬프를 저장하는 대신 문자열 값인 "1329696000" 과 동일한 형식으로 값을 설정해야 한다. 

 

expiration 속성을 사용하는 메시지가 서버에 도착한 후 시간이 만료된 경우 메시지는 큐로 삽입되지 앉고 삭제된다. 

RabbitMQ 에는 특정 상황에서만 메시지가 만료되는 다른 기능이 있다. 큐를 선언할 때 큐의 정의와 함께 x-message-ttl 속성을 인자로 전달해서 메시지를 만료할 수 있는데, 유닉스 시간이지만 밀리세컨드 정밀도 (유닉스 시간 X 1000) 의 정수로 값을 설정한다. 큐의 x-message-ttl 속성은 지정된 시간이 경과되면 메시지를 자동으로 삭제한다. 

 

3.7 배달 모드를 이용해 안정성과 속도 조절하기 

메시지를 디스크에 저장하면 RabbitMQ 서버를 정지하고 다시 시작하더라도 메시지가 소비될 때까지 큐에 남아있게 된다. delivery-mode 속성은 메시지를 저장하지 않을 경우 1, 메시지를 저장하는 경우 2, 이렇게 두 가지 값으로 지정된다. 

 

노트

RabbitMQ 의 다양한 용어와 설정에 대해 처음 접한다면, 메시지 지속성 (persistence) 이 큐의 내구성 (durable) 속성과 혼동될 수 있다. 큐의 durable 속성은 RabbitMQ 서버나 클러스터를 다시 시작한 후에도 큐 정의가 유지돼야 하는지를 나타내는 반면, delevery-mode 는 메시지를 유지할지 여부를 나타낸다. 하나의 큐에는 디스크에 저장되는 지속성 메시지와 메모리에만 보관되는 비지속성 메시지가 동시에 포함될 수 있다. 

 

웹 애플리케이션의 로그인 이벤트의 경우 delivery-mode 속성을 선택하기가 다른 케이스보다 쉽다. 로그인 이벤트가 없어진다고 해서 비즈니스가 위험에 빠지지는 않으므로 이벤트를 메모리에만 보관하는 것도 합리적인 선택이 된다. 이 경우 delivery-mode 를 1로 설정한다. 그러나 RabbitMQ 를 사용해 금융 거래 데이터를 발행하고 애플리케이션 아키텍처가 메시지 처리량보다는 정확한 전달에 초첨을 맞춘다면 delivery-mode 를 2로 지정해 지속성을 활성화한다. 

배달모드는 성능에 중요한 영향을 미친다.

 

3.8 app-id 및 user-id 를 사용해 메시지의 출처 확인하기 

app-id 와 user-id 는 소비자 애플리케이션에서 메시지를 처리하기 전에 유효성을 검증하는 용도로 활용된다. 

app-id 속성은 발행자 애플리케이션에서 자유롭게 사용할 수 있는 문자열 값이다. 

RabbitMQ 는 메시지를 발행하는 RabbitMQ 사용자에 대해 user-id 속성을 이용해 유효성을 검사한다. 

**메시지 출처를 확인하는 데 사용한다. 

 

3.8.1 app-id 속성

app-id 속성은 최대 255자의 짧은 UTF-8 문자열이다. 애플리케이션이 API 중심으로 디자인돼 버전 관리가 필요한 경우 app-id 를 사용해 생성된 메시지와 함께 특정 API 와 버전을 달리할 수 있다. 발행자와 소비자 간에 계약을 맺는 방법 중 하나로 사용한다면, 메시지를 처리하기 전에 app-id 를 검사해 알 수 없거나 지원하지 않는 출처의 메시지의 경우 애플리케이션에서 메시지를 삭제할 수 있다. 

app-id 의 다른 사용법은 통계 데이터로 수집하는 것이다. 예를 들어 메시지를 사용해 로그인 이벤트를 전달하는 경우, app-id 속성을 로그인 이벤트를 발생시키는 애플리케이션의 플랫폼과 버전으로 설정한다. 웹 기반이나 데스크톱 그리고 모바일 클라이언트 애플리케이션을 사용하는 환경에서는 메시지 본문을 검사하지 않고도 플랫폼별로 로그인을 추적하기 위해 계약을 맺어서 데이터를 추출할 수 있다. 통계 수집 전용 소비자를 구현하고 로그인 이벤트를 처리하는 소비자와 동일한 메시지를 구독한다면 이 기능이 특히 유용하다. app-id 속성을 제공하면 통계 수집 전용 소비자가 메시지 본문을 디코딩할 필요가 없다. 

 

사용법

• 버전관리
• 출처확인
• 통계

팁

큐에 대기하는 문제가 발생한 메시지의 출처를 추적할 때, app-id 를 이용하면 메시지의 출처를 쉽게 추적할 수 있으며, 이는 다수 애플리케이션이 동일한 RabbitMQ 인프라를 공유하는 대규모 환경에서 특히 유용하다. 기존 발행 애플리케이션과 동일한 익스체인지와 라우팅 키를 새로운 발행자 애플리케이션이 잘못 사용하는 경우가 종종 발생한다. 

 

3.8.2 user-id 속성

사용자 인증의 경우에는 로그인한 사용자를 식별하기 위해 user-id 속성을 사용하는 것이 유용해 보이지만, 대부분의 경우 권장되지 않는다. RabbitMQ 는 메시지를 발행하는 사용자에 대해 user-id 속성의 값으로 발행된 모든 메시지를 검사하고 두 값이 일치하지 않으면 메시지가 거부된다. 예를 들어 애플리케이션이 RabbitMQ 를 사용해 사용자 'www' 로 인증하고 메시지의 user-id 속성을 'linus' 로 설정할 경우 메시지가 거부된다. 

물론 작성하는 애플리케이션이 채팅이나 인스턴트 메시징 서비스라면 한 채팅방의 모든 사용자가 같은 user-id 를 사용해야 하며, 실제로 로그인한 실제 사용자를 식별하기 위해 user-id 를 사용할 수 는 없다. 

 

3.9 type 속성을 이용해 메시지 특정하기 

AMQP 0-9-1 버전에서 Basic.Properties 의 type 속성은 '메시지 유형 이름' 으로 정의돼 있는데, 애플리케이션 전용으로 공식적인 동작은 정해지지 않았다는 의미다. 익스체인지와 결합된 라우팅 키 값은 메시지 내용을 결정하는 데 필요한 만큼 메시지에 대한 많은 정보를 전달하는 데 반해, type 속성은 애플리케이션이 메시지 처리 방법을 결정 하는 데 또 다른 수단으로 사용된다. 

 

**자유 형식 문자열 값으로 메시지 유형을 정의하는 데 주로 사용된다. 

 

자체 설명 직렬화 형식이 충분히 빠르지 않은 경우 

type 속성은 자체 설명 (self-describing) 메시지를 만들 때, 특히 메시지 본문이 자체 설명 데이터 형식으로 직렬화되지 않은 경우 매우 유용하다. JSON, XML 같은 형식은 너무 장황한 편이다. 이럴 경우 Apache Thrift 혹은 Protobug 와 같은 직렬화 형식을 선택하기도 한다. 이들은 MessagePack 과는 달리 이진 코드화 메시지 형식에 자체 설명이 포함되지 않았으므로 직렬화 및 역직렬화를 위한 외부 정의 파일이 필요한데, 메시지에 자체 설명이 빠져있으므로 유선에서 더 적은 페이로드가 사용된다. 

발행자와 소비자 간에 실행 가능한 계약을 자체 설명하는 메시지와 달리 자체 설명하지 않는 메시지의 본문은 메시지를 소비자가 처리할 수 있는지 결정하기 전에 메시지 본문을 역직렬화해야 한다. 이 경우 type 속성을 사용해 레코드 유형이나 외부 정의 파일을 지정함으로써 소비자가 메시지를 처리하는데 사용하는 적절한 .thrift 나 .proto 파일에 접근할 수 없는 경우 처리할 수 없는 메시지로 판단해 거부할 수 있다. 

 

이제 이벤트를 데이터 웨어하우스에 저장하자. 단일 소비자가 모든 메시지를 처리하기 위해 일반 큐를 사용해 ETL (추출 extract_변환 transform-로드 load) 단계를 수행한다. ETL 을 위한 큐의 소비자는 여러 유형의 메시지를 처리하고 type 속성을 사용해 추출된 데이터를 저장할 시스템이나 테이블 또는 클러스터를 결정한다. 

 

노트 

ETL 은 최종적인 보고 목적으로 OLTP 데이터를 추출해서 데이터 웨어하우스에 로드하는 표준 방식이다. 

 

3.10 동적인 작업 흐름을 위한 reply-to 속성 사용하기 

AMQP 스펙에서 reply-to 속성은 공식적으로 정의된 동작은 없고 '애플리케이션 용도' 로만 지정돼 있다. 앞서 언급한 속성들과는 달리, 메시지에 대한 응답을 위한 개인 응답 큐를 지정하는 데 사용될 수 있다는 점을 주목할 만 하다. AMQP 스펙에 개인 응답 큐에 대한 명확한 정의가 명시돼 있지는 않지만, reply-to 속성은 특정 큐 이름이나 메시지가 원래 발행된 동일한 익스체인지의 응답 키를 전달하는 데 사용할 수 있다. 

 

경고

AMQP 0-9-1 버전에는 reply-to 속성에 대해 '요청 메시지에 사용될 때 개인 응답 큐의 읆을 보유할 수 있다.'는 경고가 있다. 이 정의는 모호하므로 사용할 때 주의해야 한다. reply-to 속성의 값 때문에 응답 메시지가 라우팅될 수 없는 경우, RabbitMQ 가 메시지 발행을 거부할 수 있다. 

**reply-to 속성은 RPC 스타일의 메시지의 응답에 소비자가 사용해야 하는 라우팅 키를 전달하는 데 사용할 수 있다. 

 

3.11 headers 를 사용해 사용자 속성 지정하기 

사용자 정의 키와 값을 갖는 테이블이다. 키는 최대 255자의 길이를 갖는 ASCII 또는 유니코드 문자열을 설정할 수 있다. 값은 유효한 AMQP 값 유형을 설정할 수 있다. 다른 속성들과 달리 headers 속성을 사용하면 원하는 모든 데이터를 headers 테이블에 추가할 수 있다. headers 속성에는 특별한 기능이 있는데, RabbitMQ 는 라우팅 키를 사용하는 대신 헤더 테이블에 채워진 값을 기반으로 메시지를 라우팅할 수 있다는 점이다. headers 속성을 통한 메시지 라우팅은 6장에서 알아본다. 

 

예시

키 (ASCII 또는 유니코드 문자열)	값 (모든 AMQP 데이터 유형)
foo	bar
corge	1
grault	True
graply	1921-08-19

 

3.12 priority 속성

RabbitMQ 3.5.0 부터 AMQP 스펙에 맞춰서 priority 필드가 구현됐다. priority 속성의 값은 큐에 포함된 메시지의 우선순위 지정에 사용하며, 0~9까지의 값을 갖는 정수로 정의된다. priority 가 9인 메시지가 발행되고 나서 priority 가 0 인 메시지가 발행되면 새로 연결된 소비자 애플리케이션은 priority 가 9인 메시지보다 0인 메시지를 먼저 받게 된다. (낮은 것 부터 받음) 흥미롭게도 RabbitMQ 는 priority 속성을 부호 없는 바이트 (unsigned byte) 로 구현해 0에서 255사이의 값을 지정할 수 있지만, AMQP 스펙과 상호운용성을 유지하려면 priority 를 0에서 9로 제한해야 한다. 

 

3.13 사용할 수 없는 속성: cluster-id/reserved 

cluster-id 는 AMQP 0-8 에서 정의됐으며 AMQP 0-9-1 에서 제거됐다. 

AMQP 0-9-1 에서 예약됨 (reserved) 으로 변경됐으므로 사용하지 말아야 한다. 

이 속성은 가급적 사용하지 않는 것이 좋다. 

 

3.14 요약

속성	유형	사용처	명시된 내용
app-id	짧은 문자열	애플리케이션	메시지르 발행하는 애플리케이션을 정의
content-encoding	짧은 문자열	애플리케이션	메시지 본문이 zlib, deflate 또는 Base64 와 같은 특별한 방법으로 인코딩 되는지 지정
content-type	짧은 문자열	애플리케이션	mime-types 를 사용해 메시지 본문의 유형 지정
correlation-id	짧은 문자열	애플리케이션	메시지가 참조하는 내용
delivery-mode	octet	RabbitMQ	1은 RabbitMQ 가 메시지를 메모리에 보관 2는 디스크에 기록
expiration	짧은 문자열	RabbitMQ	메시지가 만료되는 시기를 나타내는 데 사용하는 텍스트 문자열의 유닉스 시간 값
headers	테이블	양쪽 모두	메시지에 대한 추가적인 메타데이터를 첨부하는 데 사용하는 자유 형식 키/값 테이블. 원하는 경우 RabbitMQ 가 이 값을 기반으로 라우팅 가능
message-id	짧은 문자열	애플리케이션	메시지를 식별하는 데 사용할 수 있는 UUID 와 같은 고유 식별자
priority	octet	RabbitMQ	큐에서 메시지의 우선순위를 지정하는 속성
timestamp	timestamp	애플리케이션	메시지 작성 시점을 나타내는 데 사용하는 유닉스 시간 값
type	짧은 문자열	애플리케이션	에플리케이션이 메시지 유형 또는 페이로드를 설명하는 데 사용할 수 있는 텍스트 문자열
user-id	짧은 문자열	양쪽 모두	연결된 사용자에 대해 유효성을 검증하고 일치하지 않는 메시지를 삭제하는 자유 형식 문자열

 

4. 메시지 발행에서 성능 절충

여러 서버에 걸쳐 있는 HA 큐와 같은 다양한 메시지 배달 보장 수준을 선택할 수 있다. 4장에서는 이러한 기능을 사용하는 데 관련된 성능과 배달 보장의 성능 절충에 대해 알아보고 RabbitMQ 가 자동으로 발행자의 메시지를 조절하는 방법을 알아본다. 

 

4.1 발행 속도와 배달 보장의 균형 잡기 

RabbitMQ 서버를 재부팅해도 메시지가 유지되도록 하는 등의 일부 기능은 특정 애플리케이션에서는 너무 느리고 적합하지 않을 수 있다. 반면에 추가적인 배달 보장 없이 메시지를 발행한다면, 속도는 훨씬 빠르지만 미션 크리티컬 애플리케이션에서는 안전한 환경을 제공하지 못한다. 

 

토끼 <-> 거북이

• 배달 보장 없음
• 실패 통보
• 발행자 확인
• 대체 익스체인지 
• HA 큐 
• 트랜잭션 
• 트랜잭션 HA 큐 
• 메시지 디스크에 저장 

자체적인 성능 벤치마크를 수행해 성능과 배달 보장 간에 적절한 균형을 결정하는 것이 좋다. 

올바른 솔루션을 위한 고성능과 메시지 배달 보장 사이에 적절한 균형을 찾는 과정에서는 다음 질문을 염두에 두길 바란다. 

• 발행 시에 미시지에 큐를 넣는 것이 얼마나 중요한가?
• 메시지를 라우팅할 수 없는 경우, 발행자에게 메시지를 보내야 하는가? 
• 메시지를 라우팅할 수 없는 경우, 차후에 조정하는 다른 곳으로 메시지를 보내야 하는가? 
• RabbitMQ 서버에 장애가 발생할 때, 메시지가 손실돼도 괜찮은가? 
• RabitMQ 가 새 메시지를 처리할 때, 요청한 모든 메시지를 라우팅한 후 디스크에 저장하는 작업이 정상적으로 수행했는지 발행자가 확인해야 하는가?
• 발행자가 메시지를 한꺼번에 전달하면 RabbitMQ 는 메시지를 라우팅하고 디스크에 저장한 후 작업이 정상적으로 실행됐는지를 발행자에게 다시 알려야 하는가? 
• 다수 메시지를 라우팅한 후 디스크에 정상적으로 저장됐는지 확인하는 작업을 일괄 처리하는 경우, 메시지를 저장할 큐에 원자 커밋 (atomic commit) 이 필요한가? 
• 발행자가 적절한 성능과 메시지 처리량을 달성하는데, 메시지 배달 보장 기능 간에 절충점이 있는가? 
• 메시지 발행의 다른 측면이 메시지 처리량 및 성능에 영향을 미치는가? 

HA 큐와 필수 (mandatory) 라우팅을 결합해서 선택하거나 delivery-mode 를 2로 설정하고 트랜잭션 발행을 선택해 메시지를 디스크에 보관할 수 있다. 애플리케이션 개발 시에 균형 잡힌 조합을 발견할 때 까지, 각각 다른 옵션을 조합해 시도하는 것을 추천한다. 

 

4.1.1 배달 보장을 사용하지 않는 환경 

이상적인 환경이라면 RabbitMQ 는 추가 구성이나 설정 없이 메시지를 안정적으로 전달한다. 올바른 Basic.Publish 를 통해 익스체인지, 라우팅 정보와 함께 메시지를 발행하면 RabbitMQ 가 메시지를 수신하고 적절한 큐에 전달한다. 네트워크 문제가 없고 서버 하드웨어가 안정적이며 장애도 발생하지 않는다면, 운영체제가 RabbitMQ 메시지 브로커의 운영 상태에 영향을 미치는 문제는 발생하지 않는다. 

불행하게도 현실 세계에는 완벽한 환경에서는 결코 일어나지 않을 일들이 정기적으로 발생한다. 

미션 크리티컬한 애플리케이션이 아닌 경우, 일반적인 메시지 발행 중 발생 가능한 모든 장애를 처리할 필요는 없으며, 적절한 처리만 해도 안정적이고 예측 가능한 가동 시간을 확보할 수 있다. 

미션 크리티컬한 애플리케이션이 아닌 경우에는 RabbitMQ 의 기본 설정으로도 적절한 수준의 안정적인 메시징 환경을 구축할 수 있다. 배달 보장을 사용하지 않아도 적절한 환경에 대해 살펴보자. 

이 메시지는 서버의 CPU 로드, 메모리, 네트워크 사용과 같은 정보를 전달한다. 

RabbbitMQ 와의 연결이 끊어지면 다음 번에 통계 데이터를 보내야 할 때 다시 연결을 시도한다. 마찬가지로 소비자 애플리케이션은 연결이 끊어지면 다시 연결하고 이전에 사용하던 동일한 큐에서 다시 메시지를 소비한다. 

 

4.1.2 mandatory 플래그를 설정한 메시지를 라우팅할 수 없을 때 

서버 모니터링 데이터가 항상 RabbitMQ 로 배달되도록 보장하려면, collectd 에서 RabbitMQ 에 발행하는 메시지의 mandatory 를 설정한다. mandatory 플래그는 Basic.Publish RPC 명령과 함께 전달되는 인수인데, 메시지를 라우팅할 수 없으면 Basic.Return RPC 를 통해 RabbitMQ 가 메시지를 발행자에게 다시 보내도록 지시한다. mandatory 플래그는 오류 감지 모드를 켜는 것으로 간주할 수 있는데, 메시지 라우팅 실패를 알리는 데 사용한다. 메시지 라우팅이 올바르게 처리되면 발행자에게 별도의 메시지를 전송하지 않는다. mandatory 플래그를 설정한 메시지를 발행하기 위해 다음과 같이 익스체인지, 라우팅 키, 메시지, 속성을 전달한 후 인수를 전달한다. 

 

노트

위 예제 코드에서는 Connection 과 Channel 객체를 호출하는 새로운 방법을 사용했는데, 두 객체 모두 컨텍스츠 관리자로 생성된다. (with xxx as connection: 문법) 

 

Rabbit MQ 의 Basic.Return 은 비동기로 동작하며 메시지가 발행된 후 언제든지 발생할 수 있다. 예를 들어 RabbitMQ 에 통계 데이터를 발행하는 데 실패할 경우, collectd 가 Basic.Return 호출을 받기 전에 다른 데이터를 계속 발행할 수 있다. 

 

mandatory 예외처리

import datetime
import rabbitpy
import time

# Connect to the default URL of amqp://guest:guest@localhost:15672/%2F
url = 'amqp://guest:guest@localhost:5672/%2F'
connection = rabbitpy.Connection(url)
try:
    with connection.channel() as channel:
        properties = {'content_type': 'text/plain',
                      'timestamp': datetime.datetime.now(),
                      'message_type': 'graphite metric'}
        body = 'server.cpu.utilization 25.5 1350884514'
        message = rabbitpy.Message(channel, body, properties)
        message.publish('chapter2-example',
                        'server-metrics',
                        mandatory=True)
except rabbitpy.exceptions.MessageReturnedException as error:
    print('Publish failure: %s' % error)

 

다른 라이브러리의 경우, 메시지를 발행할 때 RabbitMQ 에서 Basic.Return RPC 를 전달받으면 실행할 콜백 메소드를 등록해야 할 수 있다. 비동기적으로 Basic.Return 메시지를 처리할 때 다른 메시지를 소비하는 것 처럼 Basic.Return 메소드 프레임, 콘텐츠 헤더 프레임, 바디 프레임을 받게 되는데, 복잡해 보여도 크게 걱정할 필요는 없다. 프로세스를 단순화하고 메시지 라우팅 오류를 처리하는 다른 방법이 있는데, 그 중 하나는 RabbitMQ 에서 발행자 확인을 사용하는 것이다. 

 

노트

rabbitpy 라이브러리는 Basic.Publish 명령을 보낼 때 최대 세 개의 인수만 받는데, 이는 추가 인수로 immediate 플래그가 포함된 AMQP 스펙과 대조적이다. immediate 플래그는 메시지 브로커가 메시지를 즉시 라우팅할 수 없는 경우 Basic.Return 을 반환하도록 지시한다. immediate 플래그는 RabitMQ 2.9 이후로 더 이상 사용되지 않으며, 사용할 경우 예외가 발생되고 채널이 닫힌다. 

 

4.1.3 트랜잭션보다 가벼운 발행자 확인 

발행자 확인 (Publisher Confirms) 은 AMQP 스펙의 확장 기능으로 RabitMQ 관련 확장을 지원하는 클라이언트 라이브러리에서만 지원된다. 디스크에 메시지를 저장하는 것으로도 메시지 손실을 막을 수 있지만, 이것으로는 발행자와 RabbitMQ 사이에 메시지가 전달됐음을 확신할 수 없다. 메시지를 발행하기 전에 메시지 발행자는 RabbitMQ 에 Confirm.Select PRC 요청을 전송하고 메시지가 전달됐는지 확인하기 위해 Confirm.SelectOk 응답을 기다린다. 이 시점부터 발행자가 RabbitMQ 에 보내는 각 메시지에 대해 서버는 수신 확인 (Baisc.Ack) 또는 부정 수신 확인 (Basic.Nack) 으로 응답하며, 메시지의 오프셋을 지정하는 정수 값을 포함하거나 확인한다. 확인 번호는 Confirm.Slect RPC 요청 다음에 수신된 순서에 따라 메시지를 참조한다. 

python 4.1.3\ Publisher\ Confirms.py

 

 

 

노트

발행자 확인의 사용 여부와 상관없이 존재하지 않는 익스체인지에 메시지를 발행할 경우, 채널은 RabbitMQ 에 의해 종료된다. 이 경우 rabbitpy 에서는 rabbitpy.exceptions.RemoteClosedChannelException 예외가 발생한다. 

 

4.1.4 라우팅할 수 없는 메시지를 위한 대체 익스체인지 사용하기 

대체 익스체인지 (alternate exchange) 는 라우팅할 수 없는 메시지를 처리하기 위해 RabbitMQ 팀이 AMQP 를 확장한 또 다른 예다. 대체 익스체인지는 처음 익스체인지를 선언할 때 명시되며, RabbitMQ 에서 익스체인지가 라우팅할 수 없으면, 새로운 익스체인지가 메시지를 라우팅할 기존의 익스체인지를 대신해 지정된다. 

**라우팅할 수 없는 메시지가 대체 익스체인지를 정의한 익스체인지에 발행되면, 메시지느 대체 익스체인지에 전달된다.

 

노트

대체 익스체인지가 설정된 익스체인지로 메시지를 보낼 때 mandatory 플래그를 설정하면 의도한 익스체인지가 메시지를 정상적으로 라우팅할 수 없는 경우 Basic.Return 이 발행자에게 직접 전송되지 않는다. 라우팅할 수 없는 메시지를 대체 익스체인지에 보내는 동작은 mandatory 플래그를 True 로 설정한 메시지에도 동일하게 적용된다. RabbitMQ 의 메시지 라우팅 패턴이 다른 익스체인지와 마찬가지로 대체 익스체인지에도 동일하게 적용된다는 점을 주의해야 한다. 큐가 원래 라우팅 키가 명시된 메시지를 수신하도록 바인딩되지 않은 경우 메시지는 큐에 추가되지 않고 손실된다. 

 

대체 익스체인지를 사용하려면, 먼저 라우팅 할 수 없는 메시지를 전송할 익스체인지를 설정해야 한다. 기본 익스체인지를 서정한 후, 메시지를 발행할 때 Exchange.Declare 명령에 alternate-exchange 인수를 추가한다. 다음 예제 코드에는 이전 예제 코드에 라우팅할 수 없는 메시지를 저장하는 메시지 큐가 더 추가됐다. 

 

alternate-exchange 는 Features 가 AE 로 표기된다.

 

대체 익스체인지를 팬아웃 (fanout) 유형으로 생성했지만, graphite 익스체인지는 토픽 (topic) 유형으로 생성했다. 팬아웃 익스체인지는 자신이 알고 있는 모든 큐에 메시지를 전달하고 토픽 익스체인지는 라우팅 키를 기반으로 선택적으로 메시지를 라우팅할 수 있다. 익스체인지의 유형에 대해서는 5장에서 자세히 알아본다. 두 익스체인지를 생성하고 unroutable-message 라는 이름의 큐를 대체 익스체인지에 연결한다. 

바인딩은 여기서 확인할 수 있다.

이후 graphite 익스체인지에 메시지가 발행되고 라우팅될 수 없는 메시지는 unroutable-message 큐에 저장된다. 

 

4.1.5 트랜잭션으로 배치 처리하기

RabbitMQ 확장 스펙인 발행자 확인을 구현하기 전에 메시지 전달을 보장하기 위한 유일한 방법은 트랜잭션이었다. AMQP 트랜잭션 혹은 TX 클래스는 메시지를 일괄로 RabbitMQ 에 발행한 후 큐에 커밋하거나 롤백할 수 있는 매커니즘을 제공한다. 다음 예제는 '4.1.5 Transactional Publishing' 노트북에 포함돼 있다. 

import rabbitpy

with rabbitpy.Connection() as connection:
    with connection.channel() as channel:
        tx = rabbitpy.Tx(channel)
        tx.select()
        message = rabbitpy.Message(channel,
                                   'This is an important message',
                                   {'content_type': 'text/plain',
                                    'delivery_mode': 2,
                                    'message_type': 'important'})
        message.publish('chapter4-example', 'important.message')
        try:
            if tx.commit():
                print('Transaction committed')
        except rabbitpy.exceptions.NoActiveTransactionError:
            print('Tried to commit without active transaction')

존재하지 않는 익스체인지와 같은 오류로 인해 RabbitMQ 가 메시지를 라우팅할 수 없으면 TX.CommitOk 응답을 보내기 전에 Basic.Return 응답이 반환된다. 발행자가 트랜잭션을 중단하려는 경우 TX.Rollback RPC 요청을 보내고 계속 진행하기 전에 미싲 브로커의 TX.RollbackOk 응답을 기다려야 한다. 

 

RabbitMQ 와 원자 트랜잭션 

불행하게도 RabbitMQ 는 발행한 모든 명령이 단일 큐에 영향을 줄 때만 원자 트랜잭션을 지원한다. 트랜잭션의 명령이 둘 이상의 큐에 영향을 주면 커밋은 원자적으로 동작하지 않는다. 

 

메시지 발행 확인을 목적으로 트랜잭션의 사용을 고려한다면 좀 더 단순한 발행자 확인을 사용하는 것을 추천하는데, 발행자 확인이 더 빠르며 성공과 실패를 확인할 수 있다. 

그러나 대부분의 경우 발행자 확인 뿐 아니라 메시지가 큐에 있는 동안 손실되지 않는 것이 중요한데, 이는 HA 큐로 보장할 수 있다. 

 

4.1.6 HA 큐를 사용해 노드 장애 대응하기 

미션 크리티컬 메시징 아키텍처에서는 가용성이 높은 HA 가 중요한 역할을 한다. HA 큐도 AMQP 스펙이 아닌 RabbitMQ 팀이 만든 확장 기능이며, 큐를 여러 서버에 중복해 복사본을 저장하는 기능을 제공한다. 

HA 큐는 클러스터로 구성된 RabbitMQ 환경이 필요하며 AMQP API 또는 웹 기반 관리자 UI 로 설정할 수 있다. 이어서는 AMQP API 를 이용해 설정하는 방법을 알아보고, 웹 기반 관리자 UI 를 사용해 HA 큐에 대한 정책을 설정하는 방법은 8장에서 자세히 알아보자. 

import rabbitpy

connection = rabbitpy.Connection()
try:
    with connection.channel() as channel:
        queue = rabbitpy.Queue(channel,
                               'my-ha-queue',
                               arguments={'x-ha-policy': 'all'})
        if queue.declare():
            print('Queue declared')
except rabbitpy.exceptions.RemoteClosedChannelException as error:
    print('Queue declare failed: %s' % error)

메시지가 HA 큐로 설정된 큐에 발행되면 HA 큐를 담당하는 클러스터의 각 서버로 메시지가 전송된다. 클러스터의 노드가 메시지를 소비하면 다른 노드의 모든 메시지 복사본이 즉시 제거된다. 

개별 노드를 지정하려면 x-ha-policy: all 대신 nodes 를 인수로 전달하고 다음 인수인 x-ha-nodes 에 큐의 노드 목록을 지정한다. 

 

4.1.6 Selective HA Queue Declaration 

import rabbitpy

connection = rabbitpy.Connection()
try:
    with connection.channel() as channel:
        arguments = {'x-ha-policy': 'nodes',
                     'x-ha-nodes': ['rabbit@node1',
                                    'rabbit@node2',
                                    'rabbit@node3']}
        queue = rabbitpy.Queue(channel,
                               'my-2nd-ha-queue',
                               arguments=arguments)
        if queue.declare():
            print('Queue declared')
except rabbitpy.exceptions.RemoteClosedChannelException as error:
    print('Queue declare failed: %s' % error)

 

노트

node1, node2, node3 가 정의돼 있지 않더라도 큐를 정의할 수 있으며 메시지를 발행하는 경우 큐에 전달된다. 나열된 노드 중 하나 이상이 존재하는 경우 메시지는 해당 서버에 저장된다. 

 

다운된 노드가 다시 추가되거나 새 노드가 클러스터에 추가되더라도 기존 노드의 큐에 이미 존재하는 메시지는 포함되지 않는다. 대신 이전에 발행한 모든 메시지가 소비되면 모든 새 메시지가 수신되고 동기화된다. 

 

4.1.7 HA 큐 트랜잭션 

HA 큐에서 트랜잭션 또는 발행자 확인을 사용하는 경우, 메시지가 HA 큐의 모든 활성 노드에 있는 것으로 확인될 때 까지 RabbitMQ 는 성공 응답을 보내지 않는다. 이로 인해 발행자 애플리케이션에 대한 응답이 지연될 수 있다. 

 

4.1.8 delivery-mode 2를 사용해 메시지를 디스크에 저장하기

이어서 또 다른 배달 보장에 대해 알아보자. RabbitMQ 서버가 메시지를 소비하기 전에 특정 이유로 노드가 다운될 경우, RabbitMQ 에 메시지를 발행할 때 디스크에 저장하도록 설정하지 않는다면 메시지는 영원히 손실된다. 

 

노트

delivery-mode 2 외에 RabbitMQ 서버를 다시 시작한 후에도 메시지가 남아있게 하려면 큐를 만들 때 durable 로 선언돼야 한다. 

**OLTP (On-line Transaction Processing, 온라인 트랜잭션 처리) 

일반적으로 대부분의 웹 애플리케이션에서 쓰기의 비율은 낮다. 

대부분의 운영체제에서 커널은 디스크에서 읽은 페이지를 버퍼링하기 위해 RAM 의 여유분을 사용하지만, 디스크 쓰기를 캐시하는 유일한 컴포넌트는 디스크 컨트롤러와 디스크다. 이 때문에 메시지를 디스크에 저장할 때 하드웨어의 스펙을 올바르게 설정하는 것이 중요하다. 소형 서버에서는 쓰기 작업량이 과한 경우 RabbitMQ 의 동작이 매우 느려질 수 있다. 

 

4.2 RabbitMQ 푸시백

AMQP 스펙에는 RabbitMQ 서버 구현에 유리하지 않은 발행자의 가정이 있다. RabbitMQ 버전 2.0 이전에는 발행자 애플리케이션이 너무 빨리 메시지를 발행해 RabbitMQ 를 압도하기 시작한 경우 발행자에게 Channel.Flow RPC 메소드를 보내 차단하고 다른 Channel.Flow 명령을 받을 때 까지 더 이상 메시지를 보내지 않도록 지시한다. 

이는 Channel.Flow 명령을 처리하지 않거나 잘못 처리하는 발행자 애플리케이션의 경우, 메시지 발행을 늦추는데 상당히 비효율적인 방법으로 알려졌다. RabbitMQ 3.2 버전 이전에 RabbitMQ 팀은 Channel.Flow 의 사용을 중단했으며, 이를 TCP 배압 (Back-Pressure) 매커니즘으로 대체해 문제를 해결했다. 발행자에게 정중하게 요청하지 않고 RabbitMQ 는 TCP 소켓에서 하위 수준의 데이터 수신을 중지한다. 이는 단일 발행자에게 RabbitMQ 가 압도당하지 않도록 보호한다. 

 

내부적으로 RabbitMQ 는 크레딧 (credit) 이라는 개념을 사용해 발행자에 대해 언제 푸시백을 할 것인지를 관리한다. 새로운 연결이 생성되면 이 연결에 미리 사용할 수 있는 크레딧의 양이 할당되고 RabbitMQ 가 각 RPC 명령을 수신하면 크레딧이 감소한다.

연결에 남은 크레딧이 없으면 크레딧이 생길 때까지 무시한다. 

RabbitMQ 3.2 부터 RabbitMQ 팀은 AMQP 스펙을 확장해 연결에 대한 크레딧이 임곗값에 도달했을 때 전송되는 알림을 추가하고 클라이언트에 연결이 차단됐다는 사실을 알린다. Connection.Blocked 와 Connection.Unblocked 는 RabbitMQ 가 발행자 클라이언트를 차단하거나 해당 블록이 제거됐을 때 언제든지 클라이언트에 알릴 수 있는 비동기 메소드다. 대부분의 주요 클라이언트는 이 기능을 구현했다. 사용중인 클라이언트 라이브러리에도 애플리케이션의 연결 상태를 결정하는 방법이 구현됐는지 확인해야 한다. 이어지는 절에는 rabbitpy 로 이 검사를 수행하는 방법과 3.2 이전의 RabbitMQ 버전에서 관리 API 를 활용해 연결의 채널이 차단됐는지 확인하는 방법을 설명한다. 

 

4.2.1 rabbitpy 로 연결 상태 확인하기 

Connection.Blocked.. 

5. 메시지를 받지 않고 소비하기 

6. 익스체인지 라우팅을 통한 메시지 패턴

2부 데이터 센터 또는 클라우드에서 RabbitMQ 운영하기

3부 통합과 맞춤 설정