如何使用RabbitMQ和Python实现广播消息

使用 RabbitMQ 和 Python 实现广播消息的过程涉及设置一个消息队列和多个消费者，以便接收相同的消息。RabbitMQ 的 “fanout” 交换机允许你将消息广播到所有绑定的队列。以下是如何实现这一过程的详细步骤。

在这里插入图片描述

1、问题背景

在将系统从Morbid迁移到RabbitMQ时，发现RabbitMQ无法提供Morbid默认提供的广播行为。在广播模式下，当一个消息被添加到队列时，所有的消费者都会收到它。然而，在RabbitMQ中，消息会以轮询的方式分发给各个监听器。

代码例子如下：

# 消费者
import stompclass MyListener(object):def on_error(self, headers, message):print 'recieved an error %s' % messagedef on_message(self, headers, message):print 'recieved a message %s' % messageconn = 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 stompclass MyListener(object):def on_error(self, headers, message):print 'recieved an error %s' % messagedef on_message(self, headers, message):print 'recieved a message %s' % messageconn = 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()

通过上述代码，将会出现问题，导致无法实现广播消息。

2、解决方案

使用交换机和队列来实现广播消息。具体方法如下：

（1）使用amqplib库来创建交换机和队列。在发送消息时，将消息发送到交换机，而不是队列。在接收消息时，将队列绑定到交换机，这样就可以收到交换机上所有消息。代码修改如下：

# 发送者
import stompclass MyListener(object):def on_error(self, headers, message):print 'recieved an error %s' % messagedef on_message(self, headers, message):print 'recieved a message %s' % messageconn = 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 = {}# 发送消息到交换机
exchange = 'my_exchange'
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' % messagedef on_message(self, headers, message):print 'recieved a message %s' % messageconn = 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()

（2）使用StompJS 库来实现广播消息。具体方法如下：

// 消费者
var stompClient = Stomp.client('ws://localhost:61613');stompClient.connect({}, function(frame) {stompClient.subscribe('/topic/demoqueue', function(message) {console.log('Received message: ' + message.body);});
});// 发送者
var stompClient = Stomp.client('ws://localhost:61613');stompClient.connect({}, function(frame) {stompClient.send('/topic/demoqueue', {}, 'Hello, world!');
});