zoukankan      html  css  js  c++  java
  • RabbitMQ广播:fanout模式

    一、

    消息的广播需要exchange:exchange是一个转发器,其实把消息发给RabbitMQ里的exchange

    fanout: 所有bind到此exchange的queue都可以接收消息,广播

    direct: 通过routingKey和exchange决定的那个唯一的queue可以接收消息

    topic:所有符合routingKey(此时可以是一个表达式)的routingKey所bind的queue可以接收消息

    headers:通过headers来决定把消息发给哪些queue,用的比较少

    原理图:

    发布者端:

    '''
    发布者publisher
    '''
    import pika
    import sys
    
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.exchange_declare(exchange='logs',  # exchange名字为logs
                             type='fanout')
    # 通过命令行自己输入消息,没输入就是hello world
    message = ' '.join(sys.argv[1:]) or "info: Hello World!"
    # 广播不需要写queue,routing_key为空
    channel.basic_publish(exchange='logs',
                          routing_key='',
                          body=message)
    print("send :", message)
    connection.close()

    订阅者端:

    '''
    订阅者subscriber
    '''
    import pika
    
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.exchange_declare(exchange='logs',
                             type='fanout')
    # 不指定queue名字,rabbit会随机分配一个唯一的queue,
    # exclusive=True会在使用此queue的消费者断开后,自动将queue删除
    # 发送端没有声明queue,为什么接收端需要queue?看上面原理图就明白
    result = channel.queue_declare(exclusive=True)
    # 拿到的随机的queue名字
    queue_name = result.method.queue
    # 需要知道从哪个转发器上去收所以需要绑定
    channel.queue_bind(exchange='logs',
                       queue=queue_name)
    print("Wait for logs...")
    def callback(ch, method, properties, body):
        print("received:", body)
    channel.basic_consume(callback,
                          queue=queue_name,
                          no_ack=True)
    channel.start_consuming()

     运行结果:

    '''
    先启动发布者,再启动订阅者,为什么订阅者收不到信息?
    原理类似于收音机收听广播:
    订阅者相当于收音机,发布者相当于广播信号
    所以这个接收是实时的,订阅者启动之后,才能收到发布者发出的广播
    '''
  • 相关阅读:
    springboot缓存注解——@CacheEvict
    springboot缓存注解——@CachePut
    springboot缓存注解——@Cacheable
    缓存环境
    JSR107缓存规范
    springboot自定义starter
    Linux 最常用命令整理,建议收藏!
    Spring Boot 2.2.0 正式发布,支持 JDK 13!
    厉害了,阿里史上最重要的三段代码…
    代码写成这样,老夫无可奈何!
  • 原文地址:https://www.cnblogs.com/staff/p/9932294.html
Copyright © 2011-2022 走看看