Stomp Broadcast с Rabbitmq и Python

Я пытаюсь перевести систему с использования 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()

python rabbitmq broadcast stomp
person Sean O Donnell    schedule 11.06.2009    source источник

Ответы (2)


arrow_upward
3
arrow_downward

По-видимому, вы не можете напрямую использовать STOMP; существует ветка списка рассылки, в которой показаны все обручи, которые вы должны перепрыгнуть, чтобы трансляция работала с stomp (это включает в себя некоторые вещи AMPQ более низкого уровня).

person Rick Copeland    schedule 11.06.2009
comment
Спасибо, я видел этот поток раньше и безуспешно пытался реализовать его предложения с помощью amqplib. Конкретное сообщение, которое касается этого, называется lists.rabbitmq.com/ pipermail/rabbitmq-discuss/2008-сентябрь/. Я обновил вопрос, чтобы отразить, что образец amqplib сделает эту работу за меня. - person Sean O Donnell; 11.06.2009

arrow_upward
3
arrow_downward

Наконец-то я понял, как это сделать, создав обмен для каждой «получающей группы», я не уверен, насколько хорошо кролик справится с тысячами обменов, поэтому вы можете тщательно протестировать это, прежде чем пробовать его в производстве.

В коде отправки:

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()
person Sean O Donnell    schedule 12.06.2009