我正在实现一个带有 kombu 的 rpc 框架,以便与兔子一起使用我想使用 rabbitmq 的直接回复功能,但我找不到使用 kombu 实现它的方法。
客户端在生成消息之前需要在"amq.rabbitmq.reply-to"队列上使用,并且生产者和消费者应使用相同的通道。我还需要使用生产者池(或某种连接池(,因为客户端是在线程环境中创建的。
到目前为止,我有这段代码,rabbitmq 不会抱怨没有生产者出现 PRECONDITION 错误(如果我删除消费者部分,它确实会抱怨(,但生产者不会生产任何东西!
class KombuRpcClient(RpcClientBase):
def __init__(self, params):
self.future = Queue.Queue()
self.logger = logger
if isinstance(params, RpcConnectionProperties):
self.rpc_connection_properties = params
else:
self.rpc_connection_properties = RpcConnectionProperties(
host=params.get('host'),
port=5672,
username=params.get('username'),
password=params.get('password'),
vhost=params.get('vhost') if params.has_key('vhost') else '/'
)
self.amqp_url = self.rpc_connection_properties.get_kombu_transport_url('pyamqp')
self.reply_queue = KombuQueue('direct_reply', exchange=default_exchange, routing_key='amq.rabbitmq.reply-to')
def call(self, exchange, key, msg, no_response=False, timeout=5):
connection = Connection(self.amqp_url)
if exchange is not None:
key = exchange + ':' + key
with producers_pool[connection].acquire(block=True) as producer:
with producer.channel.Consumer(queues=[self.reply_queue], no_ack=True, callbacks=[self._on_message],
accept=['ujson']) as consumer:
producer.publish(
msg,
exchange=default_exchange,
routing_key=key,
immediate=True,
serializer='ujson', reply_to=self.reply_queue.routing_key)
consumer.consume()
pass
res = self.future.get(block=True, timeout=timeout)
print res
def cast(self, exchange, key, msg):
pass
def _on_message(self, body, message):
print body
self.future.put(body)
在wireshark的帮助下,我意识到有时rabbitmq会响应该PRECONDITION错误,但kombu不会引发异常,我不知道为什么!无论如何,这段代码现在可以工作了:
class KombuRpcClient(RpcClientBase):
def __init__(self, params):
self.future = Queue.Queue()
self.logger = logger
if isinstance(params, RpcConnectionProperties):
self.rpc_connection_properties = params
else:
self.rpc_connection_properties = RpcConnectionProperties(
host=params.get('host'),
port=5672,
username=params.get('username'),
password=params.get('password'),
vhost=params.get('vhost') if params.has_key('vhost') else '/'
)
self.amqp_url = self.rpc_connection_properties.get_kombu_transport_url('pyamqp')
self.reply_queue = KombuQueue('amq.rabbitmq.reply-to', exchange=default_exchange, routing_key='amq.rabbitmq.reply-to')
def call(self, exchange, key, msg, no_response=False, timeout=5):
connection = Connection(self.amqp_url)
if exchange is not None:
key = exchange + ':' + key
with producers_pool[connection].acquire(block=True) as producer:
consumer = producer.channel.Consumer(queues=[self.reply_queue], no_ack=True, auto_declare=True,
callbacks=[self._on_message], accept=['ujson'])
consumer.consume(no_ack=True)
producer.publish(msg,
serializer='ujson',
exchange=default_exchange,
routing_key=key,
reply_to='amq.rabbitmq.reply-to')
consumer.connection.drain_events()
res = self.future.get(block=True, timeout=timeout)
response = Response()
response.body = res
return res
def cast(self, exchange, key, msg):
connection = Connection(self.amqp_url)
if exchange is not None:
key = exchange + ':' + key
with producers_pool[connection].acquire(block=True) as producer:
producer.publish(msg,
serializer='ujson',
exchange=default_exchange,
routing_key=key)
def _on_message(self, body, message):
print body
self.future.put(body)
顺便说一句:很抱歉变量名称,这应该替换旧的 stomp rpc 客户端,所以我不得不保留名称以兼容