1、消息队列之RabbitMQ的简单使用

一、什么是消息队列（MQ）

MQ全称为Message Queue 消息队列（MQ）是一种应用程序对应用程序的通信方法。MQ是消费-生产者模型的一个典型的代表，一端往消息队列中不断写入消息，而另一端则可以读取队列中的消息。这样发布者和使用者都不用知道对方的存在。

'''
生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，
所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，
阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。
'''

我们先不管消息(Message)这个词，来看看队列(Queue)。这一看，队列大家应该都熟悉吧。

队列是一种先进先出的数据结构。

1、消息队列之RabbitMQ的简单使用

消息队列可以简单理解为：把要传输的数据放在队列中。

1、消息队列之RabbitMQ的简单使用

二、为什么要使用消息队列

消息队列中间件是分布式系统中重要的组件，主要解决应用解耦，异步消息，流量削锋等问题，实现高性能，高可用，可伸缩和最终一致性架构。目前使用较多的消息队列有ActiveMQ，RabbitMQ，ZeroMQ，Kafka，MetaMQ，RocketMQ。

接下来利用一个外卖系统的消息推送给大家解释下MQ的意义。

1、消息队列之RabbitMQ的简单使用

详细请看：https://zhuanlan.zhihu.com/p/99783523

三、RabbitMQ

RabbitMQ 是一个由 Erlang 语言开发的 AMQP 的开源实现。

rabbitMQ是一款基于AMQP协议的消息中间件，它能够在应用之间提供可靠的消息传输。在易用性，扩展性，高可用性上表现优秀。使用消息中间件利于应用之间的解耦，生产者（客户端）无需知道消费者（服务端）的存在。而且两端可以使用不同的语言编写，大大提供了灵活性。

https://www.rabbitmq.com/tutorials/tutorial-one-python.html

Producer（生产者）： 消息的生产者，负责产生消息并把消息发到交换机

Consumer （消费者）：使用队列 Queue 从 Exchange 中获取消息的应用。

Exchange （交换机）：负责接收生产者的消息并把它转到到合适的队列。

Queue （队列）：一个存储Exchange 发来的消息的缓冲，并将消息主动发送给Consumer，或者 Consumer 主动来获取消息。

Binding （绑定）：队列 和 交换机 之间的关系。Exchange 根据消息的属性和 Binding 的属性来转发消息。绑定的一个重要属性是 binding_key。

Connection （连接）和 Channel （通道）：生产者和消费者需要和 RabbitMQ 建立 TCP 连接。一些应用需要多个connection，为了节省TCP 连接，可以使用 Channel，它可以被认为是一种轻型的共享 TCP 连接的连接。连接需要用户认证，并且支持 TLS (SSL)。连接需要显式关闭。

Message （消息）： RabbitMQ 转发的二进制对象，包括Headers（头）、Properties （属性）和 Data （数据），其中数据部分不是必要的。

routing_key：路由键、路由密钥。

3.1、rabbitmq的安装

Mac 环境下 RabbitMQ 的安装

下载 RabbitMQ 源文件，解压源文件之后进行安装。
通过 brew 命令安装。

在这里，我当然是推荐使用 brew 来安装，非常强大的 Mac 端包管理工具

有了 brew 之后，只需要一个简单的命令就搞定了。

brew install rabbitmq

安装的路径是 /usr/local/Cellar/rabbitmq/3.8.3，具体情况要视版本而定，我安装的版本是 3.8.3。

接下来就可以启动了，进入安装目录，执行命令：

./sbin/rabbitmq-server

3.2、RabbitMQ的工作模型之简单模式

3.2.1、代码

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.queue_declare(queue='hello')

channel.basic_publish(exchange='',
                      routing_key='hello',
                      body='Hello World!')

print(" [x] Sent 'Hello World!'")

### 消费者

import pika
#连接rabbitmq
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
#创建一个名为‘hello’的队列
channel.queue_declare(queue='hello')

def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)

#去像hello队列取数据
channel.basic_consume(queue='hello',
                      auto_ack=True,
                      on_message_callback=callback)


print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()

View Code

3.2.2、参数

应答参数（在消费者中操作）

'''
应答模式-当生产者监听队列，去产生任务放到队列中，消费者去取的时候，如果中间出现bug或者业务逻辑错误的时候，
导致消费者无法接收任务，从而导致数据丢失，这个时候，如果修改完bug之后，队列中已经没有任务了

手动模式- 这个时候生成者把任务放到队列中，当消费者想取的时候，队列会分复制出一份给消费者，当消费者产生bug之后，对队列中的数据也不会产生影响，
所以我们需要加以下的代码， 当修改完bug重启之后，消费者会发送给对列一个确认的信号，然后队列会移除那个任务，从而使数据完整
'''

auto_ack=True #是应答模式 Fasle 为手动模式 #需要在消费者中业务结束部分增加
ch.basic_ack(delivery_tag=method.delivery_tag)


ps：如果想要数据完整性就需要改为手动模式，如果需要速度改为应答模式

持久化参数（在生产者中操作）

'''
当生产者生产完数据的时候，rabbitmq崩了，这个时候消费者还没取数据，当重启rabbitmq的时候，队列中已经没有数据了 ，
因为之前的数据是存在内存中的， 当我们设置durable=True的时候，队列中的数据会被持久化到硬盘中，重启rabbitmq的时候，消费者可以取到数据
'''


#声明queue
channel.queue_declare(queue='hello2', durable=True)  # 若声明过，则换一个名字
 
channel.basic_publish(exchange='',
                      routing_key='hello2',
                      body='Hello World!',
                      properties=pika.BasicProperties(
                          delivery_mode=2,  # make message persistent
                          )
                      )

分发参数

有两个消费者同时监听一个的队列。其中一个线程sleep2秒，另一个消费者线程sleep1秒，但是处理的消息是一样多。这种方式叫轮询分发（round-robin）不管谁忙，都不会多给消息，总是你一个我一个。想要做到公平分发（fair dispatch），必须关闭自动应答ack，改成手动应答。使用basicQos(perfetch=1)限制每次只发送不超过1条消息到同一个消费者，消费者必须手动反馈告知队列，才会发送下一个。

channel.basic_qos(prefetch_count=1)

3.3、交换机模式

3.3.1、交换机之发布订阅（exchange_type='fanout'）

1、消息队列之RabbitMQ的简单使用

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

# 生产者
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='logs',
                         exchange_type='fanout')

message = "info: Hello World!"
channel.basic_publish(exchange='logs',
                      routing_key='',
                      body=message)
print(" [x] Sent %r" % message)
connection.close()

# 消费者
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='logs',
                         exchange_type='fanout')

result = channel.queue_declare("",exclusive=True)
queue_name = result.method.queue

channel.queue_bind(exchange='logs',
                   queue=queue_name)

print(' [*] Waiting for logs. To exit press CTRL+C')

def callback(ch, method, properties, body):
    print(" [x] %r" % body)


channel.basic_consume(queue=queue_name,
                      auto_ack=True,
                      on_message_callback=callback)

channel.start_consuming()

View Code