Handling long running tasks in pika / RabbitMQ

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!