调用多个函数作为 RABBITMQ 消息

Question

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

我一直在尝试运行这个 [tutorial] (https://www.rabbitmq.com/tutorials/tutorial-three-python.html) 的示例，它运行良好，.. 但我需要知道如何创建多个函数并通过 Rabbitmq Messages 调用它们？.. .（我也在用这个[例子]（Python and RabbitMQ - Best way to listen to consume events from multiple channels?）来指导我。）

我希望有人知道如何做到这一点...（我会再重复一遍，我对这个主题很陌生）...

这是我拥有的一些代码。

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

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()

Answer 1

如果您的函数是依赖的（我的意思是，一个函数作用于另一个函数的输出），那么您可以在回调函数中一一调用所有这些函数。成功执行最后一个函数后，您可以确认该消息。

如果这些函数是独立的，那么您可以为要在消息上执行的每个函数维护多个队列。可以使用fanout 交换将同一消息路由到多个队列，如RabbitMQ Tutorial-3 中所述。