pythonRabbitMQ使⽤详细介绍（⼩结）
上节回顾
主要讲了协程、进程、异步IO多路复⽤。
协程和IO多路复⽤都是单线程的。
epoll  在linux下通过这个模块libevent.so实现
gevent  在底层也是⽤了libevent.so
gevent可以理解为⼀个更上层的封装。
使⽤select或者selectors，每接收或发送数据⼀次都要select⼀次
twisted异步⽹络框架，强⼤⼜庞⼤，不⽀持python3 (代码量python中排top3)。⼏乎把所有的⽹络服务都重写了⼀遍。
⼀、RabbitMQ 消息队列介绍
RabbitMQ也是消息队列，那RabbitMQ和之前python的Queue有什么区别么？
py 消息队列:
线程 queue（同⼀进程下线程之间进⾏交互）
进程 Queue（⽗⼦进程进⾏交互或者同属于同⼀进程下的多个⼦进程进⾏交互）
如果是两个完全独⽴的python程序，也是不能⽤上⾯两个queue进⾏交互的，或者和其他语⾔交互有哪些实现⽅式呢。
【Disk、Socket、其他中间件】这⾥中间件不仅可以⽀持两个程序之间交互，可以⽀持多个程序，可以维护好多个程序的队列。
像这种公共的中间件有好多成熟的产品：
RabbitMQ
ZeroMQ
ActiveMQ
……
RabbitMQ：erlang语⾔开发的。
Python中连接RabbitMQ的模块：pika 、Celery(分布式任务队列) 、haigha
可以维护很多的队列
⼏个概念说明：
Broker：简单来说就是消息队列服务器实体。
Exchange：消息交换机，它指定消息按什么规则，路由到哪个队列。
Queue：消息队列载体，每个消息都会被投⼊到⼀个或多个队列。
Binding：绑定，它的作⽤就是把exchange和queue按照路由规则绑定起来。
Routing Key：路由关键字，exchange根据这个关键字进⾏消息投递。
vhost：虚拟主机，⼀个broker⾥可以开设多个vhost，⽤作不同⽤户的权限分离。
producer：消息⽣产者，就是投递消息的程序。
consumer：消息消费者，就是接受消息的程序。
channel：消息通道，在客户端的每个连接⾥，可建⽴多个channel，每个channel代表⼀个会话任务
⼆、RabbitMQ基本⽰例.
1、Rabbitmq 安装
ubuntu系统
install rabbitmq-server # 直接搞定
以下centos系统
1）Install Erlang
# For EL5:
rpm -Uvh /pub/epel/5/arch.rpm
# For EL6:
rpm -Uvh /pub/epel/6/arch.rpm
# For EL7:
rpm -Uvh /pub/epel/7/x86_64/arch.rpm
yum install erlang
2）Install RabbitMQ Server
rpm --import www.rabbitmq/rabbitmq-release-signing-key.asc
yum install rabbitmq-server-3.arch.rpm
3）use RabbitMQ Server
chkconfig rabbitmq-server on
service rabbitmq-server stop/start
2、基本⽰例
发送端 producer
import pika
# 建⽴⼀个实例
connection = pika.BlockingConnection(
pika.ConnectionParameters('localhost',5672) # 默认端⼝5672，可不写
)
# 声明⼀个管道，在管道⾥发消息
channel = connection.channel()
# 在管道⾥声明queue
channel.queue_declare(queue='hello')
# RabbitMQ a message can never be sent directly to the queue, it always needs to go through an exchange. channel.basic_publish(exchange='',
routing_key='hello', # queue名字
body='Hello World!') # 消息内容
print(" [x] Sent 'Hello World!'")
connection.close() # 队列关闭
接收端 consumer
import pika
import time
# 建⽴实例
connection = pika.BlockingConnection(pika.ConnectionParameters(
'localhost'))
# 声明管道
channel = connection.channel()
# 为什么⼜声明了⼀个‘hello'队列？
# 如果确定已经声明了，可以不声明。但是你不知道那个机器先运⾏，所以要声明两次。
channel.queue_declare(queue='hello')
def callback(ch, method, properties, body): # 四个参数为标准格式
print(ch, method, properties) # 打印看⼀下是什么
# 管道内存对象内容相关信息后⾯讲
print(" [x] Received %r" % body)
time.sleep(15)
ch.basic_ack(delivery_tag = method.delivery_tag) # 告诉⽣成者，消息处理完成
channel.basic_consume( # 消费消息
callback, # 如果收到消息，就调⽤callback函数来处理消息
queue='hello', # 你要从那个队列⾥收消息
# no_ack=True # 写的话，如果接收消息，机器宕机消息就丢了
# ⼀般不写。宕机则⽣产者检测到发给其他消费者
)
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming() # 开始消费消息
3、RabbitMQ 消息分发轮询
上⾯的只是⼀个⽣产者、⼀个消费者，能不能⼀个⽣产者多个消费者呢？
可以上⾯的例⼦，多启动⼏个消费者consumer，看⼀下消息的接收情况。
采⽤轮询机制；把消息依次分发
假如消费者处理消息需要15秒，如果当机了，那这个消息处理明显还没处理完，怎么处理？
（可以模拟消费端断了，分别注释和不注释 no_ack=True 看⼀下）
你没给我回复确认，就代表消息没处理完。
上⾯的效果消费端断了就转到另外⼀个消费端去了，但是⽣产者怎么知道消费端断了呢？
因为⽣产者和消费者是通过socket连接的，socket断了，就说明消费端断开了。
上⾯的模式只是依次分发，实际情况是机器配置不⼀样。怎么设置类似权重的操作？
RabbitMQ怎么办呢，RabbitMQ做了简单的处理就能实现公平的分发。
就是RabbitMQ给消费者发消息的时候检测下消费者⾥的消息数量，如果超过指定值（⽐如1条），就不给你发了。只需要在消费者端，channel.basic_consume前加上就可以了。
channel.basic_qos(prefetch_count=1) # 类似权重，按能⼒分发，如果有⼀个消息，就不在给你发
channel.basic_consume( # 消费消息
三、RabbitMQ 消息持久化（durable、properties）
1、RabbitMQ 相关命令
rabbitmqctl list_queues # 查看当前queue数量及queue⾥消息数量
2、消息持久化
如果队列⾥还有消息，RabbitMQ 服务端宕机了呢？消息还在不在？
把RabbitMQ服务重启，看⼀下消息在不在。
上⾯的情况下，宕机了，消息就久了，下⾯看看如何把消息持久化。
每次声明队列的时候，都加上durable，注意每个队列都得写，客户端、服务端声明的时候都得写。
# 在管道⾥声明queue
channel.queue_declare(queue='hello2', durable=True)
测试结果发现，只是把队列持久化了，但是队列⾥的消息没了。
durable的作⽤只是把队列持久化。离消息持久话还差⼀步：
发送端发送消息时，加上properties
properties=pika.BasicProperties(
delivery_mode=2, # 消息持久化
)
发送端 producer
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(
'localhost',5672)) # 默认端⼝5672，可不写
channel = connection.channel()
#声明queue
channel.queue_declare(queue='hello2', durable=True) # 若声明过，则换⼀个名字
#n RabbitMQ a message can never be sent directly to the queue, it always needs to go through an exchange.
channel.basic_publish(exchange='',
routing_key='hello2',
body='Hello World!',
properties=pika.BasicProperties(
delivery_mode=2, # make message persistent
)
)
print(" [x] Sent 'Hello World!'")
connection.close()
接收端 consumer
import pika
import time
connection = pika.BlockingConnection(pika.ConnectionParameters(
'localhost'))
channel = connection.channel()
channel.queue_declare(queue='hello2', durable=True)
def callback(ch, method, properties, body):
print(" [x] Received %r" % body)
time.sleep(10)
ch.basic_ack(delivery_tag = method.delivery_tag) # 告诉⽣产者，消息处理完成
channel.basic_qos(prefetch_count=1) # 类似权重，按能⼒分发，如果有⼀个消息，就不在给你发
channel.basic_consume( # 消费消息
callback, # 如果收到消息，就调⽤callbackactivemq默认端口
queue='hello2',
# no_ack=True # ⼀般不写，处理完接收处理结果。宕机则发给其他消费者
)
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()
四、RabbitMQ ⼴播模式（exchange）
前⾯的效果都是⼀对⼀发，如果做⼀个⼴播效果可不可以，这时候就要⽤到exchange了
exchange必须精确的知道收到的消息要发给谁。exchange的类型决定了怎么处理，
类型有以下⼏种：
fanout: 所有绑定到此exchange的queue都可以接收消息
direct: 通过routingKey和exchange决定的那个唯⼀的queue可以接收消息
topic：所有符合routingKey(此时可以是⼀个表达式)的routingKey所bind的queue可以接收消息
1、fanout 纯⼴播、all
需要queue和exchange绑定，因为消费者不是和exchange直连的，消费者是连在queue上，queue绑定在exchange上，消费者只会在queu⾥度消息
|------------------------|
|      /—— queue <—|—> consumer1
producer —|—exchange1 <bind    |
\ |      \—— queue <—|—> consumer2
\-|-exchange2  ……    |
|------------------------|
发送端 publisher 发布、⼴播
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
# 注意：这⾥是⼴播，不需要声明queue
type='fanout')
# message = ' '.join(sys.argv[1:]) or "info: Hello World!"
message = "info: Hello World!"
channel.basic_publish(exchange='logs',
routing_key='', # 注意此处空，必须有
body=message)
print(" [x] Sent %r" % message)
connection.close()
接收端 subscriber 订阅
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
type='fanout')
# 不指定queue名字,rabbit会随机分配⼀个名字,exclusive=True会在使⽤此queue的消费者断开后,⾃动将queue删除result = channel.queue_declare(exclusive=True)
# 获取随机的queue名字
queue_name = hod.queue
print("random queuename:", queue_name)
channel.queue_bind(exchange='logs', # queue绑定到转发器上
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()
注意：⼴播，是实时的，收不到就没了，消息不会存下来，类似收⾳机。
2、direct 有选择的接收消息
接收者可以过滤消息，只收我想要的消息
发送端publisher
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
type='direct')
# 重要程度级别，这⾥默认定义为 info
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()
接收端subscriber
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
type='direct')
result = channel.queue_declare(exclusive=True)
queue_name = hod.queue
# 获取运⾏脚本所有的参数
severities = sys.argv[1:]
if not severities:
sys.stderr.write("Usage: %s [info] [warning] [error]\n" % sys.argv[0])
# 循环列表去绑定
for severity in severities:

版权声明：本站内容均来自互联网，仅供演示用，请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系QQ:729038198，我们将在24小时内删除。