rabbitmq使用

线程queue只能在同一个进程里面共享数据，进程Queue只能是在父进程与子进程间或者同一父进程的子进程中共享数据。就出现了rabbitmq类似于这样的第三方队列软件，它能同时维护多个队列。

RabbitMQ实现功能：

循环调度：将消息循环发送给不同的消费者

消息确认机制：为了确保一个消息不会丢失，RabbitMQ支持消息的确认。
　　　　　　  一个 ack(acknowlegement) 是从消费者端发送一个确认去告诉RabbitMQ 消息已经接收了、处理了RabbitMQ可以释放并删除掉了。
　　　　　　　如果一个消费者死掉了（channel关闭、connection关闭、或者TCP连接断开了）而没有发送ack，RabbitMQ 就会认为这个消息没有被消费者处理
　　　　　　　并会重新发送到生产者的队列里，如果同时有另外一个消费者在线，rabbitmq将会将消息很快转发到另外一个消费者中。
            那样的话你就能确保虽然一个消费者死掉，但消息不会丢失。
        
消息持久化：将消息写入硬盘中。RabbitMQ不允许你重新定义一个已经存在、但属性不同的queue。
　　　　　　需要标记消息为持久化的 - 要通过设置 delivery_mode 属性为 2来实现。

公平调度：在一个消费者未处理完一个消息之前不要分发新的消息给它，而是将这个新消息分发给另一个不是很忙的消费者进行处理。
　　　　　为了解决这个问题我们可以在消费者代码中使用 channel.basic.qos ( prefetch_count = 1 )，将消费者设置为公平调度。

RabbitMQ注意点：

没有超时处理，当消费方（consumer）死掉后RabbitMQ会重新转发消息，即使处理这个消息需要很长很长时间也没有问题。消息的 acknowlegments 默认是打开的，关闭是： no_ack = True .

标记消息为持久化并不能完全保证消息不会丢失，尽管已经告诉RabbitMQ将消息保存到磁盘，但RabbitMQ接收到的消息在还没有保存的时候，仍然有一个短暂的时间窗口。
RabbitMQ不会对每个消息都执行同步 --- 可能只是保存到缓存cache还没有写入到磁盘中。因此这个持久化保证并不是很强，但这比我们简单的任务queue要好很多，如果想要很强的持久化保证，可以使用 publisher confirms。

Python中通过pika来调度RabbitMQ使用

Pika是AMQP 0-9-1协议的纯Python实现，试图保持与底层网络支持库相当独立

pika适配器：

 BlockingConnection - 启用对库进行阻塞，同步操作以进行简单的使用
 LibevConnection - 用于libev事件循环http://libev.schmorp.de的适配器
 SelectConnection - 快速异步适配器
 TornadoConnection - 适用于Tornado IO Loop的适配器http://tornadoweb.org
 TwistedConnection - 用于Twisted异步包的适配器http://twistedmatrix.com/

简单的pika实现rabbitmq调用的方式（生产者、消费者模型）

send端

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host="localhost"))  # 创建连接
channel = connection.channel()  # 创建管道用于通信

channel.queue_declare(queue="hello")  # 声明一个队列

channel.basic_publish(exchange="",
                      routing_key="hello",  # routing_key: 队列名,与上面声明的队列名要保持一致
                      body="Hello World!")   # body: 要发送的消息

print("[x] Sent 'Hello World!'")

connection.close()  # 关闭连接

receive端

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host="localhost"))  # 创建连接

channel = connection.channel()  # 创建管道

channel.queue_declare(queue="hello")  # 声明队列,如果确认这个队列存在,可以不声明（无法确认生产者与消费者哪个先执行,所以两边都要声明）

def callback(ch,method,properties,body): print("ch:",ch) # ch为创建的连接 print("method:",method, "properties:",properties) print("[x] Received %r" % body)

    print("method.delivery_tag:",method.delivery_tag)
    
    # rabbitmq不会去释放没有得到回应的信息
    ch.basic_ack(delivery_tag=method.delivery_tag)  # 执行完后确认,client执行完后给rabbitmq返回的一个标识,收到这个标识后rabbitmq认为这个消息处理完了,不会在重复发送给其他client继续执行

channel.basic_consume(callback,  # 回调函数
                      queue="hello",  # 指定消息队列名
                      no_ack=False)  # 是否确认,True是不确认,默认是False
                    # （如果为False,没有收到确认信息就会在处理的那个连接断开之后重新发送给其他的client）

print("[*] Waiting for messages. To exit press CTRL+C")

channel.start_consuming()   # 开始消费队列,一直收取消息,没有的话就会卡住，有多个消费队列在取消息时，会以轮询的方式取消息

RabbitMQ现在队列、消息都是保存在内存中，下面来看持久化方法（produce端）

import pika

user_pwd = pika.PlainCredentials(username="alex", password="123456")  # 使用用户名、密码进行认证

connection = pika.BlockingConnection(pika.ConnectionParameters(host="localhost",credentials=user_pwd))  # credentials：凭证
channel = connection.channel()  # 创建管道用于通信

channel.queue_declare(queue="hello",durable=True)  # durable=True持久化队列

channel.basic_publish(exchange="",
                      routing_key="hello",  # routing_key: 队列名,与上面声明的队列名要保持一致
                      body="Hello World!",  # body: 要发送的消息
                      properties=pika.BasicProperties(
                          delivery_mode=2))  # properties=pika.BasicProperties(delivery_mode=2) 消息持久化
                      # 如果队列没有设置durable=True的话消息是没有办法持久化的

print("[x] Sent 'Hello World!'")

connection.close()  # 关闭连接

持久化后重启队列与消息依然存在：

远程连接rabbitmq server方法

首先在rabbitmq server上创建一个用户
	
sudo rabbitmqctl add_user caoy 123456　　

同时还要配置权限，允许从外面访问
	
sudo rabbitmqctl set_permissions -p / alex ".*" ".*" ".*"

set_permissions [-p vhost] {user} {conf} {write} {read}

vhost：用户访问主机权限,默认是: /

user: 指定访问的用户

conf：文件配置权限

write：写权限

read：读权限

客户端连接的时候需要配置认证参数

credentials = pika.PlainCredentials('alex', 'alex3714')

connection = pika.BlockingConnection(pika.ConnectionParameters(
    '10.211.55.5',5672,'/',credentials))

channel = connection.channel()

公平调度：

如果Rabbit只管按顺序把消息发到各个消费者身上，不考虑消费者负载的话，很可能出现，一个机器配置不高的消费者那里堆积了很多消息处理不完，同时配置高的消费者却一直很轻松。为解决此问题，可以在各个消费者端，配置perfetch=1,告诉RabbitMQ在我这个消费者当前消息还没处理完的时候就不要再给我发新消息了。（每个消费者当前只处理一个消息）

 1 import pika
 2 import sys
 3 
 4 user_pwd = pika.PlainCredentials("alex","123456")
 5 
 6 connection = pika.BlockingConnection(pika.ConnectionParameters(host="10.10.1.1",credentials=user_pwd))
 7 
 8 channel = connection.channel()
 9 
10 channel.queue_declare(queue="caoy",durable=True)
11 
12 message = " ".join(sys.argv[1:]) or " hello caoy"
13 
14 channel.basic_publish(exchange="",
15                       routing_key="caoy",
16                       body=message,
17                       properties=pika.BasicProperties(
18                           delivery_mode=2,
19                       ))
20 
21 print("[x] Sent %r" % message)
22 
23 connection.close()

公平调度--produce

 1 import pika
 2 import time
 3 
 4 user_pwd = pika.PlainCredentials(username="caoy",password="123456")
 5 
 6 connection = pika.BlockingConnection(pika.ConnectionParameters(host="120.24.218.137",credentials=user_pwd))
 7 
 8 channel = connection.channel()
 9 
10 channel.queue_declare(queue="caoy",durable=True)
11 
12 print("[*] Waiting for message. To exit press CRLT+C")
13 
14 
15 def caoy(ch,method,properties,body):
16     print("[x] Received %r" % body)
17     # time.sleep(body.count(b"."))
18     time.sleep(5)
19     print("[x] Done")
20     ch.basic_ack(delivery_tag=method.delivery_tag)
21 
22 channel.basic_qos(prefetch_count=1)  # 公平调度,设置后client端只能有一个消息在执行,当消息没执行完时,不接收新消息
23 
24 channel.basic_consume(caoy,
25                       queue="caoy")
26 
27 channel.start_consuming()

公平调度--coummer

消息发布订阅（Exchange）

前面讲的都是1对1消息发送和接收,即消息是只能发送到指定的queue里,但如果想让你的消息被多个Queue收到,类似广播的效果,这时候要用到exchange

Exchange在定义的时候是有类型的,决定哪些Queue符合条件,可以接收消息
	fanout: 所有bind到此exchange的queue都可以接收消息

	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

注意：消息发布是实时的，如果recvied当时没在就不能接收消息，不会把消息存储起来

fanout类型：广播发送，exchange转发到符合要求的queue中

 1 import pika
 2 import sys
 3 
 4 conn = pika.BlockingConnection(pika.ConnectionParameters(
 5     host="localhost"))
 6 
 7 channel = conn.channel()
 8 
 9 # exchange_declare: 声明exchange（转发器）类型
10 channel.exchange_declare(exchange="logs",  # 队列需绑定exchange="logs"
11                          exchange_type="fanout")   # 绑定到这个exchange的就可以收到发布的消息
12 
13 message = " ".join(sys.argv[1:]) or "info: Hello World!"
14 
15 channel.basic_publish(exchange="logs",
16                       routing_key="",  # 在exchange_type="fanout"类型下,忽略routing_key的值
17                       body=message)
18 
19 print("[x] Sent %r" % message)
20 
21 conn.close()

emit_log.py

 1 import pika
 2 
 3 conn = pika.BlockingConnection(pika.ConnectionParameters(
 4     host="localhost"))
 5 
 6 channel = conn.channel()
 7 
 8 # 声明exchange类型
 9 channel.exchange_declare(exchange="logs",
10                          exchange_type="fanout")
11 
12 # 不指定queue名,让rabbitmq随机生成一个,exclusive=True会在使用此queue的消费者断开后,自动将queue删除
13 result = channel.queue_declare(exclusive=True)
14 
15 queue_name = result.method.queue  # 获取随机生成的queue名
16 
17 # queue绑定exchange
18 channel.queue_bind(exchange="logs",
19                    queue=queue_name)
20 
21 
22 def callback(ch,method,properties,body):
23     print("[x] %r" % body)
24 
25 channel.basic_consume(callback,
26                       queue=queue_name,
27                       no_ack=True)
28 
29 channel.start_consuming()

recived_logs.py

direct类型：(有选择的接收消息)关键字发送，即：队列绑定关键字，发送者将数据根据关键字发送到消息exchange，exchange根据关键字判定应该将数据发送至指定队列。

 1 import pika
 2 import sys
 3 
 4 conn = pika.BlockingConnection(pika.ConnectionParameters(host="localhost"))
 5 
 6 channel = conn.channel()
 7 
 8 # 声明exchange类型
 9 channel.exchange_declare(exchange="direct_logs",
10                          exchange_type="direct")
11 
12 severity = sys.argv[1] if len(sys.argv) > 2 else "info"  # 关键字(举例日志：info,error。。。)
13 
14 print("severity:",severity)
15 
16 message = " ".join(sys.argv[2:]) or "Hello World!"
17 
18 channel.basic_publish(exchange="direct_logs",
19                       routing_key=severity,   # 关键字传给routing_key
20                       body=message)
21 
22 print("[x] Sent %r:%r" % (severity,message))
23 
24 conn.close()

emit_log.py

 1 import pika
 2 import sys
 3 
 4 conn = pika.BlockingConnection(pika.ConnectionParameters(host="localhost"))
 5 
 6 channel = conn.channel()
 7 
 8 # 声明exchange类型
 9 channel.exchange_declare(exchange="direct_logs",
10                          exchange_type="direct")
11 
12 result = channel.queue_declare(exclusive=True)  # 随便生成一个queue名
13 
14 queue_name = result.method.queue  # 获取queue名方法
15 
16 severities = sys.argv[1:]  # 接收的关键字,可以有多个
17 
18 if not severities:
19     sys.stderr.write("Usage: %s [info] [warning] [error]
" % sys.argv[0])
20     sys.exit(1)
21 
22 for severity in severities:  
23     # 循环把所有关键字绑定到queue上
24     channel.queue_bind(exchange="direct_logs",
25                        queue=queue_name,
26                        routing_key=severity)
27 
28 print("Waiting for logs. To exit press CTRL+C")
29 
30 
31 def callback(ch,method,properties,body):
32     print("[x] %r:%r" %(method.routing_key,body))
33 
34 channel.basic_consume(callback,
35                       queue=queue_name,
36                       no_ack=True)
37 
38 channel.start_consuming()

recived_logs.py

topic类型：(更细致的消息过滤) routing_key能有多个值，最高是255个字节（queue中多个关键字匹配,也只会在queue中出现一次）

 1 import pika
 2 import sys
 3 
 4 conn = pika.BlockingConnection(pika.ConnectionParameters(
 5     host="localhost"))
 6 
 7 channel = conn.channel()
 8 
 9 # 声明exchange类型
10 channel.exchange_declare(exchange="topic_logs",
11                          exchange_type="topic")
12 
13 routing_key = sys.argv[1] if len(sys.argv) > 2 else "anonymous.info"
14 
15 message = " ".join(sys.argv[2:]) or "Hello World!"
16 
17 channel.basic_publish(exchange="topic_logs",
18                       routing_key=routing_key,
19                       body=message)
20 
21 print("[x] Sent %r:%r" %(routing_key,message))
22 
23 conn.close()

emit_log_topic.py

 1 import pika
 2 import sys
 3 
 4 conn = pika.BlockingConnection(pika.ConnectionParameters(
 5     host="localhost"))
 6 
 7 channel = conn.channel()
 8 
 9 # 声明exchange类型
10 channel.exchange_declare(exchange="topic_logs",
11                          exchange_type="topic")
12 
13 result = channel.queue_declare(exclusive=True)  # 生成一个随机queue名
14 
15 queue_name = result.method.queue  # 获取queue名
16 
17 binding_keys = sys.argv[1:]  # 要接收的关键字
18 
19 if not binding_keys:
20     sys.stderr.write("Usage: %s [binding_key]...
" % sys.argv[0])
21     sys.exit(1)
22 
23 for binding_key in binding_keys:
24     # loop绑定多个匹配的关键字在queue_name上
25     channel.queue_bind(exchange="topic_logs",
26                        queue=queue_name,
27                        routing_key=binding_key)  # routing_key要匹配的关键字
28 
29 print(' [*] Waiting for logs. To exit press CTRL+C')
30 
31 
32 def callback(ch,method,properties,body):
33     print("[x] %r:%r" % (method.routing_key,body))
34 
35 channel.basic_consume(callback,
36                       queue=queue_name,
37                       no_ack=True)
38 
39 channel.start_consuming()

recived_logs_topic.py

To receive all the logs run:

python receive_logs_topic.py "#"

To receive all logs from the facility "kern":

python receive_logs_topic.py "kern.*"

Or if you want to hear only about "critical" logs:

python receive_logs_topic.py "*.critical"

You can create multiple bindings:

python receive_logs_topic.py "kern.*" "*.critical"

And to emit a log with a routing key "kern.critical" type:

python emit_log_topic.py "kern.critical" "A critical kernel error"

RPC：（Remote Procedure Call Protocol）-远程过程调用协议，它是一种通过网络从远程计算机程序上请求服务，而不需要了解底层网络技术的协议

　　　　如果发送一条消息出去需要recived处理完后返回，这时候就要用到RPC协议

　　　　RPC 采用C/S模式。首先，客户机调用进程发送一个有进程参数的调用消息到服务端的消息队列里，然后等待应答信息。在服务器端，进程保持睡眠状态直到调用信息到达为止。当一个调用信息到达，服务器获得进程参数，计算结果，发送答复信息，然后等待下一个调用信息，最后，客户端调用进程接收答复信息，获得进程结果，然后调用执行继续进行。

properties参数:
	delivery_mode: 设置消息持久化(value = 2)
	content_type: 描述mime-type的编码, json编码可设置为:application/json
	reply_to: 告诉recived返回的结果放到哪个队列中
	correlation_id: 关联RPC请求与响应信息，当我们多个请求处理时，可以通过correlation_id来判断响应结果与发出的请求是否相对

 1 import pika
 2 
 3 con = pika.BlockingConnection(
 4     pika.ConnectionParameters(
 5         host="localhost"
 6     )
 7 )
 8 
 9 channel = con.channel()
10 
11 # 声明接收消息的队列
12 channel.queue_declare(queue="rpc_queue")
13 
14 # 处理的函数
15 def fib(n):
16     if n == 0:
17         return 0
18     elif n == 1:
19         return 1
20     else:
21         return fib(n-1) + fib(n-2)
22 
23 
24 # 消息过来触发的回调函数
25 def on_request(ch,method,props,body):
26     n = int(body)
27 
28     print("[.] fib(%s)" % n)
29     response = fib(n)
30 
31     # 处理后的结果发送
32     ch.basic_publish(exchange="",
33                      routing_key=props.reply_to,  # 结果要返回的queue名
34                      properties=pika.BasicProperties(
35                          correlation_id=props.correlation_id),  # 消息传过来时带的标识,与request对应的id
36                      body=str(response)
37                      )
38     ch.basic_ack(delivery_tag=method.delivery_tag)  # 处理完成后的确认值
39 
40 channel.basic_qos(prefetch_count=1)  # 公平调度
41 
42 # 消费消息的队列
43 channel.basic_consume(on_request,
44                       queue="rpc_queue")
45 
46 print("[x] Awaiting RPC requests")
47 
48 # 开始loop等待消息
49 channel.start_consuming()

RPC server 端

 1 import pika
 2 import uuid
 3 import time
 4 
 5 class FibonacciRpcClient(object):
 6     def __init__(self):
 7         self.conn = pika.BlockingConnection(
 8             pika.ConnectionParameters(
 9                 host="localhost"
10             )
11         )
12         self.channel = self.conn.channel()
13         result = self.channel.queue_declare(exclusive=True)  # 声明一个随机queue名
14         self.callback_queue = result.method.queue   # 获取随机queue名
15 
16         # server处理完消息返回的结果进行消费
17         self.channel.basic_consume(self.on_reponse,  # 回调函数
18                                    queue=self.callback_queue,  # 结果返回的queue名
19                                    no_ack=True)
20 
21     # server处理完消息返回的结果中执行的回调函数
22     def on_reponse(self,ch,method,props,body):
23         # self.corr_id: client端发消息时生成的uuid
24         # props.correlation_id： server端处理完返回结果时,返回client发送给server的uuid
25         if self.corr_id == props.correlation_id:  # 判断如果相等,request与response对应上
26             self.response = body
27 
28     # client发送消息去server的函数
29     def call(self,n):
30         self.response = None  # 请求resqust没有回应None
31         self.corr_id = str(uuid.uuid4()) # 生成uuid值
32 
33         # client发送消息
34         self.channel.basic_publish(
35             exchange="",
36             routing_key="rpc_queue",
37             properties=pika.BasicProperties(
38                 reply_to=self.callback_queue,  # server结果要返回到self.callback_queue中,也就是上面随机生成的那个列队
39                 correlation_id=self.corr_id   # 传给server端uuid值区分返回的值是否与request关联
40             ),
41             body=str(n)
42         )
43 
44         while self.response is None:  # 请求没有回应loop
45 
46             # self.conn.process_data_events():是一个等待消息的阻塞过程
47             # 连接的任何消息都可以使它脱离阻塞状态,相当于非阻塞版的start_consuming()
48             self.conn.process_data_events()  
49             time.sleep(0.5)
50             print("No message ....")
51 
52         return int(self.response)
53 
54 fibonacci_rpc = FibonacciRpcClient()
55 
56 print("[X] Requesting fib(8)")
57 
58 response = fibonacci_rpc.call(8)
59 
60 print(" [.] Got %r" % response)

PRC client 端

补充知识：

rabbitmq查询命令：

rabbitmqctl list_queues  # 查看队列中有多少个消息

# 查看队列中有多少个消息没处理、多少个消息没有回应
rabbitmqctl list_queues name messages_ready messages_unacknowledged  

rabbitmqctl list_exchanges  # 查看rabbitmq中的exchange

UUID介绍：

　uuid是128位的全局唯一标识符（univeral unique identifier），通常用32位的一个字符串的形式来表现。有时也称guid(global unique identifier)。python中自带了uuid模块来进行uuid的生成和管理工作。（具体从哪个版本开始有的不清楚。。）

　　python中的uuid模块基于信息如MAC地址、时间戳、命名空间、随机数、伪随机数来uuid。具体方法有如下几个：　　

　　uuid.uuid1()　　基于MAC地址，时间戳，随机数来生成唯一的uuid，可以保证全球范围内的唯一性。

　　uuid.uuid2()　　算法与uuid1相同，不同的是把时间戳的前4位置换为POSIX的UID。不过需要注意的是python中没有基于DCE的算法，所以python的uuid模块中没有uuid2这个方法。

　　uuid.uuid3(namespace,name)　　通过计算一个命名空间和名字的md5散列值来给出一个uuid，所以可以保证命名空间中的不同名字具有不同的uuid，但是相同的名字就是相同的uuid了

　　uuid.uuid4()　　通过伪随机数得到uuid，是有一定概率重复的

　　uuid.uuid5(namespace,name)　　和uuid3基本相同，只不过采用的散列算法是sha1