python系列之 RabbitMQ -- TOPICS

小咪咪 2022-07-29 11:19 163阅读 0赞

## Topics ##

前面的课程中我们对日志系统进行了改进。替换了仅能进行虚拟广播的 fanout 交换，我们使用了一个直接的能进行选择性接收日志的交换（direct)

尽管使用 直接交换 (direct)改进了我们的系统，但仍然有局限性 --  不能够基于多个条件进行路由

在我们的日志系统中我们可能需要订阅不只是基于严重性的的日志，还需要基于发送源的日志。你可能从unix的syslog工具知道这种思想，同时基于严重性（info/warn/crit...)和设备（/auth/corn/kern...)

这会给我们更多的灵活性 -- 我们可能想要同时监听从’cron‘来的关键错误也需要所有 'kern'来的日志

为了在我们的日志系统实现这个功能, 我们需要学习一个更复杂的 topic 交换

## Topic 交换 （Topic exchange） ##

消息发送到一个 topic交换不能是一个任意的 routing\_key -- 它必须是一个用小数点 分割的单词列表。 这个字符可以是任何单词，但是通常是指定一些连接特定消息的功能。一些有效的路由键(routing key）比如：“stock.usd.nyse",

"nyse.vmw","quick.orange.rabbit".  这里可以是你想要路由键的任意字符。最高限制为255字节

绑定key必须在同一个表单中。 Topic交换的背后的逻辑类似直接交换（direct） -- 包含特定路由键值的消息将会分发到所有绑定匹配绑定键值（binding key）的队列中。然后有两个重要的特殊情况绑定键值：

> \* (星)  可代替一个单词

> \# (井) 可代替0个或多个单词

在这个例子中很容易解释：

![Center][]

在这个例子中，我们要发送一个所有描述动物的消息。这个消息将会用一个包含三个单词（2个小数点）的路由键值（routing key）发送， 在路由键值中的第一个单词描述敏捷性、第二个描述颜色、第三个描述种类：“<celerity>.<color>.<species>".

我们创建了三个绑定(bindings): Q1 用"\*.orange.\*"绑定 ， Q2使用“ \*.\*.rabbit" 和 " lazy.\# "

这些绑定可以总结为：

> Q1 对所有 橙色的动物感兴趣

> Q2 希望得到所有 rabbit  和所有 lazy 类型 动物

一条路由key（routing key)设置为 " quick.orange.rabbit" 的消息将要同时分发到两个队列 ， 消息 “ lazy.orange.elephant" 也会发送到两个对了。 另一方“ quick.orange.fox" 将只会发送到第一个队列, ” lazy.brown.fox" 只会发送到第二个。“ lazy.pink.rabbit"  将会分发到第二个1次，尽管它匹配两个绑定。 ” quick.brown.fox" 不匹配任何绑定所以这条消息将会丢弃。

如果我们打破我们的规则只是发送1个或4个单词，不如“ orange” 或者“ quick.orange.male.rabbit" 将会发生什么呢？？ 这样的话这些消息不会匹配任何绑定而且将会丢失

另一方面” lazy.orange.male.rabbit", 尽管有4个单词，仍会匹配最后的绑定并且分发到第二个队列。

## 综合 ##

我们要在我们的日志系统中使用Topic交换, 我们开始一个工作假设日志的路由key（routing key）包含2个单词：“<facility>.<serverity>"

下面是 emit\_log\_topic.py的代码：

#!/usr/bin/env python
    import pika
    import sys
    
    connection = pika.BlockingConnection(pika.ConnectionParameters(
            host='localhost'))
    channel = connection.channel()
    
    channel.exchange_declare(exchange='topic_logs',
                             type='topic')
    
    routing_key = sys.argv[1] if len(sys.argv) > 1 else 'anonymous.info'
    message = ' '.join(sys.argv[2:]) or 'Hello World!'
    channel.basic_publish(exchange='topic_logs',
                          routing_key=routing_key,
                          body=message)
    print(" [x] Sent %r:%r" % (routing_key, message))
    connection.close()

receive\_logs\_topic.py的代码

#!/usr/bin/env python
    import pika
    import sys
    
    connection = pika.BlockingConnection(pika.ConnectionParameters(
            host='localhost'))
    channel = connection.channel()
    
    channel.exchange_declare(exchange='topic_logs',
                             type='topic')
    
    result = channel.queue_declare(exclusive=True)
    queue_name = result.method.queue
    
    binding_keys = sys.argv[1:]
    if not binding_keys:
        sys.stderr.write("Usage: %s [binding_key]...\n" % sys.argv[0])
        sys.exit(1)
    
    for binding_key in binding_keys:
        channel.queue_bind(exchange='topic_logs',
                           queue=queue_name,
                           routing_key=binding_key)
    
    print(' [*] Waiting for logs. To exit press CTRL+C')
    
    def callback(ch, method, properties, body):
        print(" [x] %r:%r" % (method.routing_key, body))
    
    channel.basic_consume(callback,
                          queue=queue_name,
                          no_ack=True)
    
    channel.start_consuming()

接收所有日志：

python receive_logs_topic.py "#"

接收所有从“kern”设备来的日志：

python receive_logs_topic.py "kern.*"

如果你指向获取“critical”的日志：

python receive_logs_topic.py "*.critical"

你可以创建多绑定：

python receive_logs_topic.py "kern.*" "*.critical"

用routing key “ kern.critical" 类型发送日志：

python emit_log_topic.py "kern.critical" "A critical kernel error"

[Center]: /images/20220728/19ecfd2c8ea5441eb4eac29c31862261.png