Python에서 RabbitMQ Connection 및 Consumer 등록

I. Import 하기

from carrot import connection as carrot_connection
from carrot import messaging

II. Connection 객체 생성

params = dict(hostname=FLAGS.rabbit_host,
                    port=FLAGS.rabbit_port,
                    ssl=FLAGS.rabbit_use_ssl,
                    userid=FLAGS.rabbit_userid,
                    password=FLAGS.rabbit_password,
                    virtual_host=FLAGS.rabbit_virtual_host)

connection = carrot_connection.BrokerConnection(params);

III. Consumer 생성

consumer = messaging.Consumer( connection=self,
                                                  topic=topic,
                                                  proxy=proxy)

IV. Sample (http://nathanborror.com/posts/2009/may/20/working-django-and-rabbitmq/)

########## Set global variables ###################

AMQP_SERVER = 'localhost'
AMQP_PORT = 5672
AMQP_USER = 'guest'
AMQP_PASSWORD = 'guest'
AMQP_VHOST = '/'

########## Create a consumer ###################

>>> from flopsy import Connection, Consumer
>>> consumer = Consumer(connection=Connection())
>>> consumer.declare(queue='books', exchange='readernaut', routing_key='importer', auto_delete=False)

>>> def message_callback(message):
...     print 'Recieved: ' + message.body
...     consumer.channel.basic_ack(message.delivery_tag)
>>>
>>> consumer.register(message_callback)
>>> consumer.wait()


########## Create a publisher ###################

>>> from flopsy import Connection, Publisher
>>> publisher = Publisher(connection=Connection(), exchange='readernaut', routing_key='importer')
>>> publisher.publish('Test message!')
>>> publisher.close()

[ Pool 사용 ]

1. Publisher 가 cast 로 호출할 때 pools.Pool을 사용 (Connection 을 Pool로 연결)

from eventlet import pools

class Pool(pools.Pool):
    def create(self):
        LOG.debug('Pool creating new connection')
        return Connection.instance(new=True)

ConnectionPool = Pool(
        max_size=FLAGS.rpc_conn_pool_size,
        order_as_stack=True)

with ConnectionPool.item() as conn:
        publisher = DirectPublisher(connection=conn, msg_id=msg_id)

2. Consumer 가 메세지를 받을 때 GreenPool을 사용 (메소드를 Pool로 실행)

from eventlet import greenpool

self.pool = greenpool.GreenPool(FLAGS.rpc_thread_pool_size)
self.pool.spawn_n(self._process_data, msg_id, ctxt, method, args)

def _process_data(self, msg_id, ctxt, method, args):
...

 
3. Consumer 를 Thread로 변경하여 실행을 분기시키고자 할 때 eventlet.spawn을 사용
    - 코드를 wait() block 없이 계속 실행
    - 해당 Thread 가 return 값으로 넘어옮

import eventlet

self._rpc_consumer_thread = eventlet.spawn(_consumer_thread)

 
4. Report 와 같이 내부 LoopingCall 을 분기시켜 실행하고자 할 때 eventlet.greenthread 를 사용

from eventlet import greenthread

def _inner():
    ...
    greenthread.sleep(interval)
    ...

greenthread.spawn(_inner)

[ 용어 정리 ]
- queue      : Queue 이름을 의미
- exchange : Queue 와 Bind 하기 위한 exchange 이름을 의미
- exchange_type :
    . direct  : routing_key 가 정확하게 매칭되어야 함
    . topic   : routing_key 를 패턴 매칭으로 사용 가능
    . fanout : routing_key 가 필요 없이 모두 통신
- routing_key : Key 이름에 해당하는 Queue 에만 메세지를 보낼 수 있음