RabbitMQ 작업 큐로 이용하기

메세지를 받아간 워커가 작업중에 죽더라도 해당 메세지는 큐에 남아서 다른 워커에게 전달할 수 있게 하기위해서 RabbitMQ는 acknowledgments를 지원함.

ack(nowledgments)는 워커가 메세지를 받아가서 처리했다는걸 RabbitMQ에게 알려서 RabbitMQ가 그 메세지를 지울 수 있게함. ack없이 워커가 죽으면 RabbitMQ는 그 메세지가 정상적으로 처리되지 않았다고 판단해서 다른 워커에게 메세지를 전달. 타임아웃은 없음.

RabbitMQ 서버가 죽더라도 큐안의 메세지를 보존하기위해서는 durable 옵션을 True로 설정해서 큐를 생성함.

라운드로빈방식으로 메세지를 보내서 특정 워커에 작업이 쌓이는 것을 방지하기위해 basic_qos 옵션을 줄 수 있음. 

메세지 흐름 구조

예제소스

new_task.py

#!/usr/bin/env python import pika import sys connection = pika.BlockingConnection(pika.ConnectionParameters( host='localhost')) channel = connection.channel() channel.queue_declare(queue='task_queue', durable=True) message = ' '.join(sys.argv[1:]) or "Hello World!" channel.basic_publish(exchange='', routing_key='task_queue', body=message, properties=pika.BasicProperties( delivery_mode = 2, # make message persistent )) print " [x] Sent %r" % (message,) connection.close()

worker.py

#!/usr/bin/env python import pika import time connection = pika.BlockingConnection(pika.ConnectionParameters( host='localhost')) channel = connection.channel() channel.queue_declare(queue='task_queue', durable=True) print ' [*] Waiting for messages. To exit press CTRL+C' def callback(ch, method, properties, body): print " [x] Received %r" % (body,) time.sleep( body.count('.') ) print " [x] Done" ch.basic_ack(delivery_tag = method.delivery_tag) channel.basic_qos(prefetch_count=1) channel.basic_consume(callback, queue='task_queue') channel.start_consuming()

여러개의 worker.py를 실행한 다음 new_task.py를 실행해 보면 테스트해볼 수 있음.

참고

http://www.rabbitmq.com/tutorials/tutorial-two-python.html