消息队列--2--RabbitMQ

Posted 2020-10-16 EagleSour

RabbitMQ

RabbitMQ是一个在AMQP基础上完整的，可复用的企业消息系统。他遵循Mozilla Public License开源协议。

MQ全称为Message Queue, 消息队列（MQ）是一种应用程序对应用程序的通信方法。应用程序通过读写出入队列的消息（针对应用程序的数据）来通信，而无需专用连接来链接它们。消 息传递指的是程序之间通过在消息中发送数据进行通信，而不是通过直接调用彼此来通信，直接调用通常是用于诸如远程过程调用的技术。排队指的是应用程序通过 队列来通信。队列的使用除去了接收和发送应用程序同时执行的要求。

RabbitMQ安装

?

1 2 3 4 5 6 7 8

安装配置epel源    $ rpm -ivh http://dl.fedoraproject.org/pub/epel/6/i386/epel-release-6-8.noarch.rpm   安装erlang    $ yum -y install erlang   安装RabbitMQ    $ yum -y install rabbitmq-server

注意：service rabbitmq-server start/stop

安装API

?

1 2 3 4 5 6 7

pip install pika or easy_install pika or 源码   https://pypi.python.org/pypi/pika

使用API操作RabbitMQ

基于Queue实现生产者消费者模型

View Code

对于RabbitMQ来说，生产和消费不再针对内存里的一个Queue对象，而是某台服务器上的RabbitMQ Server实现的消息队列。

?

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16

#!/usr/bin/env python import pika   # ######################### 生产者 #########################   connection = pika.BlockingConnection(pika.ConnectionParameters(         host=\'localhost\')) channel = connection.channel()   channel.queue_declare(queue=\'hello\')   channel.basic_publish(exchange=\'\',                       routing_key=\'hello\',                       body=\'Hello World!\') print(" [x] Sent \'Hello World!\'") connection.close()

?

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20

#!/usr/bin/env python import pika   # ########################## 消费者 ##########################   connection = pika.BlockingConnection(pika.ConnectionParameters(         host=\'localhost\')) channel = connection.channel()   channel.queue_declare(queue=\'hello\')   def callback(ch, method, properties, body):     print(" [x] Received %r" % body)   channel.basic_consume(callback,                       queue=\'hello\',                       no_ack=True)   print(\' [*] Waiting for messages. To exit press CTRL+C\') channel.start_consuming()

1、acknowledgment 消息不丢失

no-ack ＝ False，如果消费者遇到情况(its channel is closed, connection is closed, or TCP connection is lost)挂掉了，那么，RabbitMQ会重新将该任务添加到队列中。

消费者

2、durable   消息不丢失

生产者

消费者

3、消息获取顺序

默认消息队列里的数据是按照顺序被消费者拿走，例如：消费者1 去队列中获取 奇数 序列的任务，消费者1去队列中获取 偶数 序列的任务。

channel.basic_qos(prefetch_count=1) 表示谁来谁取，不再按照奇偶数排列

消费者

4、发布订阅

发布订阅和简单的消息队列区别在于，发布订阅会将消息发送给所有的订阅者，而消息队列中的数据被消费一次便消失。所以，RabbitMQ实现发布和订阅时，会为每一个订阅者创建一个队列，而发布者发布消息时，会将消息放置在所有相关队列中。

 exchange type = fanout

发布者

订阅者

5、关键字发送

 exchange type = direct

之前事例，发送消息时明确指定某个队列并向其中发送消息，RabbitMQ还支持根据关键字发送，即：队列绑定关键字，发送者将数据根据关键字发送到消息exchange，exchange根据 关键字 判定应该将数据发送至指定队列。

消费者

生产者

6、模糊匹配

 exchange type = topic

在topic类型下，可以让队列绑定几个模糊的关键字，之后发送者将数据发送到exchange，exchange将传入”路由值“和 ”关键字“进行匹配，匹配成功，则将数据发送到指定队列。

• # 表示可以匹配 0 个 或 多个 单词
• *  表示只能匹配 一个 单词

?

1 2 3

发送者路由值              队列中 old.boy.python          old.*  -- 不匹配 old.boy.python          old.#  -- 匹配

消费者

生产者

注意：

 

View Code