[rabbitmq-discuss] Messaging system for the university

Tue Oct 19 06:19:56 BST 2010

Hi,
We are building a messaging system for our university.
Currently we have a rabbitmq cluster setup running on 2
nodes(CentOS-5.5x86 and CentOS-5.5x86_64). There are message consumers
connected to both nodes. The problem is that consumers receive
messages in a round-robin fashion (e.g. Message#1 is received by
consumer #1(not by the second consumer) and message#2 is received by
the second consumer(not by the first consumer)). We want all of our
consumers to receive all of the published messages.

Any thoughts on this?

[root at bv ~]# rabbitmqctl status
Status of node rabbit at bv ...
[{running_applications,[{rabbit,"RabbitMQ","2.0.0"},
                        {os_mon,"CPO  CXC 138 46","2.1.8"},
                        {sasl,"SASL  CXC 138 11","2.1.5.4"},
                        {mnesia,"MNESIA  CXC 138 12","4.4.7"},
                        {stdlib,"ERTS  CXC 138 10","1.15.5"},
                        {kernel,"ERTS  CXC 138 10","2.12.5"}]},
 {nodes,[{disc,[rabbit at bv]},{ram,[rabbit at dcbackup]}]},
 {running_nodes,[rabbit at dcbackup,rabbit at bv]}]
...done.

[root at dcbackup ~]# rabbitmqctl status
Status of node rabbit at dcbackup ...
[{running_applications,[{rabbit,"RabbitMQ","2.0.0"},
                        {mnesia,"MNESIA  CXC 138 12","4.4.7"},
                        {os_mon,"CPO  CXC 138 46","2.1.8"},
                        {sasl,"SASL  CXC 138 11","2.1.5.4"},
                        {stdlib,"ERTS  CXC 138 10","1.15.5"},
                        {kernel,"ERTS  CXC 138 10","2.12.5"}]},
 {nodes,[{disc,[rabbit at bv]},{ram,[rabbit at dcbackup]}]},
 {running_nodes,[rabbit at bv,rabbit at dcbackup]}]
...done.

#---- amqp_consumer.py on server `dcbackup' ------
from amqplib import client_0_8 as amqp

conn = amqp.Connection(host="dcbackup:5672", userid="guest",
password="guest", virtual_host="/", insist=False)
chan = conn.channel()

chan.queue_declare(queue="po_box", durable=True, exclusive=False,
auto_delete=False)
chan.exchange_declare(exchange="sorting_room", type="direct",
durable=True, auto_delete=False,)

chan.queue_bind(queue="po_box", exchange="sorting_room", routing_key="jason")

def recv_callback(msg):
    print 'Received: ' + msg.body + ' from channel #' +
str(msg.channel.channel_id)

chan.basic_consume(queue='po_box', no_ack=True,
callback=recv_callback, consumer_tag="testtag")
while True:
    chan.wait()
chan.basic_cancel("testtag")
chan.close()
conn.close()

#---- amqp_consumer.py on server `bv' ------
from amqplib import client_0_8 as amqp

conn = amqp.Connection(host="bv:5672", userid="guest",
password="guest", virtual_host="/", insist=False)
chan = conn.channel()

chan.queue_declare(queue="po_box", durable=True, exclusive=False,
auto_delete=False)
chan.exchange_declare(exchange="sorting_room", type="direct",
durable=True, auto_delete=False,)

chan.queue_bind(queue="po_box", exchange="sorting_room", routing_key="jason")

def recv_callback(msg):
    print 'Received: ' + msg.body + ' from channel #' +
str(msg.channel.channel_id)

chan.basic_consume(queue='po_box', no_ack=True,
callback=recv_callback, consumer_tag="testtag")
while True:
    chan.wait()
chan.basic_cancel("testtag")
chan.close()
conn.close()

#---- amqp_consumer.py on server `bv' ------
from amqplib import client_0_8 as amqp
import sys

conn = amqp.Connection(host="bv:5672", userid="guest",
password="guest", virtual_host="/", insist=False)
chan = conn.channel()

msg = amqp.Message(sys.argv[1])
msg.properties["delivery_mode"] = 1
chan.basic_publish(msg,exchange="sorting_room",routing_key="jason")

chan.close()
conn.close()

[root at bv ~]# uname -a
Linux bv.localdomain 2.6.18-194.el5xen #1 SMP Fri Apr 2 15:34:40 EDT
2010 x86_64 x86_64 x86_64 GNU/Linux

[root at dcbackup ~]# uname -a
Linux dcbackup.localdomain 2.6.18-194.el5xen #1 SMP Fri Apr 2 16:16:54
EDT 2010 i686 i686 i386 GNU/Linux

Our python version is:
Python 2.6.5 (r265:79063, Apr  9 2010, 11:16:46)
[GCC 4.1.2 20080704 (Red Hat 4.1.2-48)] on linux2