真正的稳定，是自己不断成长，不断寻找新的空间。与其要稳定，不如开始拥抱这个变化的时代，让自己准备好。
python实践
【写在前面】：
只是真正的使用才能有深刻的理解。
【材料】：
【Step1】：确认之前安装的RabbiMQ环境能正常工作
浏览器打开http://localhost:15672/#/，登录。
【Step2】：对RabbitMQ 的基本认识
RabbitMQ可以简单理解为一个管道，一边是生产者负责向管道中添加消息，一边是消费者负责从管道中提取消息；
RabbitMQ简单理解
【Step3】：RabbitMQ常用的几种消息模式
一、最简单的模式：生产者-消费者模式
生产者发布一条消息，N个消费者在监听消息，最终会被一个消费者取走并执行的模式(按顺序获取数据)；
# -*- coding: utf-8 -*-
# @Time : 2020/3/18 22:52
# @Author : dashenN72
# 生产者
import pika
credentials = pika.PlainCredentials('guest', 'guest') # 配置RabbitMQ账号密码
# 连接RabbitMQ server
connection = pika.BlockingConnection(pika.ConnectionParameters(host='127.0.0.1', credentials=credentials))
channel = connection.channel() # 创建链接通道
channel.queue_declare(queue='queue_1', durable=True) # 声明一个名为queue_1的队列并持久化(durable=True)
# 向指定队列中发送数据
channel.basic_publish(exchange='',
routing_key='queue_1',
body="it's first message!"
)
connection.close()
# -*- coding: utf-8 -*-
# @Time : 2020/3/18 22:09
# @Author : dashenN72
# 消费者
import pika
credentials = pika.PlainCredentials('guest', 'guest') # 配置RabbitMQ账号密码
# 连接RabbitMQ server
connection = pika.BlockingConnection(pika.ConnectionParameters(host='127.0.0.1', credentials=credentials))
channel = connection.channel() # 创建链接通道
def callback(ch, method, properties, body):
print("Received msg: {}".format(body))
# 从aaa队列中取出一条消息，及时回调(auto_ack=True)，生产者会将该任务消息删除，同时该消息作为body值去调用callback方法处理
channel.basic_consume('queue_1', callback, auto_ack=True)
channel.start_consuming() # 消费消息并处于监听状态
- 执行生产者脚本，检查RabbitMQ manage中以下数据是否新增一条记录，说明队列中添加一条消息成功：
生产者成功添加一个队列存储消息
- 单击队列名称(queuq_1)，查看队列中包含的消息
队列中总共1条消息且待处理
- 执行消费者脚本，获取消息并做处理
队列中的消息被消费掉
返回callback方法处理结果
二、发布订阅模式，即群发模式
生产者发布一条消息，需要所有的消费者都去获取并处理消息；
- 对于该模式的理解我们需要知道以下几点：
(1). 消费者和队列是1对1的关系；
(2). RabbitMQ中可以存在N多个队列，那么到这里其实我们可以让生产者把一条消息发给N多的队列来实现群发模式；
(3). RabbitMQ提供交换器(Exchange)的中间组件，该组件上接收生产者发送的消息，下对接N多的队列，由Exchange和Queue的关系来控制消息具体发给哪些Queue，如下图所示
Exchange(1):Queue(N)关系
- 广播模式应用
# -*- coding: utf-8 -*-
# @Time : 2020/3/18 22:52
# @Author : dashenN72
# 生产者
import pika
credentials = pika.PlainCredentials('guest', 'guest') # 配置RabbitMQ账号密码
# 连接RabbitMQ server
connection = pika.BlockingConnection(pika.ConnectionParameters(host='127.0.0.1', credentials=credentials))
channel = connection.channel() # 创建链接通道
channel.exchange_declare(exchange='exchange_1', exchange_type='fanout') # 创建交换机并设置广播模式
# 向指定交换器中发送数据
channel.basic_publish(exchange='exchange_1',
routing_key='',
body="it's a exchange message!"
)
connection.close()
# -*- coding: utf-8 -*-
# @Time : 2020/3/18 22:09
# @Author : dashenN72
# 消费者
import pika
credentials = pika.PlainCredentials('guest', 'guest') # 配置RabbitMQ账号密码
# 连接RabbitMQ server
connection = pika.BlockingConnection(pika.ConnectionParameters(host='127.0.0.1', credentials=credentials))
channel = connection.channel() # 创建链接通道
def callback(ch, method, properties, body):
print("Received msg: {}".format(body))
# 设置三个消费者
for queue_name in ['queue_2', 'queue_3', 'queue_4']:
channel.queue_declare(queue=queue_name, durable=True) # 创建队列
channel.queue_bind(queue_name, exchange='exchange_1') # 绑定交换器和队列
channel.basic_consume(queue_name, callback, True)
print('queue:{} start...'.format(queue_name))
channel.start_consuming()
(1).执行生产者脚本；
Exchange
(2).执行消费者脚本；
消费者
备注：发现并没有得到消息，原因是因为广播模式生产者在发布完消息后就会删除该消息，所以消费者再去获取消息是得不到，那怎么办呢？消费者需要时时监听队列数据，才能保证数据不丢失；
(3).再执行一次生产者脚本，检查消费者监听的消息
image.png
三、指定发布模式
所谓指定发布模式，即指定消息发给哪些queue，那么对应的消费者才能拿到消息，该模式通过路由来建立交换器与queue之间的关联关系
# -*- coding: utf-8 -*-
# @Time : 2020/3/18 22:52
# @Author : dashenN72
# 生产者
import pika
credentials = pika.PlainCredentials('guest', 'guest') # 配置RabbitMQ账号密码
# 连接RabbitMQ server
connection = pika.BlockingConnection(pika.ConnectionParameters(host='127.0.0.1', credentials=credentials))
channel = connection.channel() # 创建链接通道
channel.exchange_declare(exchange='exchange_2', exchange_type='direct') # 创建交换机并设置指定发布模式
# 向指定交换器下的路由分发消息
channel.basic_publish(exchange='exchange_2',
routing_key='route_2',
body="it's exchange and queue_5 message!"
)
connection.close()
# -*- coding: utf-8 -*-
# @Time : 2020/3/18 22:09
# @Author : dashenN72
# 消费者
import pika
credentials = pika.PlainCredentials('guest', 'guest') # 配置RabbitMQ账号密码
# 连接RabbitMQ server
connection = pika.BlockingConnection(pika.ConnectionParameters(host='127.0.0.1', credentials=credentials))
channel = connection.channel() # 创建链接通道
def callback(ch, method, properties, body):
print("Received msg: {}".format(body))
# 设置2个消费者
channel.queue_declare(queue='queue_3', durable=True) # 创建队列
channel.queue_bind('queue_3', exchange='exchange_2', routing_key='route_3') # 绑定交换器和队列
channel.basic_consume('queue_3', callback, True)
print('queue:{} start...'.format('queue_3'))
channel.queue_declare(queue='queue_5', durable=True) # 创建队列
channel.queue_bind('queue_5', exchange='exchange_2', routing_key='route_2') # 绑定交换器和队列
channel.basic_consume('queue_5', callback, True)
print('queue:{} start...'.format('queue_5'))
channel.start_consuming()
最终的结果是：只有绑定了正确路由的queue才能收到消息，如下图所示
queue_5收到消息
四、模糊匹配模式
借助正则来模糊匹配交换器与队列()该模式实际就是对上面指定模式的模糊方式，即通过两个常用通配符和#，匹配唯一的1个单词，#匹配1个或多个单词，所谓单词即以.号分割的字符(如a.b是两个单词)，举例如下：
- 正则是many.*, 则many.people匹配该规则，many.chiness.people不匹配
- 正则是many.#, 则many.people和many.chiness.people都匹配
# -*- coding: utf-8 -*-
# @Time : 2020/3/18 22:52
# @Author : dashenN72
# 生产者
import pika
credentials = pika.PlainCredentials('guest', 'guest') # 配置RabbitMQ账号密码
# 连接RabbitMQ server
connection = pika.BlockingConnection(pika.ConnectionParameters(host='127.0.0.1', credentials=credentials))
channel = connection.channel() # 创建链接通道
channel.exchange_declare(exchange='exchange_3', exchange_type='topic') # 创建交换机并设置模糊发布模式
# 向指定交换器下的路由分发消息
channel.basic_publish(exchange='exchange_3',
routing_key='route.2.3',
body="it's topic message!"
)
connection.close()
# -*- coding: utf-8 -*-
# @Time : 2020/3/18 22:09
# @Author : dashenN72
# 消费者
import pika
credentials = pika.PlainCredentials('guest', 'guest') # 配置RabbitMQ账号密码
# 连接RabbitMQ server
connection = pika.BlockingConnection(pika.ConnectionParameters(host='127.0.0.1', credentials=credentials))
channel = connection.channel() # 创建链接通道
def callback(ch, method, properties, body):
print("Received msg: {}".format(body))
# 设置2个消费者
channel.queue_declare(queue='queue_6', durable=True) # 创建队列
channel.queue_bind('queue_6', exchange='exchange_3', routing_key='route.*') # 绑定交换器和队列
channel.basic_consume('queue_6', callback, True)
print('queue:{} start...'.format('queue_6'))
channel.queue_declare(queue='queue_7', durable=True) # 创建队列
channel.queue_bind('queue_7', exchange='exchange_3', routing_key='route.#') # 绑定交换器和队列
channel.basic_consume('queue_7', callback, True)
print('queue:{} start...'.format('queue_7'))
channel.start_consuming()
结果如下：
queue_7接收到了消息
