RabbitMq生产者消费者模型
生产者(producter) 队列消息的产生者,复制生产消息,并将消息传入队列
生产者代码:
![复制代码](https://common.cnblogs.com/images/copycode.gif)
import pika import json credentials = pika.PlainCredentials('admin','admin')#mq用户名和密码,用于认证 #虚拟队列需要指定参数virtual_host,如果是默认的可以不填 connection = pika.BlockingConnection(pika.ConnectionParameters(host='10.0.0.24',port=5672,virtual_host='/',credentials=credentials)) channel = connection.channel()# 创建一个AMQP信道 #声明队列,并设置durable为True,为了避免rabbitMq-server挂掉数据丢失,将durable设为True channel.queue_declare(queue='1',durable=True) for i in range(10): # 创建10个q message = json.dumps({'OrderId':"1000%s"%i}) # exchange表示交换器,可以精确的指定消息应该发到哪个队列中,route_key设置队列的名称,body表示发送的内容 channel.basic_publish(exchange='',routing_key='1',body=message) print(message) connection.close()
![复制代码](https://common.cnblogs.com/images/copycode.gif)
通过pika生命一个认证用的凭证,然后用pika创建rabbitmq的块连接,再用上面的连接创建一个AMQP信道 。创建消息队列的连接时,需要指定ip,断开,虚拟主机,凭证。
然后根据上面的信道,声明一个队列,
我们可以看到,下面信道点队列声明里的queue参数值就队列的名字。这里是遍历0到9,然后打印了下消息,这里的生成的消息,是json序列化后的数据。然后将数据作为i,信道点基础发布的body参数的值。上面信道点队列声明是创建一个队列,队列名字是’1‘,下面我们用信道点基本发布,是将我们创建的消息体发送到队列中,路由_key就是指定队列名称,指定发布消息到哪个队列,消息是作为body的参数,
最后,需要将这个消息队列的连接关闭。
我们通过页面可以看到,已经创建好了这个队列,队列名字为1,并且已经通过遍历生成的10个消息,调用十次信道点基础发布方法,将这十个产生的消息发布到消息队列中
我们可以再看下,可以看到我们创建的消息的具体内容。
消费者(consumer):队列消息的接收者,扶着接收并处理消息队列中的消息
![复制代码](https://common.cnblogs.com/images/copycode.gif)
import pika credentials = pika.PlainCredentials('admin','admin') connection = pika.BlockingConnection(pika.ConnectionParameters( host='10.0.0.24', port=5672, virtual_host='/', credentials=credentials )) channel = connection.channel() #声明消息队列,消息在这个队列中传递,如果不存在,则创建队列 channel.queue_declare(queue='1',durable=True) # 定义一个回调函数来处理消息队列中消息,这里是打印出来 def callback(ch,method,properties,body): ch.basic_ack(delivery_tag=method.delivery_tag) print(body.decode()) #告诉rabbitmq,用callback来接收消息 channel.basic_consume('1',callback) #开始接收信息,并进入阻塞状态,队列里有信息才会调用callback进行处理 channel.start_consuming()
![复制代码](https://common.cnblogs.com/images/copycode.gif)
获取消息,创建凭证,连接,信道,然后什么一下队列。指定我们要获取哪个队列中的消息,如果没有这个队列,就会创建这个队列,存在,那么后面使用这个信道,就会从这个队列中获取数据。信道是通过rabbitmq的连接对象来生成的,连接对象中放了连接用的凭证。所以,信道点基础消费方法,指定是哪个消息队列,那么就会从这个队列中获取消息。然后传参回调函数。而回调函数中,
我们可以看到,基础消费方法里面有消息回调,就是上面我们自定义的回调函数
这个方法定义了回调函数的写法。第一个参数是信道
第二个参数是方法,第三个参数是属性,第四个是body,这些不用管,只需要按如下格式,就可以从body,做个解码,就将信道点基础消费中指定的队列中的消息,取出来了,我们是用回调函数来接收消息,当需要获取消息的时候,就需要执行信道点开始消费的方法。这里好像是遍历队列一个一个的将消息获取出来。那么怎样实现,实时监听消息,实时消费呢
RabbitMq持久化
RabbitMq持久化
MQ默认建立的临时的queue和exchange,如果不声明持久化,一旦rabbitmq挂掉,queue,exchange将会全部丢失,所以我们一般在创建queue或者exchange的时候会声明持久化
1.queue声明持久化
# 声明消息队列,消息将在这个队列传递,如不存在,则创建。durable = True 代表消息队列持久化存储,False 非持久化存储 result = channel.queue_declare(queue = 'python-test',durable = True)
使用True
重启消息队列服务
消息队列还在,但是消息被清空了
当我改为false的时候,因为队列1已经存在,并且是Tue声明的,所以这里就报错了
我们设置为false,然后声明一个不存在的队列2
创建好了队列,并且10个消息
重启一下消息队列服务
刚刚上面创建的队列2已经不存在,这已经不是消息被清空了,而是队列直接被清除了
也就是这个Ture,是保留队列用的,持久化队列的。
channel.queue_declare(queue='2',durable=True)
2、exchange声明持久化
# 声明exchange,由exchange指定消息在哪个队列传递,如不存在,则创建.durable = True 代表exchange持久化存储,False 非持久化存储 channel.exchange_declare(exchange = 'python-test', durable = True)
注意:如果已存在一个非持久化的queue或exchange,执行上述代码会报错,因为当前状态不能更该queue 或 exchange存储属性,需要删除重建,如果queue和exchange中一个声明了持久化,另一个没有声明持久化,则不允许绑定
我们在1处改了,但是在2处没有修改。结果有问题。
队列2不存在,所以没有将消息放进去
而exchange这里,没有写将消息推送到声明的python-test里面,所以里面也没有消息
这次是声明的exchange,并且将消息推送到python-test里面
还是没有看到有东西呀
我们这里发布个消息,可以看到,是需要路由的
加上路由,再次执行程序
由于队列2 不存在,好像还是不行
我在这里给它bind一个路由
感觉还是没有弄明白,先放弃了
原来是如下方式呀。
首先,在python-test2里面,
给exchange绑定队列1和2
1和2目前的消息数量
我往路由1里面push一个消息
push成功
然后再看队列1里面,可以看到多了一条刚刚push的消息
接下来用程序实现,声明exchange,然后发布方法不变,发布到exchage中,因为已经绑定了两个路由了,这里指定路由key,根据路由key,可以将消息push到对应的队列中去
我们可以看到,之前是页面点击push了一条,上面程序push了十条到exchange,现在这个队列就有11条数据。可是这个exchange和队列的绑定,是我自己在页面上绑定的,这个应该不合理。以后有时间看下,怎么用程序绑定。
我们可以看到,应该是程序中缺少使用这个绑定方法吧
3、消息持久化
虽然exchange和queue都声明了持久化,但如果消息只存在内存里,rabbitmq重启后,内存里的东西还是会丢失,所以必须声明消息也是持久化,从内存转存到到硬盘
# 向队列插入数值 routing_key是队列名。delivery_mode = 2 声明消息在队列中持久化,delivery_mod = 1 消息非持久化 channel.basic_publish(exchange = '',routing_key = 'python-test',body = message, properties=pika.BasicProperties(delivery_mode = 2))
我们这里先重启一下rabbitmq,把之前的写入队列的消息清空
不过我们看到,这里已经有持久化存储的消息了,之前好像是页面点击推送的消息
总共一条,持久化1条。持久化的,即使重启服务,消息也不会丢失
我们再去推送一条
可以看到刚刚推送的这条也是持久化存储的
我们在发布的方法里面,添加属性发布的模式是2,
刚才是2条持久化的,现在新增10条数据,且是持久化的消息
如果改成1
可以看到,刚刚新增了10条消息,但是这10条消息没有持久化。
4、acknowledgement消息不丢失
消费者(consume)调用callback函数时,会存在处理消息失败的风险,如果处理失败,则消息会丢失,但是也可以选择消费者处理失败时,将消息回退给rabbitmq,重新再被消费者消费,这个时候需要设置确认标识。
channel.basic_consume(callback,queue = 'python-test', # no_ack 设置成 False,在调用callback函数时,未收到确认标识,消息会重回队列。True,无论调用callback成功与否,消息都被消费掉 no_ack = False)
目前队列2中有10条没有持久化的,有12条持久化的消息
执行消费程序
再看队列2中,可以看到之前12条持久化和10条没有持久化的消息数据都已经被消费了。我们可以看到消费者这里,多了一个消费者。消费者有个tag,还有ack的确认。在详情那里,也可以看到 消费者数量是1
我们push了一条消息,但是没有发现推送到队列中,难道是因为队列绑定exchange的原因?
push的时候,有个持久化的选择,发现还是没有push进去
在exchange这里push了,
发现队列1有数据,2没有消息
往路由key这里发送多次消息
还是没有,难道上面都是失败的发送嘛
我们再看消费者程序,我们看到运行程序之后,这个程序一直没有退出,处于监听状态,正如我们在队列中看到的那样,有个消费者是up状态,也就是这个消费者一直在监听我们上面的那个队列,程序并没有退出。因此,我们上面在页面push的sss之类的消息,都被这个消费者消费掉了,因此没有看到新增的消息。
我们将上面的消费者程序停掉之后,就可以看到队列下面已经显示没有消费者了,然后再推送消息的时候,页面选择持久化,
我们可以看到,推送的消息,是持久化的。由上面的学习,了解到,消息是否持久化,好像是取决于生产者的设置,而不是说消息没有持久化,我给它用命令持久化一下,至于是否可以用命令持久化一下,本来不需要持久化的消息,暂且不考虑。
RabbitMq发布与订阅
在上一章中,我们创建了一个工作队列,工作队列模式的设想是每一条消息只会被转发给一个消费者。本章将会讲解完全不一样的场景: 我们会把一个消息转发给多个消费者,这种模式称之为发布-订阅模式。
RabbitMq消息模式的核心思想是:一个生产者并不会直接往一个队列中发送消息,事实上,生产者根本不知道它发送的消息将被转发到哪些队列。
实际上,生产者只能把消息发送给一个exchange,exchange只做一件简单的事情:一方面它们接收从生产者发送过来的消息,另一方面,它们把接收到的消息推送给队列。一个exchage必须清楚地知道如何处理一条消息.
rabbitmq的发布与订阅要借助交换机(Exchange)的原理实现:
Exchange 一共有三种工作模式:fanout, direct, topicd
模式一:fanout
这种模式下,传递到exchange的消息将会==转发到所有于其绑定的queue上
不需要指定routing_key,即使指定了也是无效的。
需要提前将exchange和queue绑定,一个exchange可以绑定多个queue,一个queue可以绑定多个exchange。
需要先启动订阅者,此模式下的队列是consume随机生成的,发布者仅仅发布消息到exchange,由exchange转消息至queue。
exchange交换器
首先我们创建一个fanout类型的交换器,我们称之为:python-test:
channel.exchange_declare(exchange = 'python-test',durable = True, exchange_type='fanout')
广播模式交换器很简单,从字面意思也能理解,它其实就是把接收到的消息推送给所有它知道的队列。
想查看当前系统中有多少个exchange,可以从控制台查看
可以看到有很多以amq.*开头的交换器,以及(AMQP default)默认交换器,这些是默认创建的交换器。
在前面,我们并不知道交换器的存在,但是依然可以将消息发送到队列中,那其实并不是因为我们可以不使用交换器,实际上是我们使用了默认的交换器(我们通过指定交换器为字字符串:""),回顾一下我们之前是如何发送消息的:
channel.basic_publish(exchange='',routing_key='1',body=message)
第一个参数是交换器的名字,空字符串表示它是一个默认或无命名的交换器,消息将会由指定的路由键(第二个参数,routingKey,后面会讲)转发到队列。
你可能会有疑问:既然exchange可以指定为空字符串(""),那么可否指定为null?
答案是:不能!
通过跟踪发布消息的代码,在AMQImpl类中的Publish()方面中,可以看到,不光是exchange不能为null,同时routingKey路由键也不能为null,否则会抛出异常:
临时队列
在前面的例子中,我们使用的队列都是有具体的队列名,创建命名队列是很必要的,因为我们需要将消费者指向同一名字的队列。因此,要想在生产者和消费者中间共享队列就必须要使用命名队列。
发布者:
import pika import json credentials = pika.PlainCredentials('admin', 'admin') # mq用户名和密码 # 虚拟队列需要指定参数 virtual_host,如果是默认的可以不填。 connection = pika.BlockingConnection(pika.ConnectionParameters(host = '10.0.0.24',port = 5672,virtual_host = '/',credentials = credentials)) channel=connection.channel() # 声明exchange,由exchange指定消息在哪个队列传递,如不存在,则创建。durable = True 代表exchange持久化存储,False 非持久化存储 channel.exchange_declare(exchange = 'python-test',durable = True, exchange_type='fanout') for i in range(10): message=json.dumps({'OrderId':"1000%s"%i}) # 向队列插入数值 routing_key是队列名。delivery_mode = 2 声明消息在队列中持久化,delivery_mod = 1 消息非持久化。routing_key 不需要配置 channel.basic_publish(exchange = 'python-test',routing_key = '',body = message, properties=pika.BasicProperties(delivery_mode = 2)) print(message) connection.close()
订阅者1:
import pika credentials = pika.PlainCredentials('admin', 'admin') connection = pika.BlockingConnection(pika.ConnectionParameters(host = '10.0.0.24',port = 5672,virtual_host = '/',credentials = credentials)) channel = connection.channel() # 创建临时队列,队列名传空字符,consumer关闭后,队列自动删除 result = channel.queue_declare('4') # 声明exchange,由exchange指定消息在哪个队列传递,如不存在,则创建。durable = True 代表exchange持久化存储,False 非持久化存储 channel.exchange_declare(exchange = 'python-test',durable = True, exchange_type='fanout') # 绑定exchange和队列 exchange 使我们能够确切地指定消息应该到哪个队列去 channel.queue_bind(exchange = 'python-test',queue = "4") # 定义一个回调函数来处理消息队列中的消息,这里是打印出来 def callback(ch, method, properties, body): ch.basic_ack(delivery_tag = method.delivery_tag) print(body.decode()) channel.basic_consume(result.method.queue,callback,# 设置成 False,在调用callback函数时,未收到确认标识,消息会重回队列。True,无论调用callback成功与否,消息都被消费掉 auto_ack = False) channel.start_consuming()
订阅者2
import pika credentials = pika.PlainCredentials('admin', 'admin') connection = pika.BlockingConnection(pika.ConnectionParameters(host = '10.0.0.24',port = 5672,virtual_host = '/',credentials = credentials)) channel = connection.channel() # 创建临时队列,队列名传空字符,consumer关闭后,队列自动删除 result = channel.queue_declare('2',durable=True) # 声明exchange,由exchange指定消息在哪个队列传递,如不存在,则创建。durable = True 代表exchange持久化存储,False 非持久化存储 channel.exchange_declare(exchange = 'python-test',durable = True, exchange_type='fanout') # 绑定exchange和队列 exchange 使我们能够确切地指定消息应该到哪个队列去 channel.queue_bind(exchange = 'python-test',queue = "2") # 定义一个回调函数来处理消息队列中的消息,这里是打印出来 def callback(ch, method, properties, body): ch.basic_ack(delivery_tag = method.delivery_tag) print(body.decode()) channel.basic_consume(result.method.queue,callback,# 设置成 False,在调用callback函数时,未收到确认标识,消息会重回队列。True,无论调用callback成功与否,消息都被消费掉 auto_ack = False) channel.start_consuming()
当前的队列如下
发布消息,exchange类型不对
下面这就是直连类型
进去之后把找个已经存在的exchange删除了,这个暂时没用
发布,这里也没有指的路由key
可以看到新建的exchange类型是fanout
因为没有绑定队列,所以程序推送的消息,好像是丢失了
开启订阅者1,声明队列4并绑定到前面创建的python-test这个exchange。
查看,队列4已经创建
有个消费者正连接着4
并且订阅者1声明的队列,也跟指定的exchange已经绑定了,路由key,默认就是用的队列名称
pika.exceptions.ChannelClosedByBroker: (406, "PRECONDITION_FAILED - inequivalent arg 'durable' for queue '2' in vhost '/': received 'false' but current is 'true'")
开启订阅者2,但是报错了,因为队列2已经存在了,并且是Ture,是持久化的,而这里信道点队列声明2,是没有指定那个参数,那就是默认是Flase,非持久化的队列,重启下服务这个队列就不存在了。因此保持了。我们先将这个已经存在的队列删除,然后重新声明一下吧,或者是直接给它加个持久化的参数也行
加上之后,就能正常开启这个订阅者2了
我们创建的4,是非持久化的队列,这里这个d的标记,可能就是durable参数,是否持久化队列的意思吧
我们重新执行一次发布者程序,发布者并没有指定路由key,只是指定了exchange,而订阅者1和2程序里面,都是有绑定这个exchange的
我们可以看到,订阅者1获取到了发布到这个exchage的消息
订阅者2也获取到了发布到这个exchage的消息
再来看下这个exchange的情况
它对应的两个队列
队列2有个消费者
队列4也有个消费者,这两个消费者各自对应一个队列,每个消费者请求过来是的端口不同,消费者tag不同。两个队列中的消息,都被订阅者程序获取并打印在pycharm上进行消费了,因此,队列中也就没有数据了。
难道,一个队列,就是一个订阅者吗?当发布者发布消息的时候,难道是基础发布方法里面,指定exchange,不指定路由key,这样就会将生产者生产的消息,发送给所有绑定这个exchange的队列吗,而订阅者和队列一一对应,然后每个订阅者就从自己对应的队列中将这个消息消费掉吗?
把两个订阅者,都停止掉,查看目前这两个队列,都是没有消息的。
我执行发布者程序,发布消息,指定exchange,不指定路由key。
我们可以看到,这种情况下,的确是将消息发布给所有绑定这个exchange的队列了,如下,2和4队列都绑定了,所以都接收到了十条消息。
我们发布消息的参数,指定消息是持久化的,因为队列2是个持久化的队列,因此,进入队列2的消息也是持久化的
由于声明队列4,不是持久化的队列,因此,即使发布消息时,指定消息是持久化的,但是实际上这个消息也是没有在这个非持久化的队列中进行持久化,也只是临时的罢了。
我开启订阅者1
订阅者1对应着队列4,队列4的消息已经被消费了,已经在上图中打印出来了。
开启订阅者2
订阅者2对应的队列是2,也将消息消费掉了,并在订阅者2程序中打印了出来
如果,队列或者消息是临时的,消费者还没消费的消息,因为重启服务,那么就会丢失消息,消费者应该就消费不到那个丢失的消息了。
模式二:direct
这种工作模式的原理是消息发送至exchange,exchange根据**路由键(routing_key)**转发到相对应的queue上。
- 可以使用默认exchange=’ ',也可以自定义exchange
- 这种模式下不需要将exchange和任何进行绑定,当然绑定也是可以的,可以将exchange和queue,routing_key和queue进行绑定
- 传递或接收消息时,需要指定routing_key
- 需要先启动订阅者,此模式下队列是consumer随机生成的,发布者仅仅发布消息到exchange,由exchange转发消息至queue。
发布者:
import pika import json credentials = pika.PlainCredentials('admin', 'admin') # mq用户名和密码 # 虚拟队列需要指定参数 virtual_host,如果是默认的可以不填。 connection = pika.BlockingConnection(pika.ConnectionParameters(host = '10.0.0.24',port = 5672,virtual_host = '/',credentials = credentials)) channel=connection.channel() # 声明exchange,由exchange指定消息在哪个队列传递,如不存在,则创建。durable = True 代表exchange持久化存储,False 非持久化存储 channel.exchange_declare(exchange = 'python-test',durable = True, exchange_type='direct') for i in range(10): message=json.dumps({'OrderId':"1000%s"%i}) # 指定 routing_key。delivery_mode = 2 声明消息在队列中持久化,delivery_mod = 1 消息非持久化 channel.basic_publish(exchange = 'python-test',routing_key = 'OrderId',body = message, properties=pika.BasicProperties(delivery_mode = 2)) print(message) connection.close()
订阅者:
import pika credentials = pika.PlainCredentials('admin', 'admin') connection = pika.BlockingConnection(pika.ConnectionParameters(host = '10.0.0.24',port = 5672,virtual_host = '/',credentials = credentials)) channel = connection.channel() # 创建临时队列,队列名传空字符,consumer关闭后,队列自动删除 result = channel.queue_declare('',exclusive=True) # 声明exchange,由exchange指定消息在哪个队列传递,如不存在,则创建。durable = True 代表exchange持久化存储,False 非持久化存储 channel.exchange_declare(exchange = 'python-test',durable = True, exchange_type='direct') # 绑定exchange和队列 exchange 使我们能够确切地指定消息应该到哪个队列去 channel.queue_bind(exchange = 'python-test',queue = result.method.queue,routing_key='OrderId') # 定义一个回调函数来处理消息队列中的消息,这里是打印出来 def callback(ch, method, properties, body): ch.basic_ack(delivery_tag = method.delivery_tag) print(body.decode()) #channel.basic_qos(prefetch_count=1) # 告诉rabbitmq,用callback来接受消息 channel.basic_consume(result.method.queue,callback, # 设置成 False,在调用callback函数时,未收到确认标识,消息会重回队列。True,无论调用callback成功与否,消息都被消费掉 auto_ack = False) channel.start_consuming()
将之前测试用的exchanges删除,队列也删除
使用direct类型的exchange,发布消息
没有队列生成
开启消费者程序,exchange声明的类型是direct,队列绑定exchange,指定路由key,这个路由key,并没有这个名字的队列
开启上面的消费者程序之后,就生成了一个队列。这个生成的队列,进入可以看到是有消费者在监听这个队列的。这个队列,以上面命名的路由key,来绑定了前面定义的exchange。
我们进入这个exchange查看下,路由key,定向到某个队列
我们看下发布消息的程序,就是exchange声明里面,定义了direct方式,而基础发布方法里面,就指定发布到上面定义的exchange,然后指定路由key为之前执行消费者程序时,随机生成名字的队列,绑定exchange时使用的路由key。这样,我们发布消息的时候,发布给exchange,就会根据路由key,然后找到对应的队列,将消息推送到这个队列中。
由于我们的订阅者,一直在监听,当上面发布消息到队列中后,订阅者就从exchange下根据路由key,找到对应的队列,然后将队列中的消息消费,打印到pycharm上,
模式三:topicd
这种模式和第二种差不多,exchange也是通过路由键routing_key来转发消息到指定的queue。不同之处在于:
**routing_key使用正则表达式支持模糊匹配,**但匹配规则又与常规正则表达式不同,比如"#"是匹配全部,“*”是匹配一个词。
举例:routing_key =“#orderid#”,意思是将消息转发至所有 routing_key 包含 “orderid” 字符的队列中。代码和模式二 类似,
我们用上面的代码改 一下,再复制处两个订阅者,只需要修改下路由key为带2的 带3的数字就可以
我们再改一下
我们看页面,可以看到又多了两个队列了
可以看到这个exchange对应三个队列,路由key都是带有OrderId,
我们将路由key,改为匹配的方式,然后发布消息
演示失败
参考链接:https://blog.csdn.net/weixin_45144837/article/details/104335115
标签:持久,exchange,python,rabbitmq,queue,队列,消息,操作,channel From: https://www.cnblogs.com/machangwei-8/p/17500641.html