RabbitMQ使用介绍5—Topics

Posted 2020-11-21 venvive

Topics

在之前的教程中我们改进了日志系统，然而使用fanout exchange 仅是可以广播，我们使用direct选择的接收日志。即使使用direct　exchange改善我们的系统，它仍然是有限的，不能基于多个条件进行路由。

在我们的系统中我们想要订阅不仅是严重程度的日志，而且还基于发出日志的资源，您可能从syslog unix工具中了解了这个概念，该工具根据严重性（info / warn / crit ...）和工具（auth / cron / kern ...）路由日志。这将为我们提供很大的灵活性-我们可能只想听来自“ cron”的严重错误，也可以听“ kern”的所有日志。

1. *(star)可以替代一个单词(字)

2. (hash)可以代替一个或者多个单词(字)

如图解释

topic_send.py

# -*- coding: utf-8 -*-
'''
@Time    : 19-11-1 下午3:41
@Author  : jz
@FileName: topic_send.py
@Software: PyCharm
'''
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='topic_logs',exchange_type='topic')
routing_key = sys.argv[1] if len(sys.argv)>2 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()

topic_receive.py

# -*- coding: utf-8 -*-
'''
@Time    : 19-11-1 下午3:41
@Author  : jz
@FileName: topic_receive.py
@Software: PyCharm
'''
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='topic_log',exchange_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]....
" %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(queue=queue_name,
                      on_message_callback=callback,auto_ack=True)
channel.start_consuming()