Was ist der beste Weg, die Veröffentlichung von Nachrichten mit Kombu erneut zu versuchen?

Question

Ich teste gerade wie Kombu funktioniert. Ich plane, Pika in mehreren Projekten zu ersetzen. Ich sehe, dass Kombu eine Menge Dokumentation hat, aber mit dem, was ich in der Dokumentation gefunden habe, sind einige Nachrichten verloren gegangen. Hier ist der Code:

from kombu import Connection, Producer 
conn = Connection('amqp://localhost:5672') 
def errback(exc, interval): 
    logger.error('Error: %r', exc, exc_info=1) 
    logger.info('Retry in %s seconds.', interval) 
producer = Producer(conn) 
publish = conn.ensure(producer, producer.publish, errback=errback, max_retries=3) 
for i in range(1, 200000): 
    publish({'hello': 'world'}, routing_key='test_queue') 
    time.sleep(0.001) 

Wenn es Publishing Ich schließe die Verbindung mehrmals und es hält Publishing aber in der Warteschlange gibt es rund 60.000 Nachrichten, so gibt es eine Menge von verlorenen Nachrichten.

Ich habe versucht, verschiedene Alternativen z:

publish({'hello': 'world'}, retry=True, mandatory=True, routing_key='hipri') 

Dank!

Answer 1

Das Problem war, dass standardmäßig Kombu nicht 'bestätigen' nicht verwenden, müssen Sie verwenden:

 conn = Connection('amqp://localhost:5672', transport_options={'confirm_publish': True}) 

Dank