[rabbitmq-discuss] RabbitMQ + amqplib (python)

[David Wragg]

Hi Jared,

Jared Smith <jaredtsmith at gmail.com> writes:
> Everything works good for the first pass where I send 100k
> request/response pairs and I'm getting on the order of 4-5k rpc calls
> per second.  To make amqplib have blocking semantics I registered a
> callback with basic_consume and have that deposit messages that are
> read into a blocking queue which the client thread reads from this; it
> seems to work ok as far as I can tell.
>
> Problem I'm seeing is that on the second run of the program I get the
> following errors:
>
> ==============client error=================
[...]
> u'PRECONDITION_FAILED - timeout waiting for channel.flow_ok{active=false}',
> (0, 0), '')
> =======================================
>
> ============== matching server error ================
[...]
> amqplib.client_0_8.exceptions.AMQPConnectionException: (503,
> u'COMMAND_INVALID - basic.publish received after
> channel.flow_ok{active=false}', (60, 40), 'Channel.basic_publish')
> ==============================================

It's hard to be sure from the fragments of code you posted, but I wonder
if the issue is that you aren't acknowledging the messages?  To do this,
you either need to call basic_consume with no_ack=True (to tell rabbit
not to expect acknowledgements), or call basic_ack in your consuming
code to explicitly ack the messages.

If you don't acknowledge the messages, then they will stay in the queue,
and take up memory, eventually resulting in the channel.flow methods
indicated above.

You could confirm that this is the issue by doing a

    $ sudo rabbitmqctl list_queues

This will list the queues, along with the count of messages in each
queue.  If you do this after your first run, and see that a lot of
messages are still present, then it is very likely to be the lack of
acks causing the problem.

David

-- 
David Wragg
Staff Engineer, RabbitMQ
SpringSource, a division of VMware