Python에서 RabbitMQ 기반으로 JSON 메시지를 송/수신 하는 예제이다. import json channel.basic_publish(exchange='', routing_key='task_queue', body=json.dumps(message), properties=pika.BasicProperties( delivery_mode = 2, # make message persistent )) def callback(ch, method, properties, body): print(" [x] Received %r" % json.loads(body))

메시지 구성요소 - deliveryMode : 메시지의 지속성 (값 2) 또는 일시적 (기타 값)을 표시. - contentType : 내용물의 mime-type - replyTo : 일반적으로 callback queue의 이름을 지정하는데 사용 - correlationId : RPC 응답을 요청과 연관시키는데 사용 (요청을 구분할 수 있는 유일값) Correlation Id 모든 RCP 요청에 대해 Callback Queue 생성 혹은 클라이언트 당 하나의 Callback Queue 생성 Callback Queue에서 메시지를 받으면 이 속성을 보고 응답과 요청을 일치 시킨다. 알 수 없는 correlationId를 보게 되면 안전하게 그 메시지를 버린다. 동작 설명 Client는 Server에게 Re..

RabbitMQ AMQP 프로토콜을 구현해 놓은 프로그램 신뢰성, 안정성과 성능을 충족할 수 있도록 다양한 기능 제공 유연한 라우팅: Message Queue가 도착하기 전에 라우팅 되며 플러그인을 통해 더 복잡한 라우팅도 가능 클러스터링: 로컬 네트워크에 있는 여려 RabbitMQ 서버를 논리적으로 클러스터링 할 수 있고 논리적인 브로커도 가능 함. 오픈 소스로 상업적 지원이 가능 Kafka 대용량 실시간 로그 처리에 특화되어 설계된 메시징 시스템 기좀 범용 메시징 시스템 대비 TPS(Transaction Per Second)가 매우 우수함 특화된 시스템이기 때문에 범용 메시징 시스템에서 제공하는 다양한 기능들은 제공 되지 않음 AMQP 프로토콜이나 JMS API를 사용하지 않고 단순한 메시지 헤더를 ..

Exchange의 "direct" type을 이용하여, 특정 routing_key를 binding Queue에게 Publisher가 데이터를 보내고, Consumer가 데이터를 수신하는 방법을 설명 한다. 아래 그림의 경우, Publisher가 Excahnge에게 "info" routing_key로 메시지를 보내면, Consumer2가 데이터를 수신 하고, "error" routing_key로 메시지를 보내면 Consumer1 & Consumer2가 데이터를 수신 한다. 즉 아래 방식을 사용하면, Publisher는 원하는 Queue에게만 메시지를 보낼 수 있고, Consumer는 원하는 Queue의 메시지만 수신 할 수 있다. receive_logs.direct.py import pika import ..

앞에 [RabbitMQ #2]Simple Send/Receive 예제에서는 Queue에 직접 메시지를 보내고 Queue에서 직접 메시지를 받았다. 본 글에서는 Exchanges를 사용해 Pub/Sub 메시지 구조를 구현하는 방법을 설명 한다. Exchange type을 fanout으로 설정 하면, Exchange에 Binding 되어 있는 모든 Queue에게 메시지를 전달 한다. 이 기능을 이용해 Pub/Sub 메시지 구조를 구현 할 수 있다. - Subscriber #1/#2에서 "logs" Exchange를 Binding 하면 Pub/Sub 구조에서 Subscribe 하는 동작과 유사하게 동작 한다. - Publisher가 "info: Hello World!" 메시지를 "logs" Exchange에 ..

Exchanges Type Exchanges는 메시지에 포함된 routing_key와 일치하는 Queue에 메시지를 전달한다. RabbitMQ는 다음과 같은 4개의 Exchanges Type을 지원하며, Exchange에 대한 자세한 내용은 [AMQP] AMQP(Advanced Message Queuing Protocol) 에서 확인 할 수 있다. Direct Exchange Producer가 보내는 메시지의 routing_key와 일치 하는 Queue에게만 데이터를 보냄 Fanout Exchange Binding 되어 있는 모든 Queue에게 데이터를 보냄 Topic Exchange 와일드 카드(*,#)를 이용해 메시지를 Queue에 매칭하는 방법 * : 하나의 단어를 의미 # : 0개 이상의 단어를 ..