调用多个功能作为RABBITMQ消息

Question

我刚开始一般使用RabbitMQ和Python。我一直在阅读Rabbit官方页面上的Tut，但是我不知道如何使用Rabbitmq做其他事情。

我一直在尝试运行此[tutorial]（https://www.rabbitmq.com/tutorials/tutorial-three-python.html）的示例，它运行良好，..但是我需要知道如何创建多个功能并通过Rabbitmq消息将其命名为...？ （我也正在使用此[示例]（Python and RabbitMQ - Best way to listen to consume events from multiple channels?）来指导我。）

我希望有人知道如何执行此操作...（我会再说一遍，在这个主题上我很新）...

这是我的一些代码。

我使用此代码作为教程发送消息。

import pika
import sys

url = 'amqp://oamogcgg:[email protected]/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:[email protected]/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()

我刚开始一般使用RabbitMQ和Python。我一直在阅读Rabbit官方页面上的Tut，但是我不知道如何使用Rabbitmq做其他事情。我一直在尝试...

Answer 1

如果您的函数是从属的（我指的是一个函数在另一个函数的输出上工作），那么您可以在回调函数中一一调用所有这些函数。成功执行最后一个功能后，您可以确认该消息。