2

我试图将系统从使用 morbid 转移到 rabbitmq，但我似乎无法获得默认提供的相同广播行为 morbid。通过广播，我的意思是当一条消息被添加到队列中时，每个消费者都会收到它。使用rabbit，当添加消息时，它们会以循环方式分发给每个听众。

谁能告诉我如何实现相同的消息分发？

下面使用的 stomp 库是http://code.google.com/p/stomppy/

如果无法使用 stomp，即使是 amqplib 示例也会有帮助。

我目前的代码看起来像这样

消费者

import stomp

class MyListener(object):
    def on_error(self, headers, message):
        print 'recieved an error %s' % message

    def on_message(self, headers, message):
        print 'recieved a message %s' % message

conn = stomp.Connection([('0.0.0.0', 61613), ('127.0.0.1', 61613)], 'user', 'password')
conn.set_listener('', MyListener())
conn.start()
conn.connect(username="user", password="password")
headers = {}

conn.subscribe(destination='/topic/demoqueue', ack='auto')

while True:
    pass
conn.disconnect()

发件人看起来像这样

import stomp

class MyListener(object):
    def on_error(self, headers, message):
        print 'recieved an error %s' % message

    def on_message(self, headers, message):
        print 'recieved a message %s' % message

conn = stomp.Connection([('0.0.0.0', 61613), ('127.0.0.1', 61613)], 'user', 'password')
conn.set_listener('', MyListener())
conn.start()
conn.connect(username="user", password="password")
headers = {}

conn.subscribe(destination='/topic/demotopic', ack='auto')

while True:
    pass
conn.disconnect()
4

2 回答 2

3

显然你不能直接使用 STOMP；有一个邮件列表线程显示了您必须跳过的所有箍才能让广播与 stomp 一起工作（它涉及一些较低级别的 AMPQ 内容）。

于 2009-06-11T14:40:35.820 回答
3

我终于弄清楚了如何通过为每个“接收组”创建一个交换来做到这一点，我不确定兔子在数千个交换中的表现如何，所以你可能想在生产中尝试之前对其进行大量测试

在发送代码中：

conn.send(str(i), exchange=exchange, destination='')

需要空白目的地，我只关心发送到该交易所

收到

import stomp
import sys
from amqplib import client_0_8 as amqp
#read in the exchange name so I can set up multiple recievers for different exchanges to tset
exchange = sys.argv[1]
conn = amqp.Connection(host="localhost:5672", userid="username", password="password",
 virtual_host="/", insist=False)

chan = conn.channel()

chan.access_request('/', active=True, write=True, read=True)

#declare my exchange
chan.exchange_declare(exchange, 'topic')
#not passing a queue name means I get a new unique one back
qname,_,_ = chan.queue_declare()
#bind the queue to the exchange
chan.queue_bind(qname, exchange=exchange)

class MyListener(object):
    def on_error(self, headers, message):
        print 'recieved an error %s' % message

    def on_message(self, headers, message):
        print 'recieved a message %s' % message

conn = stomp.Connection([('0.0.0.0', 61613), ('127.0.0.1', 61613)], 'browser', 'browser')
conn.set_listener('', MyListener())
conn.start()
conn.connect(username="username", password="password")
headers = {}

#subscribe to the queue
conn.subscribe(destination=qname, ack='auto')

while True:
    pass
conn.disconnect()
于 2009-06-12T14:32:06.223 回答