咱们来拆解下你的实现思路，整体方向是符合测试场景需求的，但有几个关键细节需要调整和注意，才能确保队列在连接关闭后完全被清理：

核心逻辑的合理性 #

你在每个消费者线程内独立创建连接、通道，并声明auto_delete=True的队列，这是符合RabbitMQ auto-delete机制的：当绑定到该队列的最后一个消费者断开连接时，队列会自动被删除。

不过要注意一个关键点：如果队列已经存在（比如之前的线程未正常退出导致队列残留），再次调用queue_declare不会修改已存在队列的属性——也就是说，如果旧队列没有设置auto_delete，新的声明不会把它改成auto-delete模式。

需要优化的细节 #

1. 添加优雅的连接关闭逻辑 #

你给线程设置了setDaemon(True)，这意味着主线程退出时，守护线程会被强制终止，RabbitMQ连接可能来不及正常断开，导致队列无法被及时删除。建议给线程类增加关闭方法，确保连接能优雅断开：

import pika
from threading import Thread

class ConsumerThread(Thread):
    def __init__(self, callback, queue):
        Thread.__init__(self)
        self.setDaemon(True)
        self.callback = callback
        self.queue = queue
        self.connection = None  # 保存连接引用，用于后续关闭
    
    def run(self):
        self.connection = pika.BlockingConnection(pika.ConnectionParameters(CONNECTION['address'], CONNECTION['port'], CONNECTION['vhost'], CONNECTION['credentials']))
        channel = self.connection.channel()
        # 先删除可能残留的旧队列，再创建新的auto-delete队列
        channel.queue_delete(queue=self.queue)
        channel.queue_declare(queue=self.queue, auto_delete=True)
        channel.basic_qos(prefetch_count=1)
        channel.basic_consume(self.callback, queue=self.queue)
        try:
            channel.start_consuming()
        except (pika.exceptions.ConnectionClosedByBroker, pika.exceptions.AMQPChannelError):
            # 捕获连接/通道关闭异常，避免报错
            pass

    def stop(self):
        # 优雅关闭连接
        if self.connection and self.connection.is_open:
            self.connection.close()

class Factory:
    def __init__(self):
        self.queue_init = "init.queue"
        self.queue_start = "start.queue"
        self.threads = [ConsumerThread(self.init_callback, self.queue_init), ConsumerThread(self.start_callback, self.queue_start)]
        for t in self.threads:
            t.start()
    
    def init_callback(self, ch, method, properties, body):
        # 记得添加消息确认，避免测试数据混乱
        ch.basic_ack(delivery_tag=method.delivery_tag)
        # doing something
    
    def start_callback(self, ch, method, properties, body):
        ch.basic_ack(delivery_tag=method.delivery_tag)
        # doing something
    
    def cleanup(self):
        # 测试结束时调用该方法，确保所有线程和连接被清理
        for t in self.threads:
            t.stop()
            t.join()

2. 确保队列是全新的auto-delete实例 #

在测试场景下，建议每次启动线程前先删除同名队列（如上面代码中的channel.queue_delete），这样能彻底避免之前的队列残留，保证每次创建的都是全新的auto_delete=True队列。

3. 补充消息确认逻辑 #

你的回调函数里没有添加消息确认（basic_ack），如果是测试场景，未确认的消息会被RabbitMQ重新投递，可能导致测试数据混乱，建议在回调末尾添加确认操作。

总结 #

你的实现思路是正确的，只要补充优雅关闭连接的逻辑、确保队列的全新性，并添加消息确认，就能完美满足测试场景下“连接关闭后队列不留存”的需求。

内容的提问来源于stack exchange，提问作者rbullain