如何关闭连接到扇出交换的 Kombu ConsumerMixin 队列，以便在我的消费者不活动时不会从发布者累积数据？
我在 Python 2.7 中使用 Kombu 3.0.24（带有 RabbitMQ）。
下面是我的两个类的代码。我希望这些是相当通用的类，因此我可以将它们重用于直接队列和类似 RPC 的查询/回复。
问题是，如果我停止并重新启动消费者，旧数据会在消费者队列中等待我。我认为这是因为我需要在停止消费者时删除队列，但我不知道如何。谢谢。
消息消费者.py
from kombu.mixins import ConsumerMixin
from kombu import Queue, Exchange, Connection
import logging
class MessageConsumer(ConsumerMixin):
def __init__(self,
broker='amqp://',
exchange='mExchange',
queue = 'mQueue',
type='direct',
no_ack=False):
self.connection = Connection(broker)
self.mExchange = Exchange(exchange, type=type)
self.mQueue = Queue(queue, self.mExchange)
self.mQueue.no_ack = no_ack
def get_consumers(self, Consumer, channel):
return [Consumer(queues=self.mQueue,
accept=['json'],
callbacks=[self.process_task])]
def process_task(self, body, message):
logging.debug('RECEIVED: {}'.format(body))
def stop(self):
self.should_stop = True
self.connection.release()
if __name__ == '__main__':
mMessageConsumer = MessageConsumer(exchange='sensor_data',
queue='rx1_queue',
type='fanout',
no_ack=True)
try:
mMessageConsumer.run()
except KeyboardInterrupt:
mMessageConsumer.stop()
消息发布者.py
from kombu import Queue, Exchange, Connection
from kombu.pools import producers
import logging
class MessagePublisher(object):
def __init__(self,
broker='amqp://',
exchange='mExchange',
type='direct',
no_ack=False):
self.connection = Connection(broker)
self.mExchange = Exchange(exchange, type=type)
def publish(self, message, serializer='json', compression=None):
with producers[self.connection].acquire(block=True) as producer:
producer.publish(message,
serializer=serializer,
compression=compression,
exchange=self.mExchange,
declare=[self.mExchange]
)
def close(self):
self.connection.release()
if __name__ == '__main__':
mMessagePublisher = MessagePublisher(type='fanout',exchange='sensor_data')
x=0
while True:
x += 1
mMessagePublisher.publish(x)
mMessagePublisher.close()
如果我有更有效的编码方式，请提出建议。我通过谷歌搜索找到的大多数示例都使用旧版本的 Kombu，因此我很难找出 3.0.24 的最佳实现。