Python实现的消息队列

一、消息队列介绍

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

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

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

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

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

二、为什么需要MQ

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

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

三、rabbitMQ

3.1、RabbitMQ介绍

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

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

中文文档open in new window

3.2、RabbitMQ下载

rabbitmq的下载地址open in new window

启动RabbitMQ

前台运行

rabbitmq-server

后台运行

brew services start rabbitmq
brew services stop rabbitmq

3.3 、rabbitMQ简单模式

### 生产者


import pika

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

channel.queue_declare(queue='hello')

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

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

### 消费者

import pika

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

channel.queue_declare(queue='hello')

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


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


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

3.3 、参数

应答参数

auto_ack=False
ch.basic_ack(delivery_tag=method.delivery_tag)

持久化参数

#声明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）不管谁忙，都不会多给消息，总是你一个我一个。

在RabbitMQ中，channel.basic_qos(prefetch_count=1)的作用是设置消费者的QoS（Quality of Service）预取计数。QoS是指在多个消费者共享同一个队列的情况下，如何分配消息给消费者的一种机制。

当设置prefetch_count为1时，表示在消息还未被消费者确认处理完成之前，不会向该消费者投递新的消息。也就是说，当消费者处理完当前的消息后，才会接收到下一条消息。

这个设置可以有效地防止消费者在繁忙的情况下一次性接收到过多的消息，从而导致负载过重，无法处理消息。通过限制每个消费者同时处理的消息数量，可以更好地平衡消息的负载，并提高系统的稳定性和可靠性。

需要注意的是，prefetch_count的值应该根据实际的业务情况和系统负载来进行合理的设置。如果设置的值过小，在消息高峰期可能会导致消息处理速度过慢；如果设置的值过大，则可能会导致消息的负载过重，从而影响系统的性能。

# 先关闭自动应答ack，改成手动应答，即auto_ack=False ; ch.basic_ack(delivery_tag=method.delivery_tag)。
channel.basic_qos(prefetch_count=1)

3.4 、交换机之发布订阅

发布订阅和简单的消息队列区别在于，发布订阅会将消息发送给所有的订阅者，而消息队列中的数据被消费一次便消失。所以，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()

3.5、交换机之关键字

# 生产者
import pika

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

channel.exchange_declare(exchange='logs2',
                         exchange_type='direct')

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

# 消费者

import pika
import sys

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

channel.exchange_declare(exchange='logs2',
                         exchange_type='direct')

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


severities = sys.argv[1:]
if not severities:
    sys.stderr.write("Usage: %s [info] [warning] [error]\n" % sys.argv[0])
    sys.exit(1)

for severity in severities:
    channel.queue_bind(exchange='logs2',
                       queue=queue_name,
                       routing_key=severity)

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()

3.6、交换机之通配符

通配符交换机”与之前的路由模式相比，它将信息的传输类型的key更加细化，以“key1.key2.keyN....”的模式来指定信息传输的key的大类型和大类型下面的小类型，让消费者可以更加精细的确认自己想要获取的信息类型。而在消费者一段，不用精确的指定具体到哪一个大类型下的小类型的key，而是可以使用类似正则表达式(但与正则表达式规则完全不同)的通配符在指定一定范围或符合某一个字符串匹配规则的key，来获取想要的信息。

“通配符交换机”（Topic Exchange）将路由键和某模式进行匹配。此时队列需要绑定在一个模式上。符号“#”匹配一个或多个词，符号“”仅匹配一个词。因此“audit.#”能够匹配到“audit.irs.corporate”，但是“audit.”只会匹配到“audit.irs”。（这里与我们一般的正则表达式的“*”和“#”刚好相反，这里我们需要注意一下。）下面是一个解释通配符模式交换机工作的一个样例

上面的交换机制类似于一个国际新闻讯息网站的机制。

# 生产者
import pika

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

channel.exchange_declare(exchange='logs3',
                         exchange_type='topic')

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

# 消费者

import pika
import sys

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

channel.exchange_declare(exchange='logs3',
                         exchange_type='topic')

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



channel.queue_bind(exchange='logs3',
                   queue=queue_name,
                   routing_key="#.news")

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()

四、案例

# 案例1
外卖、商城案例

# 案例2
当今很多游戏系统都使用消息队列来实现实时通信、异步处理等功能。以下是一个实际的游戏系统中使用消息队列的例子：

假设有一个多人在线游戏系统，其中包括多个游戏服务器和多个客户端。为实现游戏中的实时通信和异步处理，可以采用消息队列来实现。

具体实现方式如下：

当玩家进行游戏操作时，如攻击、移动等操作，将操作信息发送到消息队列中。
游戏服务器中有一个消费者程序，负责从消息队列中获取玩家的操作信息，并进行游戏状态的更新和同步。
消费者程序将更新后的游戏状态信息发送到所有在线玩家的客户端中，实现实时通信。
同时，消费者程序也可以将部分操作信息进行异步处理，例如任务奖励、背包物品变化等，将处理结果发送到消息队列中，再由客户端获取处理结果，提高游戏的响应速度和并发处理能力。
通过以上的实现方式，可以实现多人在线游戏中的实时通信和异步处理，提高了游戏的流畅性和稳定性，也提高了玩家的游戏体验。该实现方式在实际开发中也比较常见，可以用于实现多人在线游戏、实时对战、实时聊天等功能。

# Python实现的消息队列

# 一、消息队列介绍

# 二、为什么需要MQ

# 三、rabbitMQ

# 3.1、RabbitMQ介绍

# 3.2、RabbitMQ下载

# 3.3 、rabbitMQ简单模式

# 3.3 、参数

# 3.4 、交换机之发布订阅

# 3.5、交换机之关键字

# 3.6、交换机之通配符

# 四、案例