接着上一章，使用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
希望牛小妹能跟上步伐
网友评论