python -- RabbitMq简单使用

时间 2019-12-11

原文原文链接

转至：http://blog.csdn.net/whycold/article/details/41119807

我转到这里主要是想在本身使用时方便查询。。。

引言

你是否遇到过两个（多个）系统间须要经过定时任务来同步某些数据？你是否在为异构系统的不一样进程间相互调用、通信的问题而苦恼、挣扎？若是是，那么恭喜你，消息服务让你能够很轻松地解决这些问题。
消息服务擅长于解决多系统、异构系统间的数据交换（消息通知/通信）问题，你也能够把它用于系统间服务的相互调用（RPC）。本文将要介绍的RabbitMQ就是当前最主流的消息中间件之一。
php

RabbitMQ简介

AMQP，即Advanced Message Queuing Protocol，高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。
AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。
RabbitMQ是一个开源的AMQP实现，服务器端用Erlang语言编写，支持多种客户端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持AJAX。用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。
下面将重点介绍RabbitMQ中的一些基础概念，了解了这些概念，是使用好RabbitMQ的基础。java

ConnectionFactory、Connection、Channel

ConnectionFactory、Connection、Channel都是RabbitMQ对外提供的API中最基本的对象。Connection是RabbitMQ的socket连接，它封装了socket协议相关部分逻辑。ConnectionFactory为Connection的制造工厂。
Channel是咱们与RabbitMQ打交道的最重要的一个接口，咱们大部分的业务操做是在Channel这个接口中完成的，包括定义Queue、定义Exchange、绑定Queue与Exchange、发布消息等。python

Queue

Queue（队列）是RabbitMQ的内部对象，用于存储消息，用下图表示。
redis

RabbitMQ中的消息都只能存储在Queue中，生产者（下图中的P）生产消息并最终投递到Queue中，消费者（下图中的C）能够从Queue中获取消息并消费。
安全

多个消费者能够订阅同一个Queue，这时Queue中的消息会被平均分摊给多个消费者进行处理，而不是每一个消费者都收到全部的消息并处理。
服务器

Message acknowledgment

在实际应用中，可能会发生消费者收到Queue中的消息，但没有处理完成就宕机（或出现其余意外）的状况，这种状况下就可能会致使消息丢失。为了不这种状况发生，咱们能够要求消费者在消费完消息后发送一个回执给RabbitMQ，RabbitMQ收到消息回执（Message acknowledgment）后才将该消息从Queue中移除；若是RabbitMQ没有收到回执并检测到消费者的RabbitMQ链接断开，则RabbitMQ会将该消息发送给其余消费者（若是存在多个消费者）进行处理。这里不存在timeout概念，一个消费者处理消息时间再长也不会致使该消息被发送给其余消费者，除非它的RabbitMQ链接断开。
这里会产生另一个问题，若是咱们的开发人员在处理完业务逻辑后，忘记发送回执给RabbitMQ，这将会致使严重的bug——Queue中堆积的消息会愈来愈多；消费者重启后会重复消费这些消息并重复执行业务逻辑…负载均衡

另外pub message是没有ack的。异步

Message durability

若是咱们但愿即便在RabbitMQ服务重启的状况下，也不会丢失消息，咱们能够将Queue与Message都设置为可持久化的（durable），这样能够保证绝大部分状况下咱们的RabbitMQ消息不会丢失。但依然解决不了小几率丢失事件的发生（好比RabbitMQ服务器已经接收到生产者的消息，但还没来得及持久化该消息时RabbitMQ服务器就断电了），若是咱们须要对这种小几率事件也要管理起来，那么咱们要用到事务。因为这里仅为RabbitMQ的简单介绍，因此这里将不讲解RabbitMQ相关的事务。socket

Prefetch count

前面咱们讲到若是有多个消费者同时订阅同一个Queue中的消息，Queue中的消息会被平摊给多个消费者。这时若是每一个消息的处理时间不一样，就有可能会致使某些消费者一直在忙，而另一些消费者很快就处理完手头工做并一直空闲的状况。咱们能够经过设置prefetchCount来限制Queue每次发送给每一个消费者的消息数，好比咱们设置prefetchCount=1，则Queue每次给每一个消费者发送一条消息；消费者处理完这条消息后Queue会再给该消费者发送一条消息。
分布式

Exchange

在上一节咱们看到生产者将消息投递到Queue中，实际上这在RabbitMQ中这种事情永远都不会发生。实际的状况是，生产者将消息发送到Exchange（交换器，下图中的X），由Exchange将消息路由到一个或多个Queue中（或者丢弃）。

Exchange是按照什么逻辑将消息路由到Queue的？这个将在Binding一节介绍。
RabbitMQ中的Exchange有四种类型，不一样的类型有着不一样的路由策略，这将在Exchange Types一节介绍。

routing key

生产者在将消息发送给Exchange的时候，通常会指定一个routing key，来指定这个消息的路由规则，而这个routing key须要与Exchange Type及binding key联合使用才能最终生效。
在Exchange Type与binding key固定的状况下（在正常使用时通常这些内容都是固定配置好的），咱们的生产者就能够在发送消息给Exchange时，经过指定routing key来决定消息流向哪里。
RabbitMQ为routing key设定的长度限制为255 bytes。

Binding

RabbitMQ中经过Binding将Exchange与Queue关联起来，这样RabbitMQ就知道如何正确地将消息路由到指定的Queue了。

Binding key

在绑定（Binding）Exchange与Queue的同时，通常会指定一个binding key；消费者将消息发送给Exchange时，通常会指定一个routing key；当binding key与routing key相匹配时，消息将会被路由到对应的Queue中。这个将在Exchange Types章节会列举实际的例子加以说明。
在绑定多个Queue到同一个Exchange的时候，这些Binding容许使用相同的binding key。
binding key 并非在全部状况下都生效，它依赖于Exchange Type，好比fanout类型的Exchange就会无视binding key，而是将消息路由到全部绑定到该Exchange的Queue。

Exchange Types

RabbitMQ经常使用的Exchange Type有fanout、direct、topic、headers这四种（AMQP规范里还提到两种Exchange Type，分别为system与自定义，这里不予以描述），下面分别进行介绍。

fanout

fanout类型的Exchange路由规则很是简单，它会把全部发送到该Exchange的消息路由到全部与它绑定的Queue中。

上图中，生产者（P）发送到Exchange（X）的全部消息都会路由到图中的两个Queue，并最终被两个消费者（C1与C2）消费。

direct

direct类型的Exchange路由规则也很简单，它会把消息路由到那些binding key与routing key彻底匹配的Queue中。

以上图的配置为例，咱们以routingKey=”error”发送消息到Exchange，则消息会路由到Queue1（amqp.gen-S9b…，这是由RabbitMQ自动生成的Queue名称）和Queue2（amqp.gen-Agl…）；若是咱们以routingKey=”info”或routingKey=”warning”来发送消息，则消息只会路由到Queue2。若是咱们以其余routingKey发送消息，则消息不会路由到这两个Queue中。

topic

前面讲到direct类型的Exchange路由规则是彻底匹配binding key与routing key，但这种严格的匹配方式在不少状况下不能知足实际业务需求。topic类型的Exchange在匹配规则上进行了扩展，它与direct类型的Exchage类似，也是将消息路由到binding key与routing key相匹配的Queue中，但这里的匹配规则有些不一样，它约定：

routing key为一个句点号“. ”分隔的字符串（咱们将被句点号“. ”分隔开的每一段独立的字符串称为一个单词），如“stock.usd.nyse”、“nyse.vmw”、“quick.orange.rabbit”
binding key与routing key同样也是句点号“. ”分隔的字符串
binding key中能够存在两种特殊字符“*”与“#”，用于作模糊匹配，其中“*”用于匹配一个单词，“#”用于匹配多个单词（能够是零个）

以上图中的配置为例，routingKey=”quick.orange.rabbit”的消息会同时路由到Q1与Q2，routingKey=”lazy.orange.fox”的消息会路由到Q1与Q2，routingKey=”lazy.brown.fox”的消息会路由到Q2，routingKey=”lazy.pink.rabbit”的消息会路由到Q2（只会投递给Q2一次，虽然这个routingKey与Q2的两个bindingKey都匹配）；routingKey=”quick.brown.fox”、routingKey=”orange”、routingKey=”quick.orange.male.rabbit”的消息将会被丢弃，由于它们没有匹配任何bindingKey。

headers

headers类型的Exchange不依赖于routing key与binding key的匹配规则来路由消息，而是根据发送的消息内容中的headers属性进行匹配。
在绑定Queue与Exchange时指定一组键值对；当消息发送到Exchange时，RabbitMQ会取到该消息的headers（也是一个键值对的形式），对比其中的键值对是否彻底匹配Queue与Exchange绑定时指定的键值对；若是彻底匹配则消息会路由到该Queue，不然不会路由到该Queue。
该类型的Exchange没有用到过（不过也应该颇有用武之地），因此不作介绍。

RPC

MQ自己是基于异步的消息处理，前面的示例中全部的生产者（P）将消息发送到RabbitMQ后不会知道消费者（C）处理成功或者失败（甚至连有没有消费者来处理这条消息都不知道）。
但实际的应用场景中，咱们极可能须要一些同步处理，须要同步等待服务端将个人消息处理完成后再进行下一步处理。这至关于RPC（Remote Procedure Call，远程过程调用）。在RabbitMQ中也支持RPC。

RabbitMQ中实现RPC的机制是：

客户端发送请求（消息）时，在消息的属性（MessageProperties，在AMQP协议中定义了14中properties，这些属性会随着消息一块儿发送）中设置两个值replyTo（一个Queue名称，用于告诉服务器处理完成后将通知个人消息发送到这个Queue中）和correlationId（这次请求的标识号，服务器处理完成后须要将此属性返还，客户端将根据这个id了解哪条请求被成功执行了或执行失败）
服务器端收到消息并处理
服务器端处理完消息后，将生成一条应答消息到replyTo指定的Queue，同时带上correlationId属性
客户端以前已订阅replyTo指定的Queue，从中收到服务器的应答消息后，根据其中的correlationId属性分析哪条请求被执行了，根据执行结果进行后续业务处理

总结

　　本文介绍了RabbitMQ中我的认为最重要的概念，充分利用RabbitMQ提供的这些功能就能够处理咱们绝大部分的异步业务了。

一些例子

里面都带有我本身的一些理解注释

 1 #!/usr/bin/env python
 2 # -*- coding:utf8 -*-
 3 
 4 import pika
 5 
 6 # connection = pika.BlockingConnection(pika.ConnectionParameters(
 7 #     'localhost'))
 8 
 9 user_pass = pika.PlainCredentials('txowner','helloworld')
10 connection = pika.BlockingConnection(pika.ConnectionParameters(
11     host='localhost',
12     port=5672,
13     virtual_host='/',
14     credentials=user_pass
15 ))
16 
17 channel = connection.channel()
18 
19 channel.queue_declare(queue='message',durable=True)
20 
21 #这里参数形式模块定义好了，必须这样写，函数名是自定义的
22 def callback(ch,method,properties,body):
23     print(" [x] Received %r " %body.decode('utf8'))
24     ch.basic_ack(delivery_tag=method.delivery_tag)
25     #配套生产者的消息持久化，做用：让rabbitmq在一条消息被接受执行完成后将该条消息删除
26 
27 #同时最多取prefetch_count个消息，这里为1.做用：负载均衡
28 channel.basic_qos(prefetch_count=1)
29 
30 #no_ack=True 表示不用确认，queue='message'表示监听message这个队列
31 channel.basic_consume(callback,queue='message',no_ack=True)
32 
33 print(" [*] Waiting for messages.To exit press CTRL+C")
34 channel.start_consuming() #监听通道并阻塞

rabbit_recieve

 1 #!/usr/bin/env python
 2 # -*- coding:utf8 -*-
 3 
 4 import pika
 5 
 6 connection = pika.BlockingConnection(pika.ConnectionParameters(
 7     'localhost'))  #默认链接guest用户，也有密码
 8 
 9 #链接有密码的用户
10 user_pass = pika.PlainCredentials('txowner','helloworld')
11 connection = pika.BlockingConnection(pika.ConnectionParameters(
12     host='localhost',
13     port=5672,
14     virtual_host='/',
15     credentials=user_pass
16 ))
17 
18 channel = connection.channel()
19 
20 #声明queue，默认非持久化
21 # channel.queue_declare(queue='hello')
22 
23 #声明持久化的队列,durable是把队列那根管道持久化，不是队列里面的消息
24 channel.queue_declare(queue='message',durable=True)
25 
26 # RabbitMQ a message can never be sent directly to the queue,
27 # it always needs to go through an exchange
28 '''
29 #若同时开了多个接收方，这样只有一个接收方能收到
30 channel.basic_publish(exchange='',
31                           routing_key='hello',
32                           body='hello world!')
33     print(" [x] send 'hello world!' ")
34 '''
35 #这种状况下，会平均把三条消息轮询发给三个接收方(若至少要三个接收方)
36 for i in range(3):
37     channel.basic_publish(exchange='',
38                           routing_key='message',
39                           body='hello world! %s' %i,
40                           properties=pika.BasicProperties(delivery_mode=2)
41                           #这句话是把消息设置持久化
42                           )
43     print(" [x] send 'hello world!' ")
44 connection.close()

rabbitmq_send

修改Exchange的类型，以决定究竟是哪些Queue符合条件，能够接收消息：

 channel.basic_publish(exchange='',     #在这里指定
                          routing_key='message',
                          body='hello world!'
                          )

一、fanout: 全部bind到此exchange的queue均可以接收消息：

 1 #!/usr/bin/env python
 2 # -*- coding:utf8 -*-
 3 
 4 import pika
 5 import sys
 6 
 7 connection = pika.BlockingConnection(pika.ConnectionParameters(
 8     host='127.0.0.1'))
 9 channel = connection.channel()
10 
11 channel.exchange_declare(exchange='logs', type='fanout')
12 
13 message = ' '.join(sys.argv[1:]) or "info: Hello World!"
14 channel.basic_publish(exchange='logs',
15                       routing_key='',  #fanout是绑定到此exchange的队列都能接收到消息，因此不指定队列名
16                       body=message)
17 print(" [x] Sent %r" % message)
18 connection.close()

rabbitmq_fanout_publisher

 1 #!/usr/bin/env python
 2 # -*- coding:utf8 -*-
 3 
 4 import pika
 5 
 6 connection = pika.BlockingConnection(pika.ConnectionParameters(
 7     host='127.0.0.1'))
 8 channel = connection.channel()
 9 
10 channel.exchange_declare(exchange='logs',type='fanout')
11 
12 result = channel.queue_declare(exclusive=True)  # 不指定queue名字,rabbit会随机分配一个名字,exclusive=True会在使用此queue的消费者断开后,自动将queue删除
13 queue_name = result.method.queue #生成一个随机的queue
14 
15 channel.queue_bind(exchange='logs',queue=queue_name)
16 
17 print(' [*] Waiting for logs. To exit press CTRL+C')
18 
19 
20 def callback(ch, method, properties, body):
21     print(" [x] %r" % body)
22 
23 
24 channel.basic_consume(callback,
25                       queue=queue_name,
26                       no_ack=True)
27 
28 channel.start_consuming()

rabbitmq_fanout_subscriber

二、direct: 经过routingKey和exchange决定的那个惟一的queue能够接收消息：

 1 #!/usr/bin/env python
 2 # -*- coding:utf8 -*-
 3 
 4 '''有选择的发布消息'''
 5 
 6 import pika
 7 import sys
 8 
 9 connection = pika.BlockingConnection(pika.ConnectionParameters(
10     host='127.0.0.1'))
11 channel = connection.channel()
12 
13 channel.exchange_declare(exchange='direct_logs', type='direct')
14 
15 severity = sys.argv[1] if len(sys.argv) > 1 else 'info'
16 message = ' '.join(sys.argv[2:]) or 'Hello World!'
17 channel.basic_publish(exchange='direct_logs',
18                       routing_key=severity,
19                       body=message)
20 print(" [x] Sent %r:%r" % (severity, message))
21 connection.close()

rabbitmnq_direct_publisher

 1 #!/usr/bin/env python
 2 # -*- coding:utf8 -*-
 3 
 4 import pika
 5 import sys
 6 
 7 connection = pika.BlockingConnection(pika.ConnectionParameters(
 8     host='127.0.0.1'))
 9 channel = connection.channel()
10 
11 channel.exchange_declare(exchange='direct_logs', type='direct')
12 
13 result = channel.queue_declare(exclusive=True) # 不指定queue名字,rabbit会随机分配一个名字,exclusive=True会在使用此queue的消费者断开后,自动将queue删除
14 
15 queue_name = result.method.queue
16 
17 severities = sys.argv[1:]
18 if not severities:
19     sys.stderr.write("Usage: %s [info] [warning] [error]\n" % sys.argv[0])
20     sys.exit(1)
21 
22 for severity in severities:
23     channel.queue_bind(exchange='direct_logs',
24                        queue=queue_name,
25                        routing_key=severity)
26 
27 print(' [*] Waiting for logs. To exit press CTRL+C')
28 
29 
30 def callback(ch, method, properties, body):
31     print(" [x] %r:%r" % (method.routing_key, body))
32 
33 
34 channel.basic_consume(callback,
35                       queue=queue_name,
36                       no_ack=True)
37 
38 channel.start_consuming()

rabbitmq_direct_subscriber

三、topic:全部符合routingKey(此时能够是一个表达式)的routingKey所bind的queue能够接收消息

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

 1 #!/usr/bin/env python
 2 # -*- coding:utf8 -*-
 3 
 4 import pika
 5 import sys
 6 
 7 connection = pika.BlockingConnection(pika.ConnectionParameters(
 8     host='127.0.0.1'))
 9 channel = connection.channel()
10 
11 channel.exchange_declare(exchange='topic_logs', type='topic')
12 
13 routing_key = sys.argv[1] if len(sys.argv) > 1 else 'anonymous.info'
14 message = ' '.join(sys.argv[2:]) or 'Hello World!'
15 
16 channel.basic_publish(exchange='topic_logs',
17                       routing_key=routing_key,
18                       body=message)
19 
20 print(" [x] Sent %r:%r" % (routing_key, message))

rabbitmq_topic_publisher

 1 #!/usr/bin/env python
 2 # -*- coding:utf8 -*-
 3 
 4 import pika
 5 import sys
 6 
 7 connection = pika.BlockingConnection(pika.ConnectionParameters(
 8     host='127.0.0.1'))
 9 channel = connection.channel()
10 
11 channel.exchange_declare(exchange='topic_logs', type='topic')
12 
13 result = channel.queue_declare(exclusive=True)
14 queue_name = result.method.queue
15 
16 binding_keys = sys.argv[1:]
17 if not binding_keys:
18     sys.stderr.write("Usage: %s [binding_key]...\n" % sys.argv[0])
19     sys.exit(1)
20 
21 for binding_key in binding_keys:
22     channel.queue_bind(exchange='topic_logs',
23                        queue=queue_name,
24                        routing_key=binding_key)
25 
26 print(' [*] Waiting for logs. To exit press CTRL+C')
27 
28 
29 def callback(ch, method, properties, body):
30     print(" [x] %r:%r" % (method.routing_key, body))
31 
32 
33 channel.basic_consume(callback,
34                       queue=queue_name,
35                       no_ack=True)
36 
37 channel.start_consuming()

rabbitmq_topic_subscriber

四、headers: 经过headers 来决定把消息发给哪些queue

RPC

 1 #!/usr/bin/env python
 2 # -*- coding:utf8 -*-
 3 
 4 import pika
 5 
 6 connection = pika.BlockingConnection(pika.ConnectionParameters(
 7     host='127.0.0.1'))
 8 
 9 channel = connection.channel()
10 
11 channel.queue_declare(queue='rpc_queue')
12 
13 
14 def fib(n):
15     if n == 0:
16         return 0
17     elif n == 1:
18         return 1
19     else:
20         return fib(n - 1) + fib(n - 2)
21 
22 
23 def on_request(ch, method, props, body):
24     n = int(body)
25 
26     print(" [.] fib(%s)" % n)
27     response = fib(n)
28 
29     ch.basic_publish(exchange='',
30                      routing_key=props.reply_to,
31                      properties=pika.BasicProperties(correlation_id= \
32                                                 props.correlation_id),
33                      body=str(response))
34     ch.basic_ack(delivery_tag=method.delivery_tag)
35 
36 
37 channel.basic_qos(prefetch_count=1)
38 channel.basic_consume(on_request, queue='rpc_queue')
39 
40 print(" [x] Awaiting RPC requests")
41 channel.start_consuming()

rabbitmq_rpc_server

 1 #!/usr/bin/env python
 2 # -*- coding:utf8 -*-
 3 
 4 import pika
 5 import uuid
 6 
 7 
 8 class FibonacciRpcClient(object):
 9     def __init__(self):
10         self.connection = pika.BlockingConnection(pika.ConnectionParameters(
11             host='127.0.0.1'))
12 
13         self.channel = self.connection.channel()
14 
15         result = self.channel.queue_declare(exclusive=True)
16         self.callback_queue = result.method.queue
17 
18         self.channel.basic_consume(self.on_response,
19                                    no_ack=True,
20                                    queue=self.callback_queue
21                                    )
22 
23     def on_response(self, ch, method, props, body):
24         if self.corr_id == props.correlation_id:
25             self.response = body
26 
27     def call(self, n):
28         self.response = None
29         self.corr_id = str(uuid.uuid4())
30         self.channel.basic_publish(exchange='',
31                                    routing_key='rpc_queue',
32                                    properties=pika.BasicProperties(
33                                        reply_to=self.callback_queue,
34                                        correlation_id=self.corr_id,
35                                    ),
36                                    body=str(n)
37                                 )
38 
39         while self.response is None:
40             self.connection.process_data_events()   #不断去轮询，看是否有数据返回
41         return int(self.response)
42 
43 
44 fibonacci_rpc = FibonacciRpcClient()
45 
46 print(" [x] Requesting fib(30)")
47 response = fibonacci_rpc.call(30)
48 print(" [.] Got %r" % response)

rabbitmq_rpc_client

使用rabbitmq注意点：

#!/usr/bin/env python
# -*- coding:utf8 -*-

'''
rabbitmq在非持久化状况下：
    生产者和消费者不须要像redis同样同时在线，只要rabbitmq服务器没有重启或关闭，
生成的队列就一直在。此时，若生产者发送数据后，数据就会存放在队列里面，消费者一上线就
能接收到。

rabbitmq在持久化状况下：
    一、只是生产者持久化管道队列，生产者发送数据存在管道中时，重启rabbitmq服务器将会
      使消息丢失，可是管道队列仍是存在，此时能够不用再声明，能够直接向该队列存消息
    二、同时持久化队列和消息，那么不管服务器是否重启，只要生产者发送消息到队列里面，
      消费者一上线就能接收到

一、rabbitmq默认消息不是持久化的，因此当rabbitmq-server关闭后，生产者发送到消息会丢失
二、若要rabbitmq-server重启后消息不丢失，则须要把队列设置为持久化队列
三、当一个队列被设置为非持久化(默认)时，不能直接修改成持久化队列，有两种方式：
    (1) 删除该队列从新设置为持久化队列
    (2) 从新设置其余队列(不能与已存在队列重名)

四、声明持久化的队列
(这里是把队列设置为持久化，因此重启服务器后，队列还在。并不会把消息也持久化)
# channel.queue_declare(queue='message',durable=True)
  
  声明非持久化队列
# channel.queue_declare(queue='message')

  消息持久化(在发送消息时，在消息后面一个参数设置这个)
    properties=pika.BasicProperties(delivery_mode=2)
   eg： 
    server:
    channel.basic_publish(exchange='',
                          routing_key='message',
                          body='hello world!',
                          properties=pika.BasicProperties(delivery_mode=2)
                          )
    
    client:
    同时在客户端要在回调函数里面设置这个：
    ch.basic_ack(delivery_tag=method.delivery_tag)
    做用：在客户端消费消息后，通知rabbitmq将该条消息删除
    
    def callback(ch,method,properties,body):
        print(" [x] Received %r " %body.decode('utf8'))
        ch.basic_ack(delivery_tag=method.delivery_tag)
'''