基于Python语言使用RabbitMQ消息队列(五)

释放双眼,带上耳机,听听看~!

Topics

在前面教程中我们改进了日志系统,相比较于使用fanout类型交易所只能傻瓜一样地广播,我们用direct获得了选择性接收日志的能力。

虽然使用direct类型交易所改进了我们的系统,但它仍然有所限制——它不能做基于多重条件(multiple criteria)的路由。
在日志系统中我们可能不只是想要基于严重级别来订阅日志。也想要基于产生日志的来源。你可能从unix的系统日志工具(syslog unix tool)知道了这个概念,它就是基于严重级别 (info/warn/crit…)和设施(auth/cron/kern…)来路由日志的。
这会给我们很大灵活性——我们可能只想监听来自于‘cron’的严重错误(critical errors)和来自‘kern’的所有日志。
为了在我们的日志系统中实现这个功能我们需要了解更复杂的topic类型交易所。

Topic 交易所

发往topic类型交易所的消息不能只有一个独断的路由键(routing_key)——它必须是个词汇列表,词与词之间由‘.’来界定。可以是任何词汇,但通常它们指定了一些与消息相关联的特性。一些有效的路由键例子:”stock.usd.nyse”, “nyse.vmw”, “quick.orange.rabbit”,只要你想,在路由键里可以加尽可能多的词汇,上限是255 bytes。

绑定键必须是同样的形式。topic交易所背后的逻辑与direct类似——使用特定路由键发送的消息会传递给所有拥有匹配的绑定键的队列。但对于绑定键有两种重要的特殊情形:

“*” (star) 正好代替一个词.
“#” (hash) 能代替零个或多个词.

可以很容易地用一个例子解释:
在这个例子中我们发送的消息都是描述动物的。将要发送的消息的路由键包含三个词(两个点号)。第一个词描述的是敏捷性,第二个词是颜色,第三个词是种类:“<敏捷性>.<颜色>.<种类>”。

我们创建了三个绑定: Q1 绑定键是 “.orange.” , Q2 是 “..rabbit” 和 “lazy.#”.

这些绑定可以总结为:

  • Q1 对所有橙色(orange)动物感兴趣.
  • Q2想要知道关于rabbits的每件事情, 和关于lazy 类型动物的所有.

路由键设置为 “quick.orange.rabbit” 的消息会同时传送给两个队列。消息”lazy.orange.elephant” 也会传送给两个队列。”quick.orange.fox”会传送给第一个队列。”lazy.brown.fox”只会传送给第二个队列。”quick.brown.fox” 不匹配任何绑定,所以它会被忽略。

如果我们破坏约定,发送带有一个词或四个词的消息绑定,像”orange”或者”quick.orange.male.rabbit”,会发生什么呢?当然,由于这些消息不匹配任何绑定会被丢失。

另一方面即便 “lazy.orange.male.rabbit”, 有4个词,但它匹配最后一个绑定,所以它会被传送给第二个队列。

Topic 交易所

Topic 交易所很强大并且可以拥有其他类型交易所的的表现
当一个队列使用了“#”绑定键它就会接收所有消息,不管是什么路由键。此时就像是fanout类型交易所。 当在绑定中没有使用特殊字符 “*”
(star) 和 “#” (hash) topic交易所就跟direct交易所一样。

整合

我们将在日志系统中使用topic交易所。假设日志的路由键有两个词 “场所.严重级别”。

代码同之前的教程中几乎相同

emit_log_topic.py完整代码:


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
1#!/usr/bin/env python
2import pika
3import sys
4
5connection = pika.BlockingConnection(pika.ConnectionParameters(host=&#x27;localhost&#x27;))
6channel = connection.channel()
7
8channel.exchange_declare(exchange=&#x27;topic_logs&#x27;,
9                         type=&#x27;topic&#x27;)
10
11routing_key = sys.argv[1] if len(sys.argv) &gt; 2 else &#x27;anonymous.info&#x27;
12message = &#x27; &#x27;.join(sys.argv[2:]) or &#x27;Hello World!&#x27;
13channel.basic_publish(exchange=&#x27;topic_logs&#x27;,
14                      routing_key=routing_key,
15                      body=message)
16print(&quot; [x] Sent %r:%r&quot; % (routing_key, message))
17connection.close()
18

receive_log_topic.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
1#!/usr/bin/env python
2import pika
3import sys
4
5connection = pika.BlockingConnection(pika.ConnectionParameters(host=&#x27;localhost&#x27;))
6channel = connection.channel()
7
8channel.exchange_declare(exchange=&#x27;topic_logs&#x27;,
9                         type=&#x27;topic&#x27;)
10
11result = channel.queue_declare(exclusive=True)
12queue_name = result.method.queue
13
14binding_keys = sys.argv[1:]
15if not binding_keys:
16    sys.stderr.write(&quot;Usage: %s [binding_key]...\n&quot; % sys.argv[0])
17    sys.exit(1)
18
19for binding_key in binding_keys:
20    channel.queue_bind(exchange=&#x27;topic_logs&#x27;,
21                       queue=queue_name,
22                       routing_key=binding_key)
23
24print(&#x27; [*] Waiting for logs. To exit press CTRL+C&#x27;)
25
26def callback(ch, method, properties, body):
27    print(&quot; [x] %r:%r&quot; % (method.routing_key, body))
28
29channel.basic_consume(callback,
30                      queue=queue_name,
31                      no_ack=True)
32
33channel.start_consuming()
34

我在Ubuntu终端开启了四个控制台,其中三个接收日志,日志的接收和发送情况如图:
接收所有日志

接收级别为critical日志

接收产生自kernel日志

日志的发送

给TA打赏
共{{data.count}}人
人已打赏
安全网络

CDN安全市场到2022年价值76.3亿美元

2018-2-1 18:02:50

安全技术

Dart 库预览

2022-1-12 12:36:11

个人中心
购物车
优惠劵
今日签到
有新私信 私信列表
搜索