Python操作rabbitmq系列:根据主题分配消息

Posted 2020-10-10

接着上一章，使用exchange_type=‘direct‘进行消息传递。这样消息会完全匹配后发送到对应的接收端。现在我们想干这样一件事：

C1获取消息中包含：orange内容的消息，并且消息是由3个单词组成的。

C2获取消息中包含：rabbit内容的消息，并且也是3个单词组成，同时，包含lazy开头的消息，这个消息长度>2

 

在下图中，

quick.orange.rabbit：Q1 Q2都会收到

lazy.orange.elephant：Q1 Q2都会收到

quick.orange.fox ：只会到Q1

lazy.brown.fox：只会到Q2

lazy.pink.rabbit:只会到Q2

quick.orange.male.rabbit:哪儿也去不了

lazy.orange.male.rabbit：只会到Q2

 

消息中的

*：代表一个单词

#：代表N个单词

那么，如何实现呢？

发送端：

import pika
import sys

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

#创建交换机和上一章一样。只是类型变为：exchange_type=‘topic‘
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()

接收端：

import pika
import sys

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

#和发送端保持一致 注意：exchange_type=‘topic‘
channel.exchange_declare(exchange=‘topic_logs‘,
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]...\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()

注意：今天在这里遇到一个特别奇怪的问题。就是接收端程序执行到：

result = channel.queue_declare(exclusive=True)

陷入死循环。这时候应该怎么办呢？

把rabbitmq服务重启即可。本系列文章rabbitmq使用的是：3.6.12。

接下来，我们将讨论利用rabbitmq实现RPC

希望牛小妹能跟上步伐