zoukankan      html  css  js  c++  java
  • Rabbitmq -- direct

    一、前言

      RabbitMQ还支持根据关键字发送,即:队列绑定关键字,发送者将数据根据关键字发送到消息exchange。direct类型的Exchange路由规则也很简单,它会把消息路由到那些binding key与routing key完全匹配的Queue中。

      

    二、Exchange direct

      direct类型的Exchange路由规则是完全匹配binding key与routing key

      生产端:

    # -*- coding: UTF-8 -*-
    
    import pika
    
    # 创建一个连接
    connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='localhost'))
    
    # 创建一个管道
    channel = connection.channel()
    
    # 声明exchange 及类型
    channel.exchange_declare(exchange='direct_log',
                             exchange_type='direct')
    
    # 输入信息,格式为 info from info test 类似
    input_data = input('>>:').strip()
    
    # 将输入的信息以空格为分割,转换为列表
    data_list = input_data.split(' ')
    
    # 三元运算,如果输入信息存在,就使用输入的信息data_list[0],否则用 'info'
    severity = data_list[0] if len(data_list) > 1 else 'info'
    
    message = ' '.join(data_list[2:]) or 'hello,world!'
    
    # 这里的routing_key就是 data_list[0] 或 'info'
    channel.basic_publish(exchange='direct_log',
                          routing_key=severity,
                          body=message)
    print('[x] Sent %r:%r' % (severity, message))
    
    connection.close()
    

      消费端:

    # -*- coding: UTF-8 -*-
    
    import pika
    
    connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='localhost'))
    
    channel = connection.channel()
    
    # 声明exchange 及类型
    channel.exchange_declare(exchange='direct_log',
                             exchange_type='direct')
    
    
    result = channel.queue_declare(exclusive=True)
    
    queue_name = result.method.queue
    
    # 在此我们定义一些列表,列表内容如下
    # 这3个列表分别用来测试和routing_key匹配情况
    # 第一种只允许接收info的信息
    # 第二种允许接收info 和 warning的信息
    # 第三种允许接收 info、warning和error的信息
    
    # severities = ['info']
    severities = ['info', 'warning']
    # severities = ['info', 'warning', 'error']
    
    for severity in severities:
        channel.queue_bind(exchange='direct_log',
                           queue=queue_name,
                           routing_key=severity)
    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()
    

      在目前的情况下,severities = ['info', 'warning'],消费端能接收info 和warning的信息。 但是 error的信息不能收到

      生产端输入: info from info test /   warning from warning test

      

      同理如果只有 ['info'] 就只能收 info 的信息, 如果全部都包括,则 info、warning和 error的信息都能收到

      

  • 相关阅读:
    Windows共享上网的详细设置
    使用树莓派实现微信远程监控
    数据结构——队列及循环队列
    springmvc web 大文件上传源代码
    springboot web 大文件上传源代码
    jsp web 大文件上传源代码
    csharp web 大文件上传源代码
    c# web 大文件上传源代码
    .net web 大文件上传源代码
    asp.net web 大文件上传源代码
  • 原文地址:https://www.cnblogs.com/bigberg/p/8206123.html
Copyright © 2011-2022 走看看