python怎样利用Rabbitmq

目次

一、Rabbitmq先容
二、Rabbitmq的利用场景
1、异步处置处罚
2、服务解耦
3、流量削峰
4、日记网络
5、发布订阅
6、任务调治
三、python怎样利用Rabbitmq
1、安装依靠
2、底子利用
3、消息确认
4、消息恒久化
5、公平调治
6、发布订阅
7、关键字发布

---

一、Rabbitmq先容

RabbitMQ是一个开源的消息中央件，基于AMQP（Advanced Message Queue Protocol，高级消息队列协议）协议实现。RabbitMQ被广泛应用于各种应用场景，如异步任务处置处罚、日记传输、实时消息推送等。在微服务架构中，RabbitMQ是一个常见的消息中央件选择，它可以资助服务之间实现解耦和异步通讯，进步体系的可扩展性和稳固性。RabbitMQ提供了一个简单的用户页面，用户可以监控和管理消息、队列、互换器、绑定等资源。通过管理界面，用户可以直观地相识体系的运行状态，并举行相应的设置和管理操纵。
二、Rabbitmq的利用场景

1、异步处置处罚

• 在Web应用中，当用户提交表单时，可以将表单处置处罚任务发送给RabbitMQ，由配景服务异步处置处罚，从而进步用户界面的相应速率。
  
• 在电商体系中，用户下单后，订单处置处罚、库存更新、付出关照等任务可以异步实行，制止壅闭主线程。
  

2、服务解耦

• 在微服务架构中，差别服务之间通过RabbitMQ举行通讯，可以低沉服务之间的耦合度，进步体系的可扩展性和可维护性。
  
• 当某个服务必要升级或维护时，可以通过RabbitMQ实现服务的平滑过渡，而不会影响其他服务的正常运行。
  

3、流量削峰

• 在高并发场景中，RabbitMQ可以作为一个缓冲层，吸收并存储大量的哀求，然后按照设定的速率将哀求转发给后端服务，从而制止后端服务因过载而瓦解。
  
• 通过RabbitMQ的限流和队列机制，可以有效地控制哀求的速率和数目，掩护后端服务的稳固性。
  

4、日记网络

• RabbitMQ可以用于网络分散在各个服务器上的日记信息，将它们会合到一个或多个日记处置处罚服务中，举行同一的分析和处置处罚。
  
• 通过RabbitMQ，可以实现日记的实时网络、分析和报警，进步体系的运维服从和故障排查本领。
  

5、发布订阅

• 在必要向多个客户端推送消息的场景中，如实时关照、消息推送等，可以利用RabbitMQ的Fanout互换器将消息广播给全部绑定的队列。
  
• 通过RabbitMQ的消息广播机制，可以实现实时、可靠的消息推送服务，进步用户体验。
  

6、任务调治

• RabbitMQ可以与其他任务调治框架（如Quartz）连合利用，实现定时任务、耽误任务等复杂任务调治需求。
  
• 通过RabbitMQ的任务调治功能，可以机动地控制任务的实行时间和频率，进步体系的主动化水平和运行服从。
  

三、python怎样利用Rabbitmq

Rabbitmq网址：https://www.rabbitmq.com/tutorials
1、安装依靠

安装第三方库pika：pip install -i https://pypi.tuna.tsinghua.edu.cn/simple pika

1. pip install -i https://pypi.tuna.tsinghua.edu.cn/simple pika

复制代码

2、底子利用

生产者模子代码如下

1. import pika
3. # 生产者模型
5. connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))  # 连接本地的rabbitmq
6. channel = connection.channel()  # 连接通道
7. channel.queue_declare(queue='list')  #  创建一个名叫list的队列
8. channel.basic_publish(exchange='',
9.                       routing_key='list',  # 向那个队列发布信息
10.                       body='lol',# 发布的信息)
11. connection.close()  # 关闭连接

复制代码

消耗者代码如下

1. import pika
3. # 消费者模型
5. connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))  # 连接本地的rabbitmq
6. channel = connection.channel()  # 连接通道
7. channel.queue_declare(queue='list')  #  为确保队列存在，创建一个list队列
9. def callback(ch, method, properties, body):
10.     print(f" [x] 消费者接收到了任务 {body}")
12. channel.basic_consume(queue='list',on_message_callback=callback,auto_ack=True)  # 接收信息：queue表示监听的队列，on_message_callback表示接收到信息执行的函数，auto_ack表示默认执行回复
13. channel.start_consuming()   # 开启永无止境的循环监听该队列

复制代码

3、消息确认

在队列中，实行任务大概必要几秒钟，您大概想知道如果 利用者启动一个长任务，并在完成之前制止。 利用我们当前的代码，一旦 RabbitMQ 将消息通报给消耗者，它就会 立即将其标记为删除。在这种环境下，如果您制止 worker，则它刚刚处置处罚的消息丢失了。调治的消息 对于这个尚未处置处罚的特定 worker 来说，也会丢失。为了确保消息永久不会丢失，RabbitMQ 支持消息确认。ack由 consumer 告诉 RabbitMQ 已收到特定消息， 处置处罚，而且 RabbitMQ 可以自由删除它。
这是由消耗者来举行改变的，代码如下：
必要将auto_ack改为False，然后在回调函数里到场ch.basic_ack(delivery_tag=method.delivery_tag)

1. import pika
3. # 消费者模型
5. connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))  # 连接本地的rabbitmq
6. channel = connection.channel()  # 连接通道
7. channel.queue_declare(queue='list')  #  创建一个队列
9. def callback(ch, method, properties, body):
10.     print(f" [x] 消费者接收到了任务 {body}")
11.     ch.basic_ack(delivery_tag=method.delivery_tag) # 给队列回复进行确认
13. channel.basic_consume(queue='list2',on_message_callback=callback,auto_ack=False)  # auto_ack改为False
14. channel.start_consuming()  # 开启永无止境的循环监听该队列

复制代码

4、消息恒久化

如果 RabbitMQ 服务器 制止。
当 RabbitMQ 退出或瓦解时，它会忘记队列和消息 除非你告诉它不要如许做。必要做两件事来确保 消息不会丢失：我们必要将队列和消息都标记为 耐用。然和把队列和消息生存在磁盘里。
这是有生产者来举行改变的，代码如下：
durable=True在创建队列时声明恒久化，delivery_mode=pika.DeliveryMode.Persistent让信息做恒久化

1. import pika
3. # 生产者模型
5. connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))  # 连接本地的rabbitmq
6. channel = connection.channel()  # 连接通道
7. channel.queue_declare(queue='list2',durable=True)  #  创建一个队列,durable=True表示队列支持持久化
8. channel.basic_publish(exchange='',
9.                       routing_key='list2',  # 向那个队列发布信息
10.                       body='lol',# 发布的信息
11.                       properties=pika.BasicProperties(
12.                           delivery_mode=pika.DeliveryMode.Persistent  # 让信息做持久化
13.                       ))
14. connection.close()  # 关闭连接

复制代码

5、公平调治

您大概已经注意到，调治仍旧没有完全工作 正如我们想要的那样。比方，在有两个 worker 的环境下，当全部 奇数消息很重，偶数消息很轻，一个 worker 将是 不停很忙，另一个险些不做任何工作。井 RabbitMQ 对此一无所知，仍会 dispatch 消息匀称。
发生这种环境是由于 RabbitMQ 只是在消息 进入队列。它不看未确认的数目 消息。它只是盲目地调治每 n 条消息 到第 n 个消耗者。
为了办理这个题目，我们可以利用带有 setup 的 channel 方法。它利用协议方法告诉 RabbitMQ 不要一次向 worker 提供多条消息。大概，换句话说，不要调治 向 worker 发送新消息，直到它处置处罚并确认 上一个。相反，它会将其分派给下一个仍旧不忙的 worker。
这是由消耗者来举行改变的，代码如下：
channel.basic_qos(prefetch_count=1)举行闲置派发

1. import pika
3. # 消费者模型
5. connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))  # 连接本地的rabbitmq
6. channel = connection.channel()  # 连接通道
7. channel.queue_declare(queue='list')  #  为确保队列存在，创建一个list队列
9. def callback(ch, method, properties, body):
10.     print(f" [x] 消费者接收到了任务 {body}")
12. channel.basic_qos(prefetch_count=1)  # 闲置派发
13. channel.basic_consume(queue='list',on_message_callback=callback,auto_ack=True)  # 接收信息：queue表示监听的队列，on_message_callback表示接收到信息执行的函数，auto_ack表示默认执行回复
14. channel.start_consuming()   # 开启永无止境的循环监听该队列

复制代码

6、发布订阅

发布订阅模式是将信息发布给全部的订阅者，其特点就是有互换机。
channel.exchange_declare(exchange='m1',exchange_type='fanout')声明一个互换机，范例为fanout，是将信息发给全部的订阅者
发布者代码如下：

1. import pika
3. connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))  # 连接rabbitmq
4. channel = connection.channel()  # 连接通道
6. channel.exchange_declare(exchange='m1',exchange_type='fanout')  # fanout：将信息发给所有的队列
7. channel.queue_declare(queue='')  #  创建一个队列,durable=True表示队列支持持久化
8. channel.basic_publish(exchange='m1',
9.                       routing_key='',  # 队列名称
10.                       body='lol',)# 发布的数据
11. connection.close()  # 关闭连接

复制代码

订阅者代码如下：

1. import pika
3. connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))  # 连接rabbitmq
4. channel = connection.channel()
6. channel.exchange_declare(exchange='m1',exchange_type='fanout',)  # fanout：将信息发给所有的队列
7. result=channel.queue_declare(exclusive=True,queue='')  #  创建一个随机队列
8. queue_name=result.method.queue  # 拿到队列名字
9. print(queue_name)
10. channel.queue_bind(exchange='m1',queue=queue_name)  # 对exchange和queue进行绑定
12. def callback(ch, method, properties, body):
13.     print(f" [x] 消费者接收到了任务 {body}")
15. channel.basic_consume(queue=queue_name,on_message_callback=callback,auto_ack=True)
16. channel.start_consuming()  # 开启永无止境的循环监听该队列

复制代码

7、关键字发布

关键字发布就是在发布订阅模式底子上，将差别信息发布给差别的订阅者。
channel.exchange_declare(exchange='m2',exchange_type='direct') ，声明一个互换机，互换机的范例为direct，将信息发布给指定的订阅者。
发布者代码如下：

1. import pika
3. connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))  # 连接rabbitmq
4. channel = connection.channel()  # 连接通道
6. channel.exchange_declare(exchange='m2',exchange_type='direct')  # direct：将信息发布给指定的订阅者
7. channel.queue_declare(queue='')  #  创建一个队列,durable=True表示队列支持持久化
8. channel.basic_publish(exchange='m2',
9.                       routing_key='hhq',  # 队列名称
10.                       body='cf',)# 发布的数据
11. connection.close()  # 关闭连接

复制代码

订阅者代码如下：

1. import pika
3. connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))  # 连接rabbitmq
4. channel = connection.channel()
6. channel.exchange_declare(exchange='m2',exchange_type='direct')  # exchange:交易所的名称，fanout：将信息发给所有的队列
7. result=channel.queue_declare(exclusive=True,queue='')  #  创建一个随机队列
8. queue_name=result.method.queue  # 拿到队列名字
9. print(queue_name)
10. channel.queue_bind(exchange='m2',queue=queue_name,routing_key='hhq')  # 对exchange和queue进行绑定
12. def callback(ch, method, properties, body):
13.     print(f" [x] 消费者接收到了任务 {body}")
15. channel.basic_consume(queue=queue_name,on_message_callback=callback,auto_ack=True)
16. channel.start_consuming()  # 开启永无止境的循环监听该队列

复制代码