RabbitMQ

[kagxin]

消息通信的三个概念

• 消费者
• 生产者
• 代理（代理即消息中间件，可以是rabbitmq）

AMQP三元素

交换机

• 消息发送到交换机时会带路由键
• 队列绑定到交换机时会确定绑定关系
• 所有生产者的消息都是发布到交换机的，然后交换机会根据路由键(消息标签中)，把消息路由到指定的队列，如果一条消息没有匹配到任何的绑定规则消息会进入”黑洞“
• 交换机类型有四种：direct、fanout、topic、headers
• 基于消息的路由键和交换机类型，rabbit会决定消息投递到那个队列

队列

• 所有消息最终到达队列，并被消费者接受
• 当多个消费者订阅到同一个队列上时，队列收到的消息将以循环的的方式发送给消费者
• 消息进入无人订阅的队列，消息会在队列中等待(堆积)
• 消息都会包含两部分内容：有效载荷和标签。其中标签的一个重要作用描述了绑定关系，绑定关系决定，消息可以给到感兴趣的接收方
• 消费者接受到消息之后，要对消息进行 确认。如果消费者收到一条消息，然后确认之前从rabbitmq断开(或取消订阅)，rabbitmq会认为消息没有分发，重新发送给下一个订阅者。如果程序有bug而没有对消息进行确认的话，rabbit将不会给该消费者发送更多的消息了，因为rabbit会认为消费者一直没有处理完当前这条消息。
• 消费者可以拒绝接受某条消息，消费者可以让这条被拒绝的消息可以重新入队，或者进入死信队列(rabbitmq版本有关可能是直接删除)。
• 为了避免消息进入黑洞，消费者和生产者都应该尝试去创建队列
• 队列类型，私有队列：当想要限制一个队列只有一个消费者的时候很有帮助。自动删除队列：当最后一个消费者取消订阅时，队列会自动移除。匿名队列：对于构建在AMQP上的RPC应用来首，使用匿名队列很有用。持久化队列：rabbit重启队列依然存在。

绑定

• 声明交换机路由消息到特定的队列的规则

虚拟主机

• 命名空间，起到隔离作用
• 权限控制基本单元

docker-compose

version: "3"
services:
  rabbitmq:
    image: rabbitmq:3-management
    container_name: rabbitmq
    hostname: mymq
    environment:
      - RABBITMQ_DEFAULT_USER=root
      - RABBITMQ_DEFAULT_PASS=root
    ports:
      - 15672:15672
      - 5672:5672
    restart: always
    volumes:
      - /Users/kangxin/docker/rabbitmq/data:/var/lib/rabbitmq

rabbitmq management web 插件在15672端口，rabbitmq服务在5672端口

示例

topic交换机例子 绑定关系如图

这样路由键是hola.one 的消息会进入到队列hello-queue中 路由键符合hola.* 模式的都会进入hello-queue2中

生产者

• python3.6
• pika==1.1.0

  
  import pika

credentials = pika.PlainCredentials("root", "root") conn_params = pika.ConnectionParameters("127.0.0.1", credentials=credentials)

conn_broker = pika.BlockingConnection(conn_params) channel = conn_broker.channel()

channel.exchange_declare(exchange="hello-exchange", exchange_type="topic", passive=False, durable=True, auto_delete=False)

msg_props = pika.BasicProperties() msg_props.content_type = "text/plain" # 声明消息类型

channel.basic_publish(body='i am message routing_key: hola.two', exchange="hello-exchange", properties=msg_props, routing_key="hola.two") # 发布一条消息路由键是 hola.two

channel.basic_publish(body='i am message routing_key: hola.one', exchange="hello-exchange", properties=msg_props, routing_key="hola.one") # 发布一条消息路由键是 hola.one

#### 消费者
```python
import pika

credentials = pika.PlainCredentials("root", "root")
conn_params = pika.ConnectionParameters('127.0.0.1', credentials=credentials)

conn_broker = pika.BlockingConnection(conn_params)
channel = conn_broker.channel()

channel.exchange_declare(exchange="hello-exchange", exchange_type="topic", passive=False, durable=True,
                         auto_delete=False)  # 声明交换机
channel.queue_declare("hello-queue")  # 声明队列
channel.queue_declare("hello-queue2")  # 声明队列

# 确定绑定关系，只接受routingkey为hola.one的消息
channel.queue_bind(queue="hello-queue", exchange="hello-exchange", routing_key="hola.one")

# 确定绑定关系，接受所有符合 hola.* 模式的消息
channel.queue_bind(queue="hello-queue2", exchange="hello-exchange", routing_key="hola.*")

def msg_consumer(channel, method, header, body):  # 消息处理函数， 处理hello-queue队列中消息
    print(f'msg_consumer: bind routing_key: hola.one, message: {body.decode()}')
    channel.basic_ack(delivery_tag=method.delivery_tag)

def msg_consumer_all(channel, method, header, body):  # 消息处理函数，处理hello-queue2队列中消息
    print(f'msg_consumer_all: bind routing_key: hola.*, message: {body.decode()}')
    channel.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_consume("hello-queue", msg_consumer, consumer_tag="hello-consumer")  # 注册回调函数
channel.basic_consume("hello-queue2", msg_consumer_all, consumer_tag="all-consumer")  # 注册回调函数
print('start...')
channel.start_consuming()  # 启动消费

控制台日志

start...
msg_consumer: bind routing_key: hola.one, message: i am message routing_key: hola.one
msg_consumer_all: bind routing_key: hola.*, message: i am message routing_key: hola.two
msg_consumer_all: bind routing_key: hola.*, message: i am message routing_key: hola.one

两条消息都进入了队列 hello-queue2