В настоящее время я работаю с RabbitMQ в Python, используя клиент Pika для создания сервера, который обрабатывает различные типы сообщений. Базовая установка, которую я имею, — это одна очередь, получающая все входящие сообщения, процесс маршрутизации, который направляет их в правильные места назначения, и несколько процессов для обработки запросов и приема входящих данных. Эта установка работала нормально, за исключением одного конкретного случая. Когда у меня работает сервер RabbitMQ до запуска серверных процессов и он получает сообщение, он правильно сохраняет их в очереди входящих сообщений. Однако, когда я затем пытаюсь запустить эти процессы и настроить потребителя на эту непустую входящую очередь с помощью функции pika.basic_consume, программа зависает. Итак, на данный момент, если я хочу запустить свои серверные процессы, я должен очистить все сообщения из очередей, прежде чем он заработает правильно. Как исправить это для работы с непустыми очередями?
Вот пример одного из процессов, все они настроены практически так же, как и этот.
class Router(Process):
def __init__(self,routing_table):
super(Router,self).__init__()
self.routing_table = routing_table
self.routeQueues = {
'r' : 'registration',
't' : 'util',
'p' : 'util',
's' : 'data'
}
# Create a connection to the RabbitMQ server.
self.rabbitConn = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
self.channel = self.rabbitConn.channel()
# Load all of the existing registered node queues
with open('registrations/nodes.txt','r') as nodes:
for line in nodes:
info = line.strip().split(":")
self.channel.queue_declare(info[1])
# Declare the default queues
queue_list = ["incoming","registration","util"]
for queueName in queue_list:
self.channel.queue_declare(queueName)
# Start consuming things from the incoming queue
self.channel.basic_consume(self.gotPacket,queue='incoming')
def gotPacket(self,ch,method,params,body):
# Does stuff. Not relevant here.
pass
def run(self):
self.channel.start_consuming()
