I would like to read a large volume of messages from a RabbitMQ queue, but negatively acknowledge (NACK) them. Some other consumer will ACK them and do work later. I have 100,000 messages in a queue. The following Python code is my attempt to print information about all messages: # Script settings seen_message_ids = set() retrieved_number = 0 delete_all_messages = False ids_to_delete = ['da140a51-cdd9-f71d-ac5c-ec15766e6aa4'] def on_message(ch, method_frame, header_frame, body): # If all arguments other than `ch` are None, then the `consume` method timed out due to queue inactivity. # Stop listening to the queue. if method_frame is None and header_frame is None and body is None: ch.stop_consuming() return # Messages that are not deleted are re-queued, meaning they will be consumed again, infinitely. Keep track # of `message_id`, which is a UUID to identify message that were already seen. Since this is a FIFO queue, # seeing any message again means all messages in the queue have been examined. message_id = header_frame.message_id if message_id in seen_message_ids: print("All messages evaluated.") ch.stop_consuming() return seen_message_ids.add(message_id) # Keep track of the number of retrieved messages. global retrieved_number retrieved_number += 1 # To keep the message, negatively acknowledge it and re-queue it. ch.basic_nack(delivery_tag=method_frame.delivery_tag, requeue=True) status = "READ " print(f"{status} {retrieved_number:06d} Message: {body.decode()} UUID: {header_frame.message_id}") return def main(): connection = pika.BlockingConnection( pika.ConnectionParameters(host=RABBITMQ_HOST, port=RABBITMQ_PORT, virtual_host=RABBITMQ_VHOST, credentials=credentials)) channel = connection.channel() channel.queue_declare(queue='test1') # Consume messages and acknowledge them print("START LISTENING") last_message_received_time = datetime.datetime.now() for method_frame, properties, body in channel.consume('test1', inactivity_timeout=1): # `consume` is a generator https://pika.readthedocs.io/en/stable/examples/blocking_consumer_generator.html # That means that the body of this for loop will not be reached unless a new message is found. new_message_received_time = datetime.datetime.now() if method_frame is None and properties is None and body is None: print(f"Timeout due to inactivity after {new_message_received_time - last_message_received_time} elapsed.") on_message(channel, method_frame, properties, body) try: channel.start_consuming() except KeyboardInterrupt: channel.stop_consuming() finally: print("STOP LISTENING") connection.close() This approach succeeds until about 5,000 messages have been retrieved. Then this happens: The ~5,000th message is printed: READ 005112 Message: {"job_id": "AAAAAAAAAAAAA"} UUID: 86d26605-33d6-9c61-68f8-a67abdf76a13 This message is printed: All messages evaluated. For this condition to be met, the script must have encountered a previously sent message_id, which is a Guid. Since the guids are distinct, the only way to see a repeated guid is for a re-queued message to be retrieved again. Shouldn't re-queued messages go to the back of the queue, though? I expected to see re-queued messages only after the 100,000 messages already in the queue were retrieved. The RabbitMQ management UI shows most messages as Unacked. The number of unacked messages goes to zero over several minutes, or sometimes a few seconds. The consumer remains stuck on ch.stop_consuming(), which is immediately after All messages evaluated until the management UI shows all messages as Ready. An attempt to retrieve messages using a second consumer fails to find any. An attempt to retrieve messages using the management UI fails to find any. Why does the consumer not receive and print all 100,000 messages? What work-arounds are available? Someone will probably ask why I want to read an NACK all messages. In fact, I want to ACK one or a few as a way to remove it from the queue. However, I can't ACK a message if my consumer never retrieves it. References to this approach for removing a message are here: SO1 and here: SO2 Continue reading...
