Share a single connection across all RabbitMQ consumers

[jholaj]

Fixes #649.

Every consumer opens its own connection (one per Q, plus one for its DQ), so a process consuming many queues holds a lot of mostly idle connections - a few hundred queues means several hundred connections per worker process.

With this change all consumers share one connection. pika isn't thread safe, so a small I/O thread owns it - deliveries go into per-consumer buffers, acks go back through add_callback_threadsafe (same as before). Each consumer still has its own channel, so prefetch works the same.

Only real difference: if the connection dies, all consumers of the process restart at once instead of one by one.

Existing RabbitMQ tests pass unchanged (cpython + gevent, RabbitMQ 4.2.9), and I added a test that consumers really do share one connection now.

Note

While measuring this I found a second source of connections: consume() declares the queue through the calling thread's publisher connection, and since consumers are created from consumer threads, each of them also kept an
idle publisher connection around forever. consume() now drops that connection after the declare - test_rabbitmq_broker_connections_are_lazy is updated accordingly (it asserts even lazier behavior than before).

Happy to rework any part of this or just talk it through - I use dramatiq daily now and like the project a lot, so I'm glad to help however fits.

[LincolnPuzey]

Hi, thanks for the PR. I haven't looked at the code yet.

I have seen pika has recently added ThreadSafeConnection which sounds like it works in a similar manner to your PR, in regards to running the connection in its own background thread.

Perhaps we should wait for that to be released, and update Dramatiq to use that, rather than implementing the same thing ourselves.

[LincolnPuzey]

consume() declares the queue through the calling thread's publisher connection, and since consumers are created from consumer threads, each of them also kept an
idle publisher connection around forever.

I think the ConsumerThread uses its publisher connection (on the Broker) to publish delayed messages. See here. So I don't think that connection is idle forever.

[jholaj]

Hello, sorry for the slow reply, I haven't had time lately.

Agreed - it's the same mechanism as this PR and better solved directly in pika. It won't cover everything though: reconnection and the consumer lifecycle stay in dramatiq either way, so a rewrite would drop the plumbing but keep most of the logic. It'd also mean bumping the pika floor from 1.0.

I think the ConsumerThread uses its publisher connection (on the Broker) to publish delayed messages. See here. So I don't think that connection is idle forever.

You're right that delay-queue consumer threads publish when handle_delayed_messages re-enqueues, so "idle forever" was an overstatement from me. The behavior is still correct though: the connection property is lazy, so the first re-enqueue simply reopens a connection, which is then kept and reused. Only the plain-queue consumer threads, which never publish, stop holding one forever. I'll fix the comment. 👍

Fine by me to wait - if you agree, I'll update this PR once it's released. :)