Comment faire pour vérifier en python rabbitmq si les messages ne sont pas livrés dans les 5 dernières secondes

Question

Bonjour, j'essaie d'écrire du code en python avec rabbitmq. J'ai une file d'attente où j'envoie des messages mais je dois vérifier sur le consommateur si le message a été envoyé dans les 5 dernières secondes et sinon je devrais terminer le processus. J'ai essayé de chercher sur Internet pour une telle fonction, mais il n'y a pas de réponse pertinente, pouvez-vous me suggérer quelque chose les gars?

Answer 1

RabbitMQ comprend un battement de coeur pour détecter les pairs qui ne répondent pas/ayant échoué

From the docs:

Detecting Mort TCP Connexions avec Heartbeats

Dans certains types de défaillance du réseau, la perte de paquets peut signifier que perturbé Les connexions TCP prennent un temps modérément long (environ 11 minutes avec configuration par défaut sous Linux, par exemple) pour être détectées par les opérations ystem. AMQP 0-9-1 offre une fonction de pulsation pour garantir que la couche d'application découvre rapidement les connexions perturbées (et également les homologues qui ne répondent pas du tout). Heartbeats également défendre contre certains équipements réseau qui peuvent mettre fin à "inactif" connexions TCP .

Pour activer Hearbeats avec Java Client:

ConnectionFactory cf = new ConnectionFactory(); 

// set the heartbeat timeout to 5 seconds 
cf.setRequestedHeartbeat(5); 

De même avec le client .NET:

var cf = new ConnectionFactory(); 

// set the heartbeat timeout to 5 seconds 
cf.RequestedHeartbeat = 5; 

Hope this helps.

(il y a plus sur dead-letter exchanges dans les RabbitMQ docs, aussi sur le NACK et ack/(nég/pos) la livraison/confirme on this page mais la configuration Heartbeats devrait faire l'affaire.)

EDIT: Désolé, il y a aussi un python remote procedure callback example dans les docs! Il faut 'pika' .. manqué ça!

code serveur Exemple:

#!/usr/bin/env python 
import pika 

connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost')) 

channel = connection.channel() 

channel.queue_declare(queue='rpc_queue') 

def fib(n): 
    if n == 0: 
     return 0 
    elif n == 1: 
     return 1 
    else: 
     return fib(n-1) + fib(n-2) 

def on_request(ch, method, props, body): 
    n = int(body) 

    print(" [.] fib(%s)" % n) 
    response = fib(n) 

    ch.basic_publish(exchange='', 
        routing_key=props.reply_to, 
        properties=pika.BasicProperties(correlation_id = \ 
                 props.correlation_id), 
        body=str(response)) 
    ch.basic_ack(delivery_tag = method.delivery_tag) 

channel.basic_qos(prefetch_count=1) 
channel.basic_consume(on_request, queue='rpc_queue') 

print(" [x] Awaiting RPC requests") 
channel.start_consuming() 

exemple Code client:

#!/usr/bin/env python 
import pika 
import uuid 

class FibonacciRpcClient(object): 
    def __init__(self): 
     self.connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost')) 

     self.channel = self.connection.channel() 

     result = self.channel.queue_declare(exclusive=True) 
     self.callback_queue = result.method.queue 

     self.channel.basic_consume(self.on_response, no_ack=True, 
            queue=self.callback_queue) 

    def on_response(self, ch, method, props, body): 
     if self.corr_id == props.correlation_id: 
      self.response = body 

    def call(self, n): 
     self.response = None 
     self.corr_id = str(uuid.uuid4()) 
     self.channel.basic_publish(exchange='', 
            routing_key='rpc_queue', 
            properties=pika.BasicProperties(
             reply_to = self.callback_queue, 
             correlation_id = self.corr_id, 
             ), 
            body=str(n)) 
     while self.response is None: 
      self.connection.process_data_events() 
     return int(self.response) 

fibonacci_rpc = FibonacciRpcClient() 

print(" [x] Requesting fib(30)") 
response = fibonacci_rpc.call(30) 
print(" [.] Got %r" % response)