Rabbitmq （二）

Rabbitmq （二） 主题交换机

发送到主题交换机（topic exchange）的消息不可以携带随意格式的路由键（routing_key），它的路由键必须是一个由 . 分隔开的词语列表。这些单词随便是什么都可以，但是最好是跟携带它们的消息有关系的词汇。以下是几个推荐的例子："stock.usd.nyse", "nyse.vmw", "quick.orange.rabbit"。词语的个数可以随意，但是不要超过 255 字节。

绑定键也必须拥有同样的格式。主题交换机背后的逻辑跟直连交换机很相似，一个携带着特定路由键的消息会被主题交换机投递给绑定键与之相匹配的队列。但是它的绑定键和路由键有两个特殊应用方式：

• * 用来表示一个单词
• # 用来表示任意数量（零个或多个）单词

接下来用图来介绍一下：

这个例子里，我们发送的所有消息都是用来描述小动物的。发送的消息所携带的路由键是由三个单词所组成的，这三个单词被两个 . 分割开。路由键里的第一个单词描述的是动物的手脚的利索程度，第二个单词是动物的颜色，第三个是动物的种类。

我们创建了三个绑定：

• Q1 的绑定键为 *.orange.*；
• Q2 的绑定键为 *.*.rabbit 和 lazy.#。

这三个绑定键被可以总结为：

• Q1 对所有的橘黄色动物都感兴趣；
• Q2 则是对所有的兔子和所有懒惰的动物感兴趣。

一个携带有 quick.orange.rabbit 的消息将会被分别投递给这两个队列。携带着 lazy.orange.elephant 的消息同样也会给两个队列都投递过去。另一方面携带有 quick.orange.fox 的消息会投递给第一个队列，携带有 lazy.brown.fox 的消息会投递给第二个队列。携带有 lazy.pink.rabbit 的消息只会被投递给第二个队列一次，即使它同时匹配第二个队列的两个绑定。携带着 quick.brown.fox 的消息不会投递给任何一个队列。

如果我们违反约定，发送了一个携带有一个单词或者四个单词（"orange" or "quick.orange.male.rabbit"）的消息时，发送的消息不会投递给任何一个队列，而且会丢失掉。

但是另一方面，即使 "lazy.orange.male.rabbit" 有四个单词，他还是会匹配最后一个绑定，并且被投递到第二个队列中。

主题交换机与其他交换机的联系：

主题交换机是很强大的，它可以表现出跟其他交换机类似的行为：

当一个队列的绑定键为 "#"（井号） 的时候，这个队列将会无视消息的路由键，接收所有的消息。

当 * (星号) 和 # (井号) 这两个特殊字符都未在绑定键中出现的时候，此时主题交换机就拥有的直连交换机的行为。

接下来我们会将主题交换机应用到我们的日志系统中。在开始工作前，我们假设日志的路由键由两个单词组成，路由键看起来是这样的：

.

代码跟上一节课程差不多。

编写源代码文件 /home/shiyanlou/Code/emit_log_topic.py：

#!/usr/bin/env python3
import pika
import sys

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

channel.exchange_declare(exchange='topic_logs', exchange_type='topic')

routing_key = sys.argv[1] if len(sys.argv) > 2 else 'anonymous.info'
message = ' '.join(sys.argv[2:]) or 'Hello World!'
channel.basic_publish(
    exchange='topic_logs', routing_key=routing_key, body=message)
print(" [x] Sent %r:%r" % (routing_key, message))
connection.close()

编写源代码文件 /home/shiyanlou/Code/receive_logs_topic.py：

#!/usr/bin/env python3
import pika
import sys

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

channel.exchange_declare(exchange='topic_logs', exchange_type='topic')

result = channel.queue_declare('', exclusive=True)
queue_name = result.method.queue

binding_keys = sys.argv[1:]
if not binding_keys:
    sys.stderr.write("Usage: %s [binding_key]...n" % sys.argv[0])
    sys.exit(1)

for binding_key in binding_keys:
    channel.queue_bind(
        exchange='topic_logs', queue=queue_name, routing_key=binding_key)

print(' [*] Waiting for logs. To exit press CTRL+C')


def callback(ch, method, properties, body):
    print(" [x] %r:%r" % (method.routing_key, body))


channel.basic_consume(
    queue=queue_name, on_message_callback=callback, auto_ack=True)

channel.start_consuming()

远程调用 

        可是如果我们需要将一个函数运行在远程计算机上并且等待从那儿获取结果时，该怎么办呢？这就是另外的故事了。这种模式通常被称为远程程序调用（Remote Procedure Call）或者 RPC。本篇课程有点难度，在学习的时候建议多看几遍，多多实践。

这篇教程中，我们会使用 RabbitMQ 来构建一个 RPC 系统：包含一个客户端和一个 RPC 服务器。现在的情况是，我们没有一个值得被分发的足够耗时的任务，所以接下来，我们会创建一个模拟 RPC 服务来返回斐波那契数列。

为了展示 RPC 服务如何使用，我们创建了一个简单的客户端类。它会暴露出一个名为 “call” 的方法用来发送一个 RPC 请求，并且在收到回应前保持阻塞。

fibonacci_rpc = FibonacciRpcClient()
result = fibonacci_rpc.call(4)
print "fib(4) is %r" % (result,)

RPC 使用注意

尽管 RPC 在计算领域是一个常用模式，但它也经常被诟病。当一个问题被抛出的时候，程序员往往意识不到这到底是由本地调用还是由较慢的 RPC 调用引起的。同样的困惑还来自于系统的不可预测性和给调试工作带来的不必要的复杂性。跟软件精简不同的是，滥用 RPC 会导致不可维护的 面条代码。

考虑到这一点，牢记以下建议：

• 确保能够明确的搞清楚哪个函数是本地调用的，哪个函数是远程调用的。
• 给你的系统编写文档。保持各个组件间的依赖明确。处理错误案例。明确客户端该如何处理 RPC 服务器的宕机和长时间无响应情况。
• 当对避免使用 RPC 有疑问的时候。

如果可以的话，你应该尽量使用异步管道来代替 RPC 类的阻塞。结果被异步地推送到下一个计算场景。

一般来说通过 RabbitMQ 来实现 RPC 是很容易的。一个客户端发送请求信息，服务器端将其应用到一个回复信息中。为了接收到回复信息，客户端需要在发送请求的时候同时发送一个回调队列（callback queue）的地址。我们试试看：

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

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

消息属性

AMQP 协议给消息预定义了一系列的 14 个属性。大多数属性很少会用到，除了以下几个：

• delivery_mode（投递模式）：将消息标记为持久的（值为 2）或暂存的（除了 2 之外的其他任何值）。《工作队列》实验里接触过这个属性，记得吧？
• content_type（内容类型）:用来描述编码的 mime-type。例如在实际使用中常常使用 application/json 来描述 JOSN 编码类型。
• reply_to（回复目标）：通常用来命名回调队列。
• correlation_id（关联标识）：用来将 RPC 的响应和请求关联起来。

关联标志

        上边介绍的方法中，我们建议给每一个 RPC 请求新建一个回调队列。这不是一个高效的做法，幸好这儿有一个更好的办法 —— 我们可以为每个客户端只建立一个独立的回调队列。

这就带来一个新问题，当此队列接收到一个响应的时候它无法辨别出这个响应是属于哪个请求的。correlation_id 就是为了解决这个问题而来的。我们给每个请求设置一个独一无二的值。稍后，当我们从回调队列中接收到一个消息的时候，我们就可以查看这条属性从而将响应和请求匹配起来。如果我们接收到的消息的 correlation_id 是未知的，那就直接销毁掉它，因为它不属于我们的任何一条请求。

你也许会问，为什么我们接收到未知消息的时候不抛出一个错误，而是要将它忽略掉？这是为了解决服务器端有可能发生的竞争情况。尽管可能性不大，但 RPC 服务器还是有可能在已将应答发送给我们但还未将确认消息发送给请求的情况下死掉。如果这种情况发生，RPC 在重启后会重新处理请求。这就是为什么我们必须在客户端优雅的处理重复响应，同时 RPC 也需要尽可能保持 幂等性。

总结

我们的 RPC 如此工作:

• （1）当客户端启动的时候，它创建一个匿名独享的回调队列。
• （2）在 RPC 请求中，客户端发送带有两个属性的消息：一个是设置回调队列的 reply_to 属性，另一个是设置唯一值的 correlation_id 属性。
• （3）将请求发送到一个 rpc_queue 队列中。
• （4）RPC 工作者（又名：服务器）等待请求发送到这个队列中来。当请求出现的时候，它执行他的工作并且将带有执行结果的消息发送给 reply_to 字段指定的队列。
• （5）客户端等待回调队列里的数据。当有消息出现的时候，它会检查 correlation_id 属性。如果此属性的值与请求匹配，将它返回给应用。

回调队列 

        一般来说通过 RabbitMQ 来实现 RPC 是很容易的。一个客户端发送请求信息，服务器端将其应用到一个回复信息中。为了接收到回复信息，客户端需要在发送请求的时候同时发送一个回调队列（callback queue）的地址。我们试试看：

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

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

消息属性

AMQP 协议给消息预定义了一系列的 14 个属性。大多数属性很少会用到，除了以下几个：

• delivery_mode（投递模式）：将消息标记为持久的（值为 2）或暂存的（除了 2 之外的其他任何值）。《工作队列》实验里接触过这个属性，记得吧？
• content_type（内容类型）:用来描述编码的 mime-type。例如在实际使用中常常使用 application/json 来描述 JOSN 编码类型。
• reply_to（回复目标）：通常用来命名回调队列。
• correlation_id（关联标识）：用来将 RPC 的响应和请求关联起来。

关联标志

上边介绍的方法中，我们建议给每一个 RPC 请求新建一个回调队列。这不是一个高效的做法，幸好这儿有一个更好的办法 —— 我们可以为每个客户端只建立一个独立的回调队列。

这就带来一个新问题，当此队列接收到一个响应的时候它无法辨别出这个响应是属于哪个请求的。correlation_id 就是为了解决这个问题而来的。我们给每个请求设置一个独一无二的值。稍后，当我们从回调队列中接收到一个消息的时候，我们就可以查看这条属性从而将响应和请求匹配起来。如果我们接收到的消息的 correlation_id 是未知的，那就直接销毁掉它，因为它不属于我们的任何一条请求。

你也许会问，为什么我们接收到未知消息的时候不抛出一个错误，而是要将它忽略掉？这是为了解决服务器端有可能发生的竞争情况。尽管可能性不大，但 RPC 服务器还是有可能在已将应答发送给我们但还未将确认消息发送给请求的情况下死掉。如果这种情况发生，RPC 在重启后会重新处理请求。这就是为什么我们必须在客户端优雅的处理重复响应，同时 RPC 也需要尽可能保持 幂等性。

总结

我们的 RPC 如此工作:

• （1）当客户端启动的时候，它创建一个匿名独享的回调队列。
• （2）在 RPC 请求中，客户端发送带有两个属性的消息：一个是设置回调队列的 reply_to 属性，另一个是设置唯一值的 correlation_id 属性。
• （3）将请求发送到一个 rpc_queue 队列中。
• （4）RPC 工作者（又名：服务器）等待请求发送到这个队列中来。当请求出现的时候，它执行他的工作并且将带有执行结果的消息发送给 reply_to 字段指定的队列。
• （5）客户端等待回调队列里的数据。当有消息出现的时候，它会检查 correlation_id 属性。如果此属性的值与请求匹配，将它返回给应用。

编写源代码文件 /home/shiyanlou/Code/shiyanlou_cs637/rpc_server.py：

#-*- coding:utf-8 –*-
#!/usr/bin/env python3
import pika

# 像往常一样，我们建立连接，声明队列
connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='localhost'))

channel = connection.channel()

channel.queue_declare(queue='rpc_queue')

# 声明我们的 fibonacci 函数，它假设只有合法的正整数当作输入。
def fib(n):
    if n == 0:
        return 0
    elif n == 1:
        return 1
    else:
        return fib(n-1) + fib(n-2)

#为 basic_consume 声明了一个回调函数，这是RPC服务器端的核心。它执行实际的 *** 作并且作出响应。
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)

# 或许我们希望能在服务器上多开几个线程。
# 为了能将负载平均地分摊到多个服务器，我们需要将 prefetch_count 设置好
channel.basic_qos(prefetch_count=1)

channel.basic_consume(queue='rpc_queue', on_message_callback=on_request)

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

• rpc_client.py 代码，编写源代码文件 /home/shiyanlou/Code/rpc_client.py：

#-*- coding:utf-8 –*-
#!/usr/bin/env python3
import pika
import uuid

class FibonacciRpcClient(object):
    def __init__(self):
# 建立连接、通道并且为回复（replies）声明独享的回调队列
        self.connection = pika.BlockingConnection(pika.ConnectionParameters(
                host='localhost'))

        self.channel = self.connection.channel()

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

        # 订阅这个回调队列，以便接收RPC的响应
        self.channel.basic_consume(
            queue=self.callback_queue,
            on_message_callback=self.on_response,
            auto_ack=True)

# on_response回调函数对每一个响应执行一个非常简单的 *** 作，检查每一个响应消息的correlation_id属性是否与我们期待的一致，如果一致，将响应结果赋给self.response，然后跳出consuming循环
    def on_response(self, ch, method, props, body):
        if self.corr_id == props.correlation_id:
            self.response = body

# 接下来，我们定义我们的主要方法 call 方法。它执行真正的RPC请求。
# 在这个方法中，首先我们生成一个唯一的 correlation_id，值并且保存起来，'on_response'回调函数会用它来获取符合要求的响应。
# 接下来，我们将带有 reply_to 和 correlation_id 属性的消息发布出去
    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)

此处呈现的设计并不是实现 RPC 服务的唯一方式，但是他有一些重要的优势：

1. 如果 RPC 服务器运行的过慢的时候，你可以通过运行另外一个服务器端轻松扩展它。试试在控制台中运行第二个 rpc_server.py 。
2. 在客户端， RPC 请求只发送或接收一条消息。不需要像 queue_declare 这样的异步调用。所以 RPC 客户端的单个请求只需要一个网络往返。

我们的代码依旧非常简单，而且没有试图去解决一些复杂（但是重要）的问题，如：

• 当没有服务器运行时，客户端如何作出反映。
• 客户端是否需要实现类似 RPC 超时的东西。
• 如果服务器发生故障，并且抛出异常，应该被转发到客户端吗？
• 在处理前，防止混入无效的信息（例如检查边界）