RabbitMQ 中的许多客户端和许多服务器

Posted 2023-02-16

【中文标题】RabbitMQ 中的许多客户端和许多服务器

【英文标题】：Many clients and many servers in RabbitMQ

【发布时间】：2014-09-26 17:08:00

【问题描述】：

我读了this，你能解释一下吗？ 例如，我在终端的不同选项卡（3 个选项卡）中运行了 rpc_server.py。

rpc_server.py 来自该教程：

#!/usr/bin/env python
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='localhost'))

channel = connection.channel()

channel.queue_declare(queue='rpc_queue')

def fib(n):
    if n == 0:
        return 0
    elif n == 1:
        return 1
    else:
        return fib(n-1) + fib(n-2)

def on_request(ch, method, props, body):
    n = int(body)

    print " [.] fib(%s)"  % (n,)
    response = fib(n)

    ch.basic_publish(exchange='',
                     routing_key=props.reply_to,
                     properties=pika.BasicProperties(correlation_id = \
                                                     props.correlation_id),
                     body=str(response))
    ch.basic_ack(delivery_tag = method.delivery_tag)

channel.basic_qos(prefetch_count=1)
channel.basic_consume(on_request, queue='rpc_queue')

print " [x] Awaiting RPC requests"
channel.start_consuming()

很好，我需要发送 send.py 3 个请求：

#!/usr/bin/env python
import pika
import uuid

class FibonacciRpcClient(object):
    def __init__(self):
        self.connection = pika.BlockingConnection(pika.ConnectionParameters(
                host='localhost'))

        self.channel = self.connection.channel()

        result = self.channel.queue_declare(exclusive=True)
        self.callback_queue = result.method.queue

        self.channel.basic_consume(self.on_response, no_ack=True,
                                   queue=self.callback_queue)

    def on_response(self, ch, method, props, body):
        if self.corr_id == props.correlation_id:
            self.response = body

    def call(self, n):
        self.response = None
        self.corr_id = str(uuid.uuid4())
        self.channel.basic_publish(exchange='',
                                   routing_key='rpc_queue',
                                   properties=pika.BasicProperties(
                                         reply_to = self.callback_queue,
                                         correlation_id = self.corr_id,
                                         ),
                                   body=str(n))
        while self.response is None:
            self.connection.process_data_events()
        return int(self.response)

fibonacci_rpc = FibonacciRpcClient()

print " [x] Requesting fib(30)"
response = fibonacci_rpc.call(30)
print " [.] Got %r" % (response,)

fibonacci_rpc1 = FibonacciRpcClient()

print " [x] Requesting fib(30)"
response1 = fibonacci_rpc1.call(30)
print " [.] Got %r" % (response1,)

fibonacci_rpc2 = FibonacciRpcClient()

print " [x] Requesting fib(30)"
response2 = fibonacci_rpc2.call(30)
print " [.] Got %r" % (response2,)

这是否意味着脚本将等待第一个请求的响应，然后发送第二个请求，再次等待响应，然后发送第三个请求？

我想一次做 3 个请求，而不是等待响应然后发送新请求。这该怎么做？

我需要如何更改send.py 或使用其他技术？我必须使用线程还是多处理？ RabbitMQ 支持吗？

谢谢！

【问题讨论】：

老实说，我无法理解您在这里要做什么。您似乎在同一个对象中发布和消费，这完全违背了排队的目的。

例如。呼叫（30），呼叫（60），呼叫（90）

这种反应似乎完全无关紧要。排队系统的重点是您有（一个或多个）应用程序将内容放在队列中，以及一组完全独立的客户端订阅队列并处理消息。在同一个类和同一个过程中做这两个动作是完全没有意义的。

@DanielRoseman 你能举个例子吗，它必须是怎样的？

【参考方案1】：

如果您希望一次发送所有三个请求，则需要使用线程。一个简单的解决方案如下所示：

import threading
from time import sleep

def make_rpc_call(value):
    fibonacci_rpc = FibonacciRpcClient()
    print " [x] Requesting fib(0)".format(value)
    response = fibonacci_rpc.call(value)
    print " [.] Got %r" % (response,)

for index in xrange(5):
    thread_ = threading.Thread(target=make_rpc_call, args=(30, ))
    thread_.start()
    sleep(0.1)

请记住，Pika 不是线程安全的，因此您需要为每个线程创建一个连接。作为替代方案，您可以查看我的 Flask 示例here for Pika。它很容易修改以允许您异步执行多个请求。

def rpc_call():
    # Fire of all three requests.
    corr_id1 = rpc_client.send_request('1')
    corr_id2 = rpc_client.send_request('2')
    corr_id3 = rpc_client.send_request('3')

    # Wait for the response on all three requests.
    while not rpc_client.queue[corr_id1] \
            or not rpc_client.queue[corr_id2] \
            or not rpc_client.queue[corr_id3]:
        sleep(0.1)

    # Print the result of all three requests.
    print rpc_client.queue[corr_id1]
    print rpc_client.queue[corr_id2]
    print rpc_client.queue[corr_id3]

if __name__ == '__main__':
    rpc_client = RpcClient('rpc_queue')
    sleep(1)
    print rpc_call()