rabbitmq的简单介绍一

该博客的主要讲解了以下几种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()