天天热讯:python操作rabbitmq

rabbitmq安装部署

RabbitMq生产者消费者模型

生产者（producter）队列消息的产生者，复制生产消息，并将消息传入队列生产者代码：

import pikaimport jsoncredentials = 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设为Truechannel.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()

通过pika生命一个认证用的凭证，然后用pika创建rabbitmq的块连接，再用上面的连接创建一个AMQP信道 。创建消息队列的连接时，需要指定ip，断开，虚拟主机，凭证。

【资料图】

然后根据上面的信道，声明一个队列，

我们可以看到，下面信道点队列声明里的queue参数值就队列的名字。这里是遍历0到9，然后打印了下消息，这里的生成的消息，是json序列化后的数据。然后将数据作为i，信道点基础发布的body参数的值。上面信道点队列声明是创建一个队列，队列名字是’1‘，下面我们用信道点基本发布，是将我们创建的消息体发送到队列中，路由_key就是指定队列名称，指定发布消息到哪个队列，消息是作为body的参数，

最后，需要将这个消息队列的连接关闭。

我们通过页面可以看到，已经创建好了这个队列，队列名字为1，并且已经通过遍历生成的10个消息，调用十次信道点基础发布方法，将这十个产生的消息发布到消息队列中

我们可以再看下，可以看到我们创建的消息的具体内容。

消费者（consumer）:队列消息的接收者，扶着接收并处理消息队列中的消息

import pikacredentials = 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()

获取消息，创建凭证，连接，信道，然后什么一下队列。指定我们要获取哪个队列中的消息，如果没有这个队列，就会创建这个队列，存在，那么后面使用这个信道，就会从这个队列中获取数据。信道是通过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 pikaimport jsoncredentials = 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 pikacredentials = 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 pikacredentials = 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 pikaimport jsoncredentials = 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 pikacredentials = 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