zoukankan      html  css  js  c++  java
  • Python操作rabbitmq系列(五):根据主题分配消息

    接着上一章,使用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]... " % 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

    希望牛小妹能跟上步伐

  • 相关阅读:
    告诉自己的十大忠告
    深度学习框架
    深度学习-Caffe编译测试的小总结
    深度学习-Windows平台下的Caffe编译教程
    C#中使用多款LevelDB.Net封装测试性能
    LevelDB初体验测试
    机器学习-分类器-Adaboost原理
    机器学习-分类器-级联分类器训练(Train CascadeClassifier )
    C# Directory.GetFiles()获取多个类型格式的文件
    部分Android或IOS手机拍照后照片被旋转的问题
  • 原文地址:https://www.cnblogs.com/zijiyanxi/p/7663509.html
Copyright © 2011-2022 走看看