Handling long running tasks in pika / RabbitMQ
Question: We're trying to set up a basic directed queue system where a producer will
generate several tasks and one or more consumers will grab a task at a time,
process it, and acknowledge the message.

The problem is, the processing can take 10-20 minutes, and we're not
responding to messages at that time, causing the server to disconnect us.

Here's some pseudo code for our consumer:

    
    
    #!/usr/bin/env python
    import pika
    import time
    
    connection = pika.BlockingConnection(pika.ConnectionParameters(
            host='localhost'))
    channel = connection.channel()
    
    channel.queue_declare(queue='task_queue', durable=True)
    print ' [*] Waiting for messages. To exit press CTRL+C'
    
    def callback(ch, method, properties, body):
        long_running_task(connection)
        ch.basic_ack(delivery_tag = method.delivery_tag)
    
    channel.basic_qos(prefetch_count=1)
    channel.basic_consume(callback,
                          queue='task_queue')
    
    channel.start_consuming()
    

After the first task completes, an exception is thrown somewhere deep inside
of BlockingConnection, complaining that the socket was reset. In addition, the
RabbitMQ logs show that the consumer was disconnected for not responding in
time (why it resets the connection rather than sending a FIN is strange, but
we won't worry about that).

We searched around a lot because we believed this was the normal use case for
RabbitMQ (having a lot of long running tasks that should be split up among
many consumers), but it seems like nobody else really had this issue. Finally
we stumbled upon a thread where it was recommended to use heartbeats and to
spawn the `long_running_task()` in a separate thread.

So the code has become:

    
    
    #!/usr/bin/env python
    import pika
    import time
    import threading
    
    connection = pika.BlockingConnection(pika.ConnectionParameters(
            host='localhost',
            heartbeat_interval=20))
    channel = connection.channel()
    
    channel.queue_declare(queue='task_queue', durable=True)
    print ' [*] Waiting for messages. To exit press CTRL+C'
    
    def thread_func(ch, method, body):
        long_running_task(connection)
        ch.basic_ack(delivery_tag = method.delivery_tag)
    
    def callback(ch, method, properties, body):
        threading.Thread(target=thread_func, args=(ch, method, body)).start()
    
    channel.basic_qos(prefetch_count=1)
    channel.basic_consume(callback,
                          queue='task_queue')
    
    channel.start_consuming()
    

And this seems to work, but it's very messy. Are we sure that the `ch` object
is thread safe? In addition, imagine that `long_running_task()` is using that
connection parameter to add a task to a new queue (i.e. the first part of this
long process is done, let's send the task on to the second part). So, the
thread is using the `connection` object. Is that thread safe?

More to the point, what's the preferred way of doing this? I feel like this is
very messy and possibly not thread safe, so maybe we're not doing it right.
Thanks!


Answer: For now, your best bet is to turn off heartbeats, this will keep RabbitMQ from
closing the connection if you're blocking for too long. I am experimenting
with pika's core connection management and IO loop running in a background
thread but it's not stable enough to release.

