python - 使用 RabbitMQ 接收消息然后处理它然后发回结果

Question

我想从脚本(直接)发送消息，然后处理它，然后发回结果。 所以这就像双重发布-订阅。

我有 2 个脚本:

• 处理器
• 客户端

客户端直接向处理器发送消息(简单字符串)，然后处理器脚本对字符串中的字符进行计数并将结果发送回客户端。

这就是我尝试做的事情:

处理器等待消息，计算一些内容，然后回复原始发送者。

#Processer.py:
import pika
import sys

#Sends back the score
#addr: Connection address
#exchName: Exchange name (where to send)
#rKey: Name of the queue for direct messages
#score: The detected score
def SendActualScore(addr, exchName, rKey, score):
    #Send the image thru the created channel with the given routing key (queue name)
    channel.basic_publish(exchange=exchName, routing_key=rKey, body=score)
    print "(*) Sent: " + score

#When we receive something this is called
def CallbackImg(ch, method, properties, body):
    print "(*) Received: " + str(body)
    score = str(len(body))
    #Send back the score
    SendActualScore('localhost', 'valami', rKey, score)


#Subscribe connection
#Receive messages thru this
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
#RECEIVE MESSAGES - Subscribe
channel.exchange_declare(exchange='valami', type='direct')
#Define a queue, where we don't need the name
#After we disconnected delete the queue (exclusive flag)
result = channel.queue_declare(exclusive=True)
#We need the name of our temporary queue
queue_name = result.method.queue

rKeys = sys.argv[1:]
for rKey in rKeys:
    channel.queue_bind(exchange='valami', queue=queue_name, routing_key = rKey)

channel.basic_consume(CallbackImg, queue=queue_name, no_ack=True)
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()

客户端只需发送消息，然后等待答案。

#Client.py:
import pika
import sys

connAddr = 'localhost'

#Establish connection
connection = pika.BlockingConnection(pika.ConnectionParameters(connAddr))
channel = connection.channel()

#Define an exchange channel, we don't need a queue
channel.exchange_declare(exchange='valami', type='direct')

#Send the image thru the created channel
channel.basic_publish(exchange='valami', routing_key='msg', body='Message in the body')

print "[*] Sent"

def Callback(ch, method, properties, body):
    print "(*) Received: " + str(body)

result = channel.queue_declare(exclusive=True)
#We need the name of our temporary queue
queue_name = result.method.queue

channel.queue_bind(exchange='valami', queue=queue_name)

channel.basic_consume(Callback, queue=queue_name, no_ack=True)
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()

可能有多个客户端，我不知道如何将消息直接发回给他们。

Answer 1

您检查过 RabbitMQ w/python 和 pika 中的 RPC 教程吗？ http://www.rabbitmq.com/tutorials/tutorial-six-python.html

<小时/>

您需要在客户端执行的操作要点可在 RPC 教程中找到，但需要进行一些修改。

在您的客户端中，您将需要创建一个独占队列 - 与您在服务器中所做的方式相同。

当您从客户端发送消息时，需要将reply_to设置为客户端独占队列的名称

来自教程:

channel.basic_publish(exchange='',
                      routing_key='rpc_queue',
                      properties=pika.BasicProperties(
                            reply_to = callback_queue,
                            ),
                      body=request)

在服务器上，当您收到消息时，您需要从消息中读取 reply_to header ，然后basic_publish 将回复发送到该队列。

<小时/>

与其考虑“客户端”和“服务器”，不如用“消息生产者”和“消息消费者”来构建它可能会有所帮助。

在您的场景中，您需要两个进程既是发布者又是消费者。 “客户端”将发布原始消息并使用响应。 “服务器”将使用原始消息并发布响应。

代码中唯一真正的区别是在原始消息上使用 reply_to header 。这是您应将响应发布到的队列的名称。

希望有帮助!

<小时/>

附注我在我的 RabbitMQ Patterns 中介绍了这一点的核心概述。电子书 - RPC 和请求/回复，就像您需要的那样。这本书讲的是原理和模式，而不是具体的编程语言(尽管我主要写node.js，不太懂python)。