I'd like to make sure that my message was delivered to a queue.
To do so I'm adding the mandatory param to the basic_publish.
What else should I do to receive the basic.return
message if my message wasn't successfully delivered?
I can't use channel.wait()
to listen for the basic.return
because when my message is succe开发者_Go百科ssfully delivered the wait()
function hangs forever. (There is no timeout)
On the other hand. When I don't call channel.wait()
the channel.returned_messages
will remain empty, even if the message isn't delivered.
I use py-amqplib
version 0.6.
Any solution is welcome.
It is currently impossible as the basic.return
is sent asynchronously when a message is dropped in broker. When message was sent successfully no data is reported from server.
So pyAMQP can't listen for such messages.
I've read few threads about this problem. Possible solution were:
- use txAMQP, twisted version of amqp that handles basic.return
- use pyAMQP with wait with timeout. (I'm not sure if that is currently possible)
- ping server frequently with synchronous commands so that pyAMQP will able to pick
basic.return
messages when they arrive.
Because the level of support for pyAMQP and rabbitMQ in general is quite low, we decided not to use amqp broker at all.
Have you tried the only Python AMQP library that is complete? It isn't as widely used because it is not neatly packaged.
Step 1. compile the C library - you may need sudo apt-get install autotools-dev autoconf automake libtool
mkdir rabbitc
cd rabbitc
hg clone http://hg.rabbitmq.com/rabbitmq-codegen/
hg clone http://hg.rabbitmq.com/rabbitmq-c/
cd rabbitmq-c
autoreconf -i
make clean
./configure --prefix=/usr
make
sudo make install
Step 2. Install the Python library
pip install pylibrabbitmq
You can't do this synchronously as it is an asynchronous system. But you can solve this problem using threads.
The basic idea is that you start a thread which does the wait on the channel, whenever it comes out of the wait it calls the call_back function for any returned message in the returned message queue. You can then deal with that message however you want to in the call_back function
def registerCallback(channel, call_back): """ This method sets up a thread which deals with the asynchronous callback for a message which could not be routed by the exchange. """ def wait(): try: channel.wait() except Exception, e: print("Problem waiting on publish channel: %s" % str(e)) while not channel.returned_messages.empty(): returnedMessage = channel.returned_messages.get() processReturnedMessageThread = Thread(target=call_back, args=(returnedMessage)) processReturnedMessageThread.start() wait() waiting = Thread(target=wait) waiting.start()
精彩评论