该博客的主要讲解了以下几种rabbitmq的用法
1、实现简单的生产者发送消息给消费者
2、实现序列持久化
3、实现消息持久化
4、实现消息公平分发
5、实现广播
6、实现组播
7、实现细分组播
先来看下rabbitmq的简单介绍，我们后面介绍具体的代码
python操作rabbitmq的模块叫做pika
1个生产者对应1个消费者
如果生产者发送一条消息，那么消费者也只能接受到1条消息
如果生产者发送两条消息，那么消费者就可以接受到2条消息
1个生产者对应2个消费者
如果生产者发送1条消息，那么只有1个消费者能接受到消息
如果生产者发送2条消息，那么每个消费者都接受到1条消息
如果生产者先启动了，生产者的代码已经执行完成，程序退出，这个时候消费者才启动，消费者也能接受到消息
work queue
如果一个生产者对应多个消费者，那么生产者会依次把消息轮训到多个消费者上，实现一个负载均衡的效果
但是如果rabbitmq server重启后，则队列和消息就会丢失
1、队列持久化
durable=True
2、消息持久化
delivery_mode = 2
如果申明一个queue为持久化，那么就需要在服务端和客户端都需要设置
如果申明一个消息的持久化，加一个
delivery_mode = 2
消息公平分发
确保每个消费者同时只能有固定的数量的任务在处理，比如这个固定的任务是1或者2或者3之类的
这个需要在消费者端设置，设置我这个消费者同时只能处理几个任务
设置方法如下，这个1就是同时只能处理一个任务
channel.basic_qos(prefetch_count=1)
发布订阅
前面的一条消息只能被一个客户端消费，那么发布和订阅就是一条消息可以被
多个客户端消息，这里就需要用到“Exchange”，Exchange在定义的时候是有
类型的，以决定到底哪些queue符合条件，可以接受消息，一个exchange可以绑
定多个queue
一个exchange可以绑定多个queue
fanout：所有绑定在这个exchange上的队列都会接受到消息
direct：通过routingKey和exchange决定的哪个唯一的queue可以接收消息
可以根据关键字发送，即：队列绑定关键字，发送者将根据关键字发送消息
到指定的队列中
topic：所有符合routingKey（此时可以是一个表达式）的routingKey所绑定
的queue都可以接收到消息
表达式符合说明：#号代表一个或多个字符，*代表人和字符
例如
#a.a会匹配a.a,aa.a,aaa.a
*.a会匹配a.a,b.a,c.a
headers:通过headers来决定把消息发给哪些queue
一、先看下rabbitmq的最简单的使用，实现简单的通信
1、先看生产者的代码
# 1、创建一个连接
# 2、创建一个管道
# 3、创建一个队列
import pika
test_connection = pika.BlockingConnection(pika.ConnectionParameters(host="localhost"))
# 1、创建一个连接
test_channel = test_connection.channel()
# 2、创建一个管道
test_channel.queue_declare(queue="test1")
#申明一个队列，名称为“test1”
test_channel.basic_publish(exchange="",
routing_key="test1",
body="hi,this is 1")
#通过管道发送消息
# 1、exchange主要是队列选择，控制消息往哪个队列中发送，这里还没有用到，后面才会遇到
# 2、routing_key就是指明队列的名称
# 3、body就是发送的消息
print("pub send message success!")
#没有特殊的意义
test_channel.close()
#关闭连接
2、在看消费者的代码
import pika
test_connection = pika.BlockingConnection(pika.ConnectionParameters(host="localhost"))
#创建一个连接
test_channel = test_connection.channel()
#申明一个管道
test_channel.queue_declare(queue="test1")
#在管道中创建一个队列，名称为“test1”，该名称要和生产者的名称一样，
def f(ch,method,properties,body):
print("pub:",body)
#定义一个函数，这个函数的作用就是只要消费者端收到信息，则会立刻触发执行这个函数
test_channel.basic_consume(f,
queue="test1",
no_ack=True)
#这里的意思是，写好监控的代码，但是还没有开始监控，这个作用就，兼容队列“test1”，只有一有消息进来，则立刻触发f这个函数，函数中的body参数
# 就是队列中的消息
print("sub waiting for message!")
test_channel.start_consuming()
#这句话的才是真正的开始监控队列中的消息，如果没有收到消息，则会一直阻塞，如果有消息，则不阻塞
二、这里我们介绍下如何实现序列持久化，何为序列持久化，我们在这里需要说明一下
序列持久化：我们在管道中申明了一个序列，如果rabbitmq重启后，则这个序列就会丢失；如果我们实现序列持久化，得到的效果就是如果我们开始申明过一个序列，且把该序列持久化，那么就算rabbitmp服务器重启，我们也不需要在重新申明序列，如果测试我们可以这样做：先执行服务端的代码，执行完成后重启rabbitmq，然后修改生产者的代码，把申明队列的代码注释掉，代码还是可以正常的执行，不会报错，启动消费者的代码，可以正常接受到消息。
核心的代码如下
在消费端的代码和服务端的代码是一样的，只需要在申明队列的加一个参数即可
注：这里需要注意的是，如果在前面定义一个名称为队列的名称为“a”，如果我们在没有重启rabbitma之前，如果想把队列“a”修改为持久化，则会报错
test_channel.queue_declare(queue="test1",durable=True)
三、这里我们介绍下如何实现消息的持久化，何为消息持久化，我们在这里需要说明一下
消息持久化：如果生产者的代码已经执行完成，即消息已经发送出去了，此时消费者还未消费这条消息，此时rabbitmq重启，重启成功后，此时启动消费者的代码，消费者还可以正常的接受到消息
核心代码如下，实现消息持久化，生产者和消费者的代码是不一样
先看生产者的核心代码，在发布消息的时候加这么一行即可：
test_channel.basic_publish(exchange="",
routing_key="test1",
body="hi,this is 1",
properties=pika.BasicProperties(delivery_mode=2))
在看消费者的核心代码，在回调函数中加一行即可
def f(ch,method,properties,body):
ch.basic_ack(delivery_tag=method.delivery_tag)
print("pub:",body)
四、我们在看看下如何实现消息的公平分发，主要处理的场景是这样的：正常情况下，如果有多个消费者，消息处理的原则是这样的，第一条消息由A消费者处理，第二天消息由B消费者处理，第三条消息由C处理，那么如果消费者A、B、C的处理消息的能力不一样，那么就会出现配置高的消费者处于空闲状态，配置低的消费者一直在不停处理问题的；消息的公平分发就是要解决这个问题，最终的效果就是同一个时间一个消费者只能处理处理一个消息，如果A消费者没有处理完，则会把消息发送给消费者2处理，而不会一直等待消费者A去处理
消息的公平处理的代码也很简单，消费者端实现，生产者端不需要做任何修改
核心代码如下，只有一行代码，这里的1就是同一时间一个消费者只处理一条消息
channel.basic_qos(prefetch_count=1)
五、实现广播，这个是什么意思呢？我们这里简单介绍一下，前面我们讲的所有的例子都是一条消息只能被一个消费者消费，而广播实现的效果就是一条消息可以发送多个消费者
要实现广播，我们这里需要用到exchange了，只要绑定在一个exchange的队列都可以接受到同一条消息
先看一个简单的实现
生产者端绑定3个queue到exchange上，三个消费者每个绑定了一个queue到exchange上，我们看下效果
生产者的代码
import pika
test_connection = pika.BlockingConnection(pika.ConnectionParameters(host="localhost"))
test_channel = test_connection.channel()
test_channel.exchange_declare(exchange="test",
exchange_type="fanout")
queue_list = []
test_channel.queue_declare(queue="hello1",durable=True)
queue_list.append("hello1")
test_channel.queue_declare(queue="hello2",durable=True)
queue_list.append("hello2")
test_channel.queue_declare(queue="hello3",durable=True)
queue_list.append("hello3")
for i in queue_list:
test_channel.basic_publish(exchange="test",
routing_key=i,
body="broadcast information!")
test_channel.close()
消费者的代码
import pika
test_connection = pika.BlockingConnection(pika.ConnectionParameters(host="localhost"))
test_channel = test_connection.channel()
test_channel.exchange_declare(exchange="test",
exchange_type="fanout")
test_channel.queue_declare(queue="hello1",durable=True)
test_channel.queue_bind(exchange="test",queue="hello1")
def callback(ch,method,properties,body):
print("server:",body)
test_channel.basic_consume(callback,
queue="hello1",
no_ack=True)
test_channel.start_consuming()
我们看下结果：每个消费者都接受到了3条消息，但是我们在生产者端明明是往每个队列中发一条消息，为什么消费者1、2、3都收到了3条消息呢，因为我们什么exchange的类型fanout了，exchange类型就会把消息广播到所有绑定到该exchange上的队列
server: b'broadcast information!' server: b'broadcast information!' server: b'broadcast information!'
在看下官方的代码例子，其实如果exchange的类似fanout，就不需要申明队列了，只需要申明exchange的类型和名称即可
先看生产者端代码
import pika
import sys
# 前面的例子都基本是1对1的消息发送和接受，即消息之恩能够发送到指定的队列中，但是
# 有的时候我们需要消息被所有的queue接受，类似的广播的效果
# 如果要实现广播的效果，这时候就要用到exchange了
# exchange在定义的时候是有类型的，以决定到底哪些队列符合条件，可以接受消息
# 如果是fanout，则不需要申明queue
client_connection = pika.BlockingConnection(pika.ConnectionParameters(host="localhost"))
#首先也是要连接上rabbitmq，和生产者端是一样的
channel = client_connection.channel()
channel.exchange_declare(exchange="logs",
exchange_type="fanout")
message = "".join(sys.argv[1:]) or "info:hello world!"
channel.basic_publish(exchange="logs",
routing_key="",
body=message)
print("[x] send %r" % message)
client_connection.close()
在看消费者端的代码，消费者端需要申明一个队列，这里我们没有生成一个具体的名称，而是让系统自动替我们申明了一个队列，该队列在代码执行完成后自动销毁
自动生成队列的如下
result = channel.queue_declare(exclusive=True) #不指定queue的名字，rabbitmq自动随机分配一个名字，并且在这个queue会在消费者 #断开后自动被删除 queue_name = result.method.queue
消费者的核心代码如下
import pika
client_connection = pika.BlockingConnection(pika.ConnectionParameters(host="localhost"))
#首先也是要连接上rabbitmq，和生产者端是一样的
channel = client_connection.channel()
channel.exchange_declare(exchange="logs",
exchange_type="fanout")
result = channel.queue_declare(exclusive=True)
#不指定queue的名字，rabbitmq自动随机分配一个名字，并且在这个queue会在消费者
#断开后自动被删除
queue_name = result.method.queue
channel.queue_bind(exchange="logs",
queue=queue_name)
#把queue绑定到exchange上，他才能接受消息
print("waiting for message")
def callback(ch,method,properties,body):
print("receive message %r" %body)
channel.basic_consume(callback,
queue=queue_name,
no_ack=True)
channel.start_consuming()
原创文章，作者：Maggie-Hunter，如若转载，请注明出处：https://blog.ytso.com/20863.html