Effectuer des tâches de longue durée à pika / RabbitMQ

nous essayons de mettre en place un système de base de file d'attente dirigée où un producteur générera plusieurs tâches et un ou plusieurs consommateurs saisiront une tâche à la fois, la traiteront et reconnaîtront le message.

le problème est que le traitement peut prendre de 10 à 20 minutes, et nous ne répondons pas aux messages à ce moment-là, ce qui provoque la déconnexion du serveur.

voici un pseudo code pour notre consommateur:

#!/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()

une fois la première tâche terminée, une exception est jeté quelque part à l'intérieur de la connexion de blocage, se plaindre que la prise a été réinitialisée. De plus, les registres de RabbitMQ montrent que le consommateur a été déconnecté parce qu'il n'a pas répondu à temps (pourquoi il réinitialise la connexion plutôt que d'envoyer un aileron est étrange, mais nous ne nous en soucierons pas).

nous avons beaucoup cherché parce que nous pensions que C'était le cas normal D'utilisation de RabbitMQ (ayant beaucoup de tâches de longue durée qui devraient être réparties entre de nombreux consommateurs), mais il semble que personne d'autre n'avait vraiment ce problème. Enfin, nous sommes tombés sur un fil où il était recommandé d'utiliser les battements de coeur et de frayer le long_running_task() dans un thread séparé.

alors le code est devenu:

#!/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()

Et cela semble fonctionner, mais c'est très salissant. Sommes-nous sûrs que l' ch l'objet est-il sûr? En outre, imaginez que long_running_task() utilise ce paramètre de connexion pour ajouter une tâche à une nouvelle file d'attente (i.e. la première partie de ce long processus est terminée, envoyons la tâche sur pour la deuxième partie). Donc, le thread utilise le connection objet. C'est que "thread-safe"?

plus précisément, quelle est la meilleure façon de procéder? J'ai l'impression que c'est très bordélique et peut-être pas sûr du fil, donc peut-être qu'on ne le fait pas bien. Merci!