Вызов более чем одной функции как сообщение RABBITMQ - PullRequest
Новая
       84

Вызов более чем одной функции как сообщение RABBITMQ

0 голосов
/ 19 февраля 2020

Я только начинаю использовать RabbitMQ и Python в целом. Я читал статьи на официальной странице кролика, но не знаю, как использовать Rabbitmq для других целей.

Я пытался запустить пример этого [учебника] (https://www.rabbitmq.com/tutorials/tutorial-three-python.html), он работает хорошо, НО мне нужно знать, как я могу создать более одной функции и называть их через сообщения Rabbitmq? ... (я также использую этот [пример] (Python и RabbitMQ - лучший способ прослушивания событий потребления из нескольких каналов? ), чтобы направлять меня.)

Надеюсь, у кого-то есть идея, как это сделать ... (Повторюсь еще раз, я очень новичок в этом вопросе) ...

Это какой-то код, который у меня есть.

Я использую этот код для отправки сообщения в качестве учебника .. 

import pika
import sys

url = 'amqp://oamogcgg:xxxxxxxxxxxxxxxxxxxxxxxxx@salamander.rmq.cloudamqp.com/oamogcgg'
params = pika.URLParameters(url)
connection = pika.BlockingConnection(params)
channel = connection.channel()


channel.exchange_declare(exchange='logs', exchange_type='fanout')

message = ' '.join(sys.argv[1:]) or "info: THIS IS A TEST MESSAGE !!!!!!"
channel.basic_publish(exchange='logs', routing_key='', body=message)
print(" [x] Sent %r" % message)
connection.close()

И в этом файле я получаю сообщение, соответствующее мне. 

import pika
import sys
import threading

threads=[]


#function 1 

def validator1(channel):    
channel.queue_declare(queue='queue_name')
print (' [*] Waiting messsaes for valiadtor1 press CTRL+C')


def callback(ch, method, properties, body):
   print (" Received %s" % (body))
   sleep(2) #I need stop it for two minutes

channel.basic_consume(callback, queue='queue_name', no_ack=True)
channel.start_consuming()

#function 2

def validator2(channel):    
channel.queue_declare(queue='queue_name')
print (' [*] Waiting messsaes for valiadtor2 press CTRL+C')


def callback(ch, method, properties, body):
  print (" Received %s" % (body))
  sleep(2) #I need stop it for two minutes

channel.basic_consume(callback, queue='queue_name', no_ack=True)
channel.start_consuming()



def manager():
 url = 'amqp://oamogcgg:xxxxxxxxxxxxxxxxxxxxxxxxx@salamander.rmq.cloudamqp.com/oamogcgg'
 params = pika.URLParameters(url)

#channel 1
 connection1= pika.BlockingConnection(params)
 channel1 = connection1.channel()

 channel1.exchange_declare(exchange='logs', exchange_type='fanout')
 result = channel1.queue_declare(queue='', exclusive=True)
 queue_name = result.method.queue
 channel1.queue_bind(exchange='logs', queue=queue_name)

#channel 2

 connection2= pika.BlockingConnection(params)
 channel2 = connection2.channel()


channel2.exchange_declare(exchange='logs', exchange_type='fanout')
result = channel2.queue_declare(queue='', exclusive=True)
queue_name = result.method.queue
channel2.queue_bind(exchange='logs', queue=queue_name)



#creating threads

t1 = threading.Thread(target=validator1, args=(channel1,))
t1.daemon = True
threads.append(t1)
t1.start()  

t2 = threading.Thread(target=valiadtor2, args=(channel2,))
t2.daemon = True
threads.append(t2)


t2.start()
for t in threads:
    t.join()


manager()

1 Ответ

0 голосов
/ 21 февраля 2020

Если ваши функции являются зависимыми (я имею в виду, зависимая, одна функция работает на выходе другой), то вы можете вызывать все эти функции одну за другой в своей функции обратного вызова. Как только последняя функция выполнена успешно, вы можете подтвердить сообщение.

Если эти функции независимы, вы можете поддерживать несколько очередей для каждой функции, которую вы хотите выполнить в сообщении. Одно и то же сообщение можно направить в несколько очередей с использованием обмена fanout, как указано в RabbitMQ Tutorial-3 .

...