RabbitMQ消息队列

rabbitmq官方文档：http://www.rabbitmq.com/tutorials/tutorial-two-python.html

rabbitmq中文文档：http://rabbitmq.mr-ping.com/tutorials_with_python/[2]Work_Queues.html

介绍

　　RabbitMQ是一款基于AMQP协议的消息中间件，它能够在应用之间提供可靠的消息传输。在易用性，扩展性，高可用性上表现优秀。而且使用消息中间件利于应用之间的解耦，生产者（客户端）无需知道消费者（服务端）的存在。而且两端可以使用不同的语言编写，大大提供了灵活性。简单来说，RabbitMQ就是一个消息代理 – 一个消息系统的媒介。它可以为你的应用提供一个通用的消息发送和接收平台，并且保证消息在传输过程中的安全。

应用场景

　　在项目中，将一些无需即时返回且耗时的操作提取出来，进行了异步处理，而这种异步处理的方式大大的节省了服务器的请求响应时间，从而提高了系统的吞吐量。

RabbitMQ结构图

消息队列的使用过程大概：

　　（1）客户端连接到消息队列服务器，打开一个channel。

　　（2）客户端声明一个exchange，并设置相关属性。

　　（3）客户端声明一个queue，并设置相关属性。

　　（4）客户端使用routing key，在exchange和queue之间建立好绑定关系。

　　（5）客户端投递消息到exchange。 exchange接收到消息后，就根据消息的key和已经设置的binding，进行消息路由，将消息投递到一个或多个队列里。 exchange也有几个类型，完全根据key进行投递的叫做Direct交换机，例如，绑定时设置了routing key为”abc”，那么客户端提交的消息，只有设置了key为”abc”的才会投递到队列。对key进行模式匹配后进行投递的叫做Topic交换机，符号”#”匹配一个或多个词，符号”*”匹配正好一个词。例如”abc.#”匹配”abc.def.ghi”，”abc.*”只匹配”abc.def”。还有一种不需要key的，叫做Fanout交换机，它采取广播模式，一个消息进来时，投递到与该交换机绑定的所有队列。

1、RabbitMQ实现一个简单的通信

RabbitMQ库

　　RabbitMQ使用的是AMQP协议。要使用她你就必须需要一个使用同样协议的库。几乎所有的编程语言都有可选择的库。python也是一样，可以从以下几个库中选择：

py-amqplib
txAMQP
pika

pip install pika

　　大致的设计是这样的：

　　生产者（producer）把消息发送到一个名为“hello”的队列中。消费者（consumer）从这个队列中获取消息。

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

#建立一个到RabbitMQ服务器的连接。
connection=pika.BlockingConnection(pika.ConnectionParameters('localhost'))

#创建一个通道
channel=connection.channel()

# 声明一个名为hello的队列
channel.queue_declare(queue='hello')

#n RabbitMQ a message can never sent directly to the  queue,it always need to
#go through exchange

channel.basic_publish(exchange='',
                      routing_key='hello',#要发送到的queue名字
                      body='Hello world')
print("[x] Sent 'Hell World'")
#退出程序前，通过关闭连接保证消息已经投递到RabbitMq   
connection.close()

send端

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
# 声明一个管道
channel = connection.channel()

#you may ask why declare the queue again - we have already declared in our previous code.We cloud avoid that if we were sure that the queue already exists .For example if send.py program was run before .But we're not yet sure which program to run first（send or receive are possible） .In such cases it's a good practice to repeat declaring the queue in both programs
# channel.queue_declare(queue='hello')
channel.queue_declare(queue='hello')

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

channel.basic_consume(
                      callback,#如果收到消息就调用callback函数处理消息
                      queue='hello',#queue名
                      no_ack=True)
print(' [*] Waiting for messages. To exit press CTRL+C')

channel.start_consuming()

receive端

2、Work Queues

　　Work Queues（工作队列）（又称：任务队列——Task Queues）是为了避免等待一些占用大量资源、时间的操作。当我们把任务（Task）当作消息发送到队列中，一个运行在后台的工作者（worker）进程就会取出任务然后处理。当你运行多个工作者（workers），任务就会在它们之间共享。

　　这个概念在网络应用中是非常有用的，它可以在短暂的HTTP请求中处理一些复杂的任务。

1、Round-robin dispatching（循环调度）

　　默认来说，RabbitMQ会按顺序得把消息发送给每个消费者（consumer）。平均每个消费者都会收到同等数量得消息。这种发送消息得方式叫做——轮询（round-robin）　　

在这种模式下，RabbitMQ会默认把p发的消息依次分发给各个消费者(c),跟负载均衡差不多，使用工作队列的一个好处就是它能够并行的处理队列。如果堆积了很多任务，我们只需要添加更多的工作者（workers）就可以了，扩展很简单。

# #!/usr/bin/env/ python
# # -*-coding:utf-8 -*-
import pika
import sys
#建立一个到RabbitMQ服务器的连接。
connection=pika.BlockingConnection(pika.ConnectionParameters('localhost'))

#创建一个通道
channel=connection.channel()

# 声明一个名为hello的队列
channel.queue_declare(queue='hello')

#n RabbitMQ a message can never sent directly to the  queue,it always need to
#go through exchange

message = ' '.join(sys.argv[1:]) or "Hello World!"

channel.basic_publish(exchange='',
                      routing_key='hello',
                      body=message)
print(" [x] Sent %r" % (message,))

connection.close()

task

import pika
import time

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
# 声明一个管道
channel = connection.channel()
#声明一个队列
channel.queue_declare(queue='hello')

def callback(ch, method, properties, body):
    print (" [x] Received %r" % (body,))
    time.sleep( body.count('.'.encode('utf-8')) )
    print(" [x] Done")

channel.basic_consume(
                      callback,#如果收到消息就调用callback函数处理消息
                      queue='hello',#queue名
                      no_ack=True)
print(' [*] Waiting for messages. To exit press CTRL+C')

channel.start_consuming()

worker

 2、Message acknowledgment(消息确认)

　　当处理一个比较耗时得任务的时候，你也许想知道消费者（consumers）是否运行到一半就挂掉。当前的代码中，当消息被RabbitMQ发送给消费者（consumers）之后，马上就会在内存中移除。这种情况，你只要把一个工作者（worker）停止，正在处理的消息就会丢失。同时，所有发送到这个工作者的还没有处理的消息都会丢失。

　　我们不想丢失任何任务消息。如果一个工作者（worker）挂掉了，我们希望任务会重新发送给其他的工作者（worker）。

　　为了防止消息丢失，RabbitMQ提供了消息响应（acknowledgments）。消费者会通过一个ack（响应），告诉RabbitMQ已经收到并处理了某条消息，然后RabbitMQ就会释放并删除这条消息。

　　如果消费者（consumer）挂掉了，没有发送响应，RabbitMQ就会认为消息没有被完全处理，然后重新发送给其他消费者（consumer）。这样，及时工作者（workers）偶尔的挂掉，也不会丢失消息。

　　消息是没有超时这个概念的；当工作者与它断开连的时候，RabbitMQ会重新发送消息。这样在处理一个耗时非常长的消息任务的时候就不会出问题了。

　　消息响应默认是开启的。之前的例子中我们可以使用no_ack=True（True表示无论这条消息发完与否都不会给服务端发消息）标识把它关闭。是时候移除这个标识了。当工作者（worker）完成了任务，就发送一个响应。

# #!/usr/bin/env/ python
# # -*-coding:utf-8 -*-
import pika
import sys
#建立一个到RabbitMQ服务器的连接。
connection=pika.BlockingConnection(pika.ConnectionParameters('localhost'))

#创建一个通道
channel=connection.channel()

# 声明一个名为hello的队列
channel.queue_declare(queue='hello')

#n RabbitMQ a message can never sent directly to the  queue,it always need to
#go through exchange

message = ' '.join(sys.argv[1:]) or "Hello World!"

channel.basic_publish(exchange='',
                      routing_key='hello',
                      body=message)
print(" [x] Sent %r" % (message,))

connection.close()

task

import pika
import time

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
# 声明一个管道
channel = connection.channel()
#声明一个队列
channel.queue_declare(queue='hello')

def callback(ch, method, properties, body):
    print (" [x] Received %r" % (body,))
    time.sleep( body.count('.'.encode('utf-8')) )
    print(" [x] Done")
    ch.basic_ack(delivery_tag=method.delivery_tag)#当工作者（worker）完成了任务，就发送一个响应。

channel.basic_consume(
                      callback,#如果收到消息就调用callback函数处理消息
                      queue='hello',#queue名
                    )
print(' [*] Waiting for messages. To exit press CTRL+C')

workers

忘记确认

　　一个很容易犯的错误就是忘了basic_ack，后果很严重。消息在你的程序退出之后就会重新发送，如果它不能够释放没响应的消息，RabbitMQ就会占用越来越多的内存。

为了排除这种错误，你可以使用rabbitmqctl命令，输出messages_unacknowledged字段：

$ sudo rabbitmqctl list_queues name messages_ready messages_unacknowledged
Listing queues ...
hello    0       0
...done.

如果windows环境下的话，去掉sudo

3、Message durability(消息持久化)

　　如果你没有特意告诉RabbitMQ，那么在它退出或者崩溃的时候，将会丢失所有队列和消息。为了确保信息不会丢失，有两个事情是需要注意的：我们必须把“队列”和“消息”设为持久化。

　　首先，为了不让队列消失，需要把队列声明为持久化（durable）：

channel.queue_declare(queue='hello', durable=True)#durable=True队列持久化，注意：task和workers端均要写上

　　这时候，我们只是确保在RabbitMq重启之后queue_declare队列不会丢失。另外，我们需要把我们的消息也要设为持久化——将delivery_mode的属性设为2。

channel.basic_publish(exchange='',
                      routing_key="task_queue",
                      body=message,
                      properties=pika.BasicProperties(
                         delivery_mode = 2, # make message persistent
                      ))

　　注意：消息持久化将消息设为持久化并不能完全保证不会丢失。以上代码只是告诉了RabbitMq要把消息存到硬盘，但从RabbitMq收到消息到保存之间还是有一个很小的间隔时间。因为RabbitMq并不是所有的消息都使用fsync(2)——它有可能只是保存到缓存中，并不一定会写到硬盘中。并不能保证真正的持久化，但已经足够应付我们的简单工作队列。如果你一定要保证持久化，你需要改写你的代码来支持事务（transaction）。

4、Fair dispatch(公平调度)

　　你应该已经发现，它仍旧没有按照我们期望的那样进行分发。比如有两个工作者（workers），处理奇数消息的比较繁忙，处理偶数消息的比较轻松。然而RabbitMQ并不知道这些，它仍然一如既往的派发消息。

　　这时因为RabbitMQ只管分发进入队列的消息，不会关心有多少消费者（consumer）没有作出响应。它盲目的把第n-th条消息发给第n-th个消费者。那么问题就来了，Rabbit这样只按顺序把消息发到各个消费者身上，不考虑消费者负载的话，很可能出现，一个机器配置不高的消费者那里堆积了很多消息处理不完，同时配置高的消费者却一直很轻松

　　我们可以使用basic.qos方法，并设置prefetch_count=1。这样是告诉RabbitMQ，再同一时刻，不要发送超过1条消息给一个工作者（worker），直到它已经处理了上一条消息并且作出了响应。这样，RabbitMQ就会把消息分发给下一个空闲的工作者（worker）。

channel.basic_qos(prefetch_count=1)

关于队列大小

　　不过，这样有可能会出现这样的问题：如果所有的工作者都处理繁忙状态，你的队列就会被填满。你需要留意这个问题，要么添加更多的工作者（workers），要么使用其他策略。

带消息持久化+公平调度的完整代码：

import pika
import sys
 
connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='localhost'))
channel = connection.channel()
 
channel.queue_declare(queue='task_queue', durable=True)
 
message = ' '.join(sys.argv[1:]) or "Hello World!"
channel.basic_publish(exchange='',
                      routing_key='task_queue',
                      body=message,
                      properties=pika.BasicProperties(
                         delivery_mode = 2, # make message persistent
                      ))
print(" [x] Sent %r" % message)
connection.close()

task

#!/usr/bin/env python
import pika
import time
 
connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='localhost'))
channel = connection.channel()
 
channel.queue_declare(queue='task_queue', durable=True)
print(' [*] Waiting for messages. To exit press CTRL+C')
 
def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)
    time.sleep(body.count(b'.'))
    print(" [x] Done")
    ch.basic_ack(delivery_tag = method.delivery_tag)
 
channel.basic_qos(prefetch_count=1)
channel.basic_consume(callback,
                      queue='task_queue')
 
channel.start_consuming()

work

5、Publish/Subscribe(发布/订阅)

　　前面，我们搭建了一个工作队列，每个任务只分发给一个工作者（worker）。但现在我们要做的跟之前完全不一样 —— 分发一个消息给多个消费者（consumers）。这种模式被称为“发布／订阅”。

　　发布订阅和简单的消息队列区别在于，发布订阅会将消息发送给所有的订阅者，而消息队列中的数据被消费一次便消失。所以，RabbitMQ实现发布和订阅时，会为每一个订阅者创建一个队列，而发布者发布消息时，会将消息放置在所有相关队列中。

　　为了描述这种模式，我们将会构建一个简单的日志系统。它包括两个程序——第一个程序负责发送日志消息，第二个程序负责获取消息并输出内容。

　　在我们的这个日志系统中，所有正在运行的接收方程序都会接受消息。我们用其中一个接收者（receiver）把日志写入硬盘中，另外一个接受者（receiver）把日志输出到屏幕上。

　　最终，日志消息被广播给所有的接受者（receivers）。

　　要到达这种效果，这时候exchanges就派上用场了

Exchanges(交换机)

　　RabbitMQ消息模型的核心理念是：发布者（producer）不会直接发送任何消息给队列。事实上，发布者（producer）甚至不知道消息是否已经被投递到队列。

　　发布者（producer）只需要把消息发送给一个交换机（exchange）。交换机非常简单，它一边从发布者方接收消息，一边把消息推送到队列。交换机必须知道如何处理它接收到的消息，是应该推送到指定的队列还是是多个队列，或者是直接忽略消息。这些规则是通过交换机类型（exchange type）来定义的。

　　有几个可供选择的交换机类型：

direct（直连交换机）：通过routingKey和exchange决定的那个唯一的queue可以接收消息

topic（主题交换机）：所有符合routingKey(此时可以是一个表达式)的routingKey所bind的queue可以接收消息

 　　表达式符号说明：#代表一个或多个字符，*代表任何字符
      　　例：#.a会匹配a.a，aa.a，aaa.a等
         　　 *.a会匹配a.a，b.a，c.a等
     　　注：使用RoutingKey为#，Exchange Type为topic的时候相当于使用fanout

headers（头交换机）：通过headers 来决定把消息发给哪些queue

fanout（扇形交换机）：所有bind到此exchange的queue都可以接收消息

　　我们在这里主要说明最后一个 —— 扇型交换机（fanout）。先创建一个fanout类型的交换机，命名为logs：

channel.exchange_declare(exchange='logs',
                         type='fanout')

　　扇型交换机（fanout）很简单，你可能从名字上就能猜测出来，它把消息发送给它所知道的所有队列。这正是我们的日志系统所需要的。

交换器列表：

#rabbitmqctl能够列出服务器上所有的交换器：

$ sudo rabbitmqctl list_exchanges
Listing exchanges ...
logs      fanout
amq.direct      direct
amq.topic       topic
amq.fanout      fanout
amq.headers     headers
...done.

#这个列表中有一些叫做amq.*的交换器。这些都是默认创建的，不过这时候你还不需要使用他们。

匿名的交换器：

　　前面，我们对交换机一无所知，但仍然能够发送消息到队列中。因为我们使用了命名为空字符串(“”)默认的交换机。

回想我们之前是这样发布一则消息：

channel.basic_publish(exchange='',
                      routing_key='hello',
                      body=message)

#exchange参数就是交换机的名称。空字符串代表默认或者匿名交换机：消息将会根据指定的routing_key分发到指定的队列。

现在，我们就可以发送消息到一个具名交换机了：

channel.basic_publish(exchange='logs',
                      routing_key='',#广播，故不用指定queue名
                      body=message)

临时队列

　　你还记得之前我们使用的队列名吗（ hello和task_queue）？给一个队列命名是很重要的——我们需要把工作者（workers）指向正确的队列。如果你打算在发布者（producers）和消费者（consumers）之间共享同队列的话，给队列命名是十分重要的。

　　但是这并不适用于我们的日志系统。我们打算接收所有的日志消息，而不仅仅是一小部分。我们关心的是最新的消息而不是旧的。为了解决这个问题，我们需要做两件事情。

首先，当我们连接上RabbitMQ的时候，我们需要一个全新的、空的队列。我们可以手动创建一个随机的队列名，或者让服务器为我们选择一个随机的队列名（推荐）。我们只需要在调用queue_declare方法的时候，不提供queue参数就可以了：

result = channel.queue_declare()

　　这时候我们可以通过result.method.queue获得已经生成的随机队列名。它可能是这样子的：amq.gen-U0srCoW8TsaXjNh73pnVAw==。

第二步，当与消费者（consumer）断开连接的时候，这个队列应当被立即删除。exclusive标识符即可达到此目的。

注意：因为广播是即时连通的，消费者是无法接收在其开启前生产者所发送的消息的，这就跟收音机的原理类似。

result = channel.queue_declare(exclusive=True)#exclusive排他的，唯一的，不指定queue名，rabbit会随机分配一个名字，exclusive=True会在使用此queue的消费者断开后，自动将queue删除

queue_name=result.method.queue #获得已经生成的随机队列名

绑定（Bindings）

　　我们已经创建了一个扇型交换机（fanout）和一个队列。这时候我们需要告诉交换机如何发送消息给我们的队列。交换器和队列之间的联系我们称之为绑定（binding）。

channel.queue_bind(exchange='logs',
                   queue=result.method.queue)

　　现在，logs交换机将会把消息添加到我们的队列中

rabbitmqctl list_bindings #列出所有现存的绑定。

代码整合：

import pika
import sys

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

channel.exchange_declare(exchange='logs',
                         type='fanout')

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

publisher

import pika

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

channel.exchange_declare(exchange='logs',
                         type='fanout')

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

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

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

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

channel.basic_consume(callback,
                      queue=queue_name,
                      no_ack=True)

channel.start_consuming()

subscriber

6、Routing路由

　　RabbitMQ还支持根据关键字发送，即：队列绑定关键字，发送者将数据根据关键字发送到消息exchange，exchange根据 关键字 判定应该将数据发送至指定队列。

　　在前面中，我们使用的扇型交换机（fanout exchange）没有足够的灵活性 —— 它能做的仅仅是广播。这时候我们将会使用直连交换机（direct exchange）来代替。路由的算法很简单 —— 交换机将会对绑定键（binding key）和路由键（routing key）进行精确匹配，从而确定消息该分发到哪个队列。

direct exchange

　　在这个场景中，我们可以看到直连交换机 X和两个队列进行了绑定。第一个队列使用error作为绑定键，第二个队列有三个绑定，一个使用info作为绑定键，第二个使用error，第三个使用warning。

这样一来，当路由键为error的消息发布到交换机，就会被路由到队列Q1和Q2。路由键为info或者warning的消息就会路由到Q2。其他的所有消息都将会被丢弃。

import pika
import sys
 
connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='localhost'))
channel = connection.channel()
 
channel.exchange_declare(exchange='direct_logs',
                         type='direct')
 
severity = sys.argv[1] if len(sys.argv) > 1 else 'info'
message = ' '.join(sys.argv[2:]) or 'Hello World!'
channel.basic_publish(exchange='direct_logs',
                      routing_key=severity,
                      body=message)
print(" [x] Sent %r:%r" % (severity, message))
connection.close()

publisher

import pika
import sys
 
connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='localhost'))
channel = connection.channel()
 
channel.exchange_declare(exchange='direct_logs',
                         type='direct')
 
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
 
severities = sys.argv[1:]
if not severities:
    sys.stderr.write("Usage: %s 
 
 [error]\n" % sys.argv[0])
    sys.exit(1)
 
for severity in severities:
    channel.queue_bind(exchange='direct_logs',
                       queue=queue_name,
                       routing_key=severity)
 
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(callback,
                      queue=queue_name,
                      no_ack=True)
 
channel.start_consuming()

subscriber

 7、topic exchange主题交换机

　　尽管直连交换机能够改善我们的系统，但是它也有它的限制 —— 没办法基于多个标准执行路由操作。

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

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

• * (星号) 用来表示一个单词.
• # (井号) 用来表示任意数量（零个或多个）单词。

下边用图说明：

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

我们创建了三个绑定：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" 有四个单词，他还是会匹配最后一个绑定，并且被投递到第二个队列中。

主题交换机

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

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

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

组合在一起

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

整合：

import pika
import sys
 
connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='localhost'))
channel = connection.channel()
 
channel.exchange_declare(exchange='topic_logs',
                         type='topic')
 
routing_key = sys.argv[1] if len(sys.argv) > 1 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()

publisher

import pika
import sys
 
connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='localhost'))
channel = connection.channel()
 
channel.exchange_declare(exchange='topic_logs',
                         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(callback,
                      queue=queue_name,
                      no_ack=True)
 
channel.start_consuming()

subscriber

#执行下边命令 接收所有日志：
python receive_logs_topic.py "#"

#执行下边命令 接收来自”kern“设备的日志：
python receive_logs_topic.py "kern.*"

#执行下边命令 只接收严重程度为”critical“的日志：
python receive_logs_topic.py "*.critical"

#执行下边命令 建立多个绑定：
python receive_logs_topic.py "kern.*" "*.critical"

#执行下边命令 发送路由键为 "kern.critical" 的日志：
python emit_log_topic.py "kern.critical" "A critical kernel error"

#另外，上边代码不会对路由键和绑定键做任何假设，所以你可以在命令中使用超过两个路由键参数。

8、RPC远程过程调用

　　当我们需要将一个函数运行在远程计算机上并且等待从那儿获取结果时，这时候我们就需要用到RPC（Remote Procedure Call）了。

构建一个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(exclusive=True)
callback_queue = result.method.queue

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

# ... and some code to read a response message from the callback_queue ...

消息属性:

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的工作流程：

我们的RPC如此工作:

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

创建一个模拟RPC服务来返回斐波那契数列：

服务器端：

• （4）像往常一样，我们建立连接，声明队列
• （11）我们声明我们的fibonacci函数，它假设只有合法的正整数当作输入。（别指望这个函数能处理很大的数值，函数递归你们都懂得…）
• （19）我们为 basic_consume 声明了一个回调函数，这是RPC服务器端的核心。它执行实际的操作并且作出响应。
• （32）或许我们希望能在服务器上多开几个线程。为了能将负载平均地分摊到多个服务器，我们需要将 prefetch_count 设置好。

#!/usr/bin/env/ python
# -*-coding:utf-8 -*-
import pika
import time

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,#客户端指定返回消息的queue
                     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()

rpc_server

客户端：

• （7）建立连接、通道并且为回复（replies）声明独享的回调队列。
• （16）我们订阅这个回调队列，以便接收RPC的响应。
• （18）“on_response”回调函数对每一个响应执行一个非常简单的操作，检查每一个响应消息的correlation_id属性是否与我们期待的一致，如果一致，将响应结果赋给self.response，然后跳出consuming循环。
• （23）接下来，我们定义我们的主要方法 call 方法。它执行真正的RPC请求。
• （24）在这个方法中，首先我们生成一个唯一的 correlation_id 值并且保存起来，’on_response’回调函数会用它来获取符合要求的响应。
• （25）接下来，我们将带有 reply_to 和 correlation_id 属性的消息发布出去。
• （32）现在我们可以坐下来，等待正确的响应到来。
• （33）最后，我们将响应返回给用户。

#!/usr/bin/env/ python
# -*-coding:utf-8 -*-

import pika
import uuid
#uuid 通用唯一标识符 ( Universally Unique Identifier )，通过MAC地址、时间戳、命名空间、随机数、伪随机数来保证生成ID的唯一性
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

#声明一个consumer，用于处理请求
        self.channel.basic_consume(self.on_response, #只要一收到消息，就调用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())#uuid4()——基于随机数
        self.channel.basic_publish(exchange='',
                                   routing_key='rpc_queue',
                                   properties=pika.BasicProperties(
                                       reply_to=self.callback_queue,#让服务器端执行完命令后，把消息返回到这个queue里
                                       correlation_id=self.corr_id,
                                   ),
                                   body=str(n))
        # 接收返回的数据
        while self.response is None:
            self.connection.process_data_events() #非阻塞版的start_consuming(),没有消息不阻塞
        return int(self.response)

fibonacci_rpc = FibonacciRpcClient()

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

rpc_client