怪兽和他的朋友们

Monster

0%

Python 和 RabbitMQ 的那点事

发表于 分类于
本文字数： 9.9k 阅读时长 ≈ 9 分钟

关于 Python 系列其他文章的传送门

  1. 《 Python 基础语法 》

  2. 《 Python 进阶：网络、多线程、异步、数据库、WEB 》

  3. 《 Python 和 RabbitMQ 的那点事 》　　　　　　您当前所在位置

  4. 《 Python 和 Redis 的那点事 》

  5. 《 Python 爬虫基础和包 XPath、BeautifulSoup4、PhantomJS、Selenium 》

  6. 《 边爬豆瓣，边学习 Python 爬虫框架 Scrapy 》

  7. 《 用了 Scrapy-Redis 分布式爬虫后，一个能打的都没有！ 》

  8. 《 全栈开发入门 JavaScript、React 和 Django 》

  9. 《 Django REST framework 的那点事 》

logo

Python 和 RabbitMQ 的那点事

RabbitMQ 是由 LShift 提供的一个 Advanced Message Queuing Protocol ( AMQP ) 的开源协议实现，由以高性能、健壮以及可伸缩性出名的 Erlang 语言写成，因此也是继承了这些优点。

优点：很成熟，久经考验，应用广泛；文档详细，客户端丰富，几乎常用语言都有 RabbitMQ 的开发库。

docker 启动 RabbitMQ

1
2
3
4
5
6
7
8
9
10
# 启动 RabbitMQ
# 设置管理员账户
# rabbitmq:3-management 表示启动 WEB 管理插件
docker run -d --hostname my-rabbit --name this-rabbit \
                                   -e RABBITMQ_DEFAULT_USER=monster \
                                   -e RABBITMQ_DEFAULT_PASS=123qwe \
                                   -p 5672:5672 \
                                   -p 15672:15672 \
                                   -p 25672:25672 \
                                   rabbitmq:3-management

默认端口：

  • 5672：协议端口；
  • 15672：+10000 为 WEB 管理端口；
  • 25672：+20000 为 内部节点和客户端工具通信用。

非 docker 时：

  • 插件管理：
    • # rabbitmq-plugins list：列出所有可用插件；
    • # rabbitmq-plugins enable rabbitmq_management：启动 WEB 管理插件。
  • 用户管理：
    • rabbitmqctl add_user username password：添加用户；
    • rabbitmqctl delete_user username：删除用户；
    • rabbitmqctl change_password username newpassword：更改密码；
    • rabbitmqctl set_user_tags username tag：设置权限 Tags，其实就是分配组。

原理及应用：
https://www.rabbitmq.com/getstarted.html

名词解释

  • Server：服务器，接收客户端连接，实现消息队列及路由功能的进程（服务），也称为消息代理。注意：客户端包括生产者和消费者；
  • Connection：网络物理连接；
  • Channel：一个连接允许多个客户端连接；
  • Exchange：交换器，接收生产者发来的消息，决定如何路由给服务器中的队列。常用的类型有：
    • direct( point-to-point ): 直接；
    • topic( publish-subscribe ): 话题订阅；
    • fanout( multicast ): 扇出。；
  • Message：消息；
  • Message Queue：消息队列，数据的存储载体；
  • Bind：绑定，建立消息队列和交换器之间的关系，也就是说交换器拿到数据，把什么样的数据送给哪个队列；
  • Virtual Host：虚拟主机，一批交换机、消息队列和相关对象的集合。为了多用户互不干 扰，使用虚拟主机分组交换机、消息队列；
  • Topic：主题、话题；
  • Broke：可等价为 Server。

logo

pika 库

Pika 是纯 Python 实现的支持 AMQP 协议的库。

1
pip install pika

应用示例

RabbitMQ 支持多种应用模型，下面举例各种应用如何通过 pika 库使用 RabbitMQ。

先通过 ip:15672 WEB 管理界面创建测试的虚拟主机 test，之后的举例都使用它。

1. 队列应用

rabbitmq_1.png

这种模式就是最简单的生产者消费者模型，消息队列就是一个 FIFO(先进先出) 的队列。

生产者 send.py：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import pika
import time

# 建立连接
params = pika.URLParameters('amqp://monster:123qwe@192.168.50.30:5672/test')
connection = pika.BlockingConnection(params)  # 阻塞连接
channel = connection.channel()

channel.queue_declare(queue='hello')  # # queue 命名为 hello。必须的，没有就创建

for i in range(40):  # 产生 40 条消息
    channel.basic_publish(exchange='',  # 不指定则使用缺省 exchange,
                          routing_key='hello',  # 使用缺省 exchange 时， routing-key 必须指定，指向上面创建的 queue
                          body=bytes(f'data{i:02}', encoding='utf8'))
    # time.sleep(0.1)

print("[x] Sent 'Hello world!'")
connection.close()

消费者 receive.py：

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

# 建立连接
params = pika.URLParameters('amqp://monster:123qwe@192.168.50.30:5672/test')
connection = pika.BlockingConnection(params)  # 阻塞连接
channel = connection.channel()

channel.queue_declare(queue='hello')  # queue 命名为 hello。必须的，没有就创建


# 回调函数
def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)

# 消费
channel.basic_consume(queue='hello', on_message_callback=callback, auto_ack=False)

# 循环取队列数据
print(' [*] waiting for messages. To exit press CTRL+C')
channel.start_consuming()  # ctl+c

2. 轮询队列

rabbitmq_2.png

这种工作模式是一种竞争工作方式，对某一个消息来说，通过轮询的方式，多个消费者只能有一个拿走它。

把 recieve.py ，再启动一个消费者，就有两个消费者。

注意：虽然上面的图中没有画出 exchange，但其实也用到了缺省的 exchange

3. 发布、订阅模式 Publish/Subscribe

rabbitmq_3.png

订阅者和消费者之间有一个 exchange(交换机)，生产者把消息发送到 exchange，由 exchange 来决定如何发送到 queue。

消费者和生产者谁都可以去创建 queue，谁创建都行。

生产者 send.py：

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

# 建立连接
params = pika.URLParameters('amqp://monster:123qwe@192.168.50.30:5672/test')
connection = pika.BlockingConnection(params)  # 阻塞连接
channel = connection.channel()
# 增加：声明一个交换机 exchange
channel.exchange_declare(exchange='logs',
                         exchange_type='fanout')  # 类型为广播（扇出）

# 生产者不创建 queue，由消费者创建


for i in range(40):  # 产生 40 条消息
    channel.basic_publish(exchange='logs',  # 把消息发送到指定的 exchange 上
                          routing_key='',  # 如果 exchange 是 fanout，也就是广播了，routing_key 就无所谓是什么了
                          body=bytes("pc-data{: 02}".format(i), encoding='utf8'))
    time.sleep(1)

print("[x] Sent 'Hello world!'")
connection.close()

消费者 receive.py：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
import pika

# 建立连接
params = pika.URLParameters('amqp://monster:123qwe@192.168.50.30:5672/test')
connection = pika.BlockingConnection(params)  # 阻塞连接
channel = connection.channel()
# 增加：声明一个交换机 exchange
channel.exchange_declare(exchange='logs',
                         exchange_type='fanout')  # 类型为广播（扇出）

# 生成 2 个 queue
q1 = channel.queue_declare('myQ1', exclusive=True)
q2 = channel.queue_declare('myQ2', exclusive=True)
q1name = q1.method.queue
q2name = q2.method.queue
# 看一下 q 的名字
print(q1name, q2name)

# 把 exchange 和 queue 绑定 起来
channel.queue_bind(exchange='logs', queue='myQ1')
channel.queue_bind(exchange='logs', queue='myQ2')


# 回调函数
def callback(ch, method, properties, body):
    print(f" [x] Received {method} {body}")

# 消费
channel.basic_consume('myQ1', on_message_callback=callback, auto_ack=True)
channel.basic_consume('myQ2', on_message_callback=callback, auto_ack=True)

# 循环取队列数据
print(' [*] waiting for messages. To exit press CTRL+C')
channel.start_consuming()  # ctl+c

上面的代码，如果先启动生产者会怎么样呢？

因为生产者没有绑定 queue，如果先启动生产者，那么当生产者把消息发给路由器后，没有 queue 来接收，这部分数据就丢失了。

可以把 exchange 想象成交换机，只负责转发数据，没人接收就丢弃

4. 路由 Routing

rabbitmq_4.png

路由其实就是数据经过 exchange 的时候，通过匹配规则，对数据进行过滤，决定数据的去向。

而 fanout 不做数据过滤。

生产者 send.py：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
import pika
import time
import random

exchange_name = 'color'
colors = ('orange', 'black', 'green')

# 创建连接
params = pika.URLParameters('amqp://monster:123qwe@192.168.50.30:5672/test')
connection = pika.BlockingConnection(params)
channel = connection.channel()
channel.exchange_declare(exchange=exchange_name,  # 不是用默认 exchange
                         exchange_type='direct')  # 类型为路由
for i in range(40):
    rk = colors[random.randint(0, 2)] # 随机选一个颜色作为 routing_key 名
    msg = "pc-data{:02}".format(i)
    channel.basic_publish(exchange=exchange_name,  # 发送到指定 exchange
                          routing_key=rk,  # 指定 routing_key
                          body=bytes(msg, encoding='utf8'))
    print(rk, msg)
    time.sleep(2)

connection.close()

消费者 receive.py：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
import pika

exchangename = 'color'
colors = ('orange', 'black', 'green')

# 建立连接
params = pika.URLParameters('amqp://monster:123qwe@192.168.50.30:5672/test')
connection = pika.BlockingConnection(params)  # 阻塞连接
channel = connection.channel()
# 增加：声明一个交换机 exchange
channel.exchange_declare(exchange=exchangename,
                         exchange_type='direct')  # 类型为路由

# 生成 2 个 queue
q1 = channel.queue_declare('myQ1', exclusive=True)
q2 = channel.queue_declare('myQ2', exclusive=True)
q1name = q1.method.queue
q2name = q2.method.queue
# 看一下 q 的名字
print(q1name, q2name)

# 把 exchange 和 queue 绑定 起来，一定要绑定 routing_key
channel.queue_bind(exchange=exchangename, queue='myQ1', routing_key=colors[0])  # 1 个匹配 orange
channel.queue_bind(exchange=exchangename, queue='myQ2', routing_key=colors[1])  # 下面 2 个匹配 black 和 green
channel.queue_bind(exchange=exchangename, queue='myQ2', routing_key=colors[2])


# 回调函数
def callback(ch, method, properties, body):
    print(f" [x] Received {method.routing_key} {body}")


# 消费
channel.basic_consume('myQ1', on_message_callback=callback, auto_ack=True)
channel.basic_consume('myQ2', on_message_callback=callback, auto_ack=True)

# 循环取队列数据
print(' [*] waiting for messages. To exit press CTRL+C')
channel.start_consuming()  # ctl+c

之后在 WEB 管理界面中，点击 Exchanges => 选择 color 交换机 => 就可以看到 Routing key 绑定信息。

5. 话题 Topic

rabbitmq_5.png

Topic 的 routing_key 必须使用 . 点号分割的单词组成！！！

支持使用通配符：比如：'phone.*', '*.red'

  • * 表示严格的一个单词；
  • # 表示 0 个或者多个单词。

如果 queue 绑定的 routing_key 只是一个 #，那么这个 queue 其实可以接收所有的消息，类似于 fanout。

如果没有使用任何通配符，效果类似于 direct。

生产者 send.py：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
import pika
import time
import random

exchange_name = 'products'
topics = ('phone.*', '*.red')
product_type = ('phone', 'pc', 'tv')
colors = ('red', 'green', 'blue')

# 创建连接
params = pika.URLParameters('amqp://monster:123qwe@192.168.50.30:5672/test')
connection = pika.BlockingConnection(params)
channel = connection.channel()
# 增加：声明一个交换机 exchange，类型为 Topic
channel.exchange_declare(exchange=exchange_name,  # 不是用默认 exchange
                         exchange_type='topic')  # 类型为 Topic

for i in range(40):
    # 随机选一个类型和颜色组合起来，必须使用 . 号分割作，为 routing_key 名
    rk = f'{product_type[random.randint(0,2)]}.{colors[random.randint(0, 2)]}'
    msg = "{} {:02}".format(rk, i)
    channel.basic_publish(exchange=exchange_name,  # 发送到指定 exchange
                          routing_key=rk,  # 指定 routing_key
                          body=bytes(msg, encoding='utf8'))
    print(msg)
    time.sleep(0.2)

connection.close()

消费者 receive.py：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
import pika

exchange_name = 'products'
topics = ('phone.*', '*.red')
producttype = ('phone', 'pc', 'tv')
colors = ('red', 'green', 'blue')

# 建立连接
params = pika.URLParameters('amqp://monster:123qwe@192.168.50.30:5672/test')
connection = pika.BlockingConnection(params)  # 阻塞连接
channel = connection.channel()
# 增加：声明一个交换机 exchange，类型为 Topic
channel.exchange_declare(exchange=exchange_name,  # 不是用默认 exchange
                         exchange_type='topic')  # 类型为 Topic

# 生成 2 个 queue
q1 = channel.queue_declare('myQ1', exclusive=True)
q2 = channel.queue_declare('myQ2', exclusive=True)
q1name = q1.method.queue
q2name = q2.method.queue
# 看一下 q 的名字
print(q1name, q2name)

# 把 exchange 和 queue 绑定 起来，一定要绑定 routing_key
channel.queue_bind(exchange=exchange_name, queue='myQ1', routing_key=topics[0])  # 匹配 'phone.*'
channel.queue_bind(exchange=exchange_name, queue='myQ2', routing_key=topics[1])  # 匹配 '*.red'


# 回调函数
def callback(ch, method, properties, body):
    print(f" [x] {body}")


# 消费
channel.basic_consume('myQ1', on_message_callback=callback, auto_ack=True)
channel.basic_consume('myQ2', on_message_callback=callback, auto_ack=True)

# 循环取队列数据
print(' [*] waiting for messages. To exit press CTRL+C')
channel.start_consuming()  # ctl+c

Topic 其实就是更加高级的路由，支持模式匹配而已。

消息队列的作用

  1. 系统间解耦；
  2. 解决生产者、消费者速度匹配。

由于稍微上规模的项目都会分层、分模块开发，模块间或系统间尽量不要直接耦合，需要开放公共接口供别的模块或系统调用，而调用可能触发并发问题，为了缓冲和解耦，往往采用中间件技术。

RabbitMQ 只是消息中间件中的一种应用程序