Close
Glad You're Ready. Let's Get Started!

Let us know how we can contact you.

Thank you!

We'll respond shortly.

LABS
RabbitMQ, AMQP gem, and EventMachine

I recently had a chance to work with RabbitMQ and the AMQP gem.

The first problem we ran into with subscribing to a queue was if we forcefully kill a client, a large number of messages would disappear from the queue (far more than we had processed on the client). I’ll get to the “why” of this in a minute, but the solution was simply to turn on acknowledgments (or acks) for our messages which was something we knew we wanted to do anyway.

So after we turned on acks and started processing items from the queue we noticed that the number of items in the queue was not actually going down even though we were correctly sending the ack when we were done processing.

The AMQP gem uses EventMachine under the covers to manage the connections to the RabbitMQ server. It turned out that when you subscribe to a queue, it is a one time thing. You establish a connection and that is it. The server then sends you messages from that queue over the socket. RabbitMQ pre-fetches messages for you, meaning it crams a bunch of data over the socket and doesn’t wait for you to ask for more, it notices that you’ve read data off the socket, and pushes more to you.

The repercussions of this in the EventMachine world is a major blockage of data. EventMachine has an internal loop where it goes through registered sockets, and processes all the data off any sockets that are ready to be read from before it continues its loop. The server was basically keeping the socket full, so EventMachine would only complete a internal loop after we processed a full socket of data, and then it would get blocked on the next loop since the server has already filled the socket up again.

This means that all of our acks were sitting inside of EventMachine waiting for the loop to continue so they could be sent out. It also explains why when we weren’t using acks we were losing messages. The server had sent them to our socket and they were waiting to be processed and by killing the process we lost that data.

My first reaction was that the AMQP gem should be pulling all the data off the socket and caching it locally, then processing a single record off of that cache every time the EventMachine loop ran. This of course won’t work because as soon as we empty the socket, RabbitMQ is just going to fill it up again (until we have all the messages from the queue in our local cache).

So the solution? RabbitMQ 1.6 has an option to set a pre-fetch limit. So we simply set the pre-fetch limit to 1, and our EventMachine loop runs nice and fast now. You’ll want to tweak your pre-fetch limit depending on how long it takes to process each message. If you can churn through a hundred messages a second, you probably won’t even notice this problem and the prefetching will help you, but if it takes you a few seconds (or minutes) per message, you’ll wonder why things aren’t popped off the queue for several minutes (or hours).

Comments
  1. Paolo Negri says:

    You might want to take a look at my rabbit_starter project on github.
    Is a project I started in the context of the talk I gave at http://www.rails-underground.com/ about RabbitMQ and EventMachine

    It provides various examples of RabbitMQ clients

    here’s the link to the project, suggestions comments and forks are really welcome.

    http://github.com/hungryblank/rabbit_starter/tree/master

  2. Matt Wynne says:

    We’ve had similar issues though we’re living with them for now – thanks for documenting your experiences.

    One thing I’ve not had a chance to check out yet are a couple of alternate libraries (carrot and bunny) which I believe were originally written as a reaction to this problem with the tmm1-amqp library.

    See http://wiki.github.com/celldee/bunny

Post a Comment

Your Information (Name required. Email address will not be displayed with comment.)

* Copy This Password *

* Type Or Paste Password Here *