zoukankan      html  css  js  c++  java
  • (转)RabbitMQ学习之Headers交换类型(java)

    http://blog.csdn.net/zhu_tianwei/article/details/40923131

    Headers类型的exchange使用的比较少,它也是忽略routingKey的一种路由方式。是使用Headers来匹配的。Headers是一个键值对,可以定义成Hashtable。发送者在发送的时候定义一些键值对,接收者也可以再绑定时候传入一些键值对,两者匹配的话,则对应的队列就可以收到消息。匹配有两种方式all和any。这两种方式是在接收端必须要用键值"x-mactch"来定义。all代表定义的多个键值对都要满足,而any则代码只要满足一个就可以了。fanout,direct,topic exchange的routingKey都需要要字符串形式的,而headers exchange则没有这个要求,因为键值对的值可以是任何类型。

    1.生产者Producer.Java

    [java] view plain copy
     
     print?
    1. package cn.slimsmart.rabbitmq.demo.headers;  
    2.   
    3. import java.util.Date;  
    4. import java.util.Hashtable;  
    5. import java.util.Map;  
    6.   
    7. import org.springframework.amqp.core.ExchangeTypes;  
    8.   
    9. import com.rabbitmq.client.AMQP;  
    10. import com.rabbitmq.client.AMQP.BasicProperties;  
    11. import com.rabbitmq.client.AMQP.BasicProperties.Builder;  
    12. import com.rabbitmq.client.Channel;  
    13. import com.rabbitmq.client.Connection;  
    14. import com.rabbitmq.client.ConnectionFactory;  
    15.   
    16. public class Producer {  
    17.     private final static String EXCHANGE_NAME = "header-exchange";  
    18.       
    19.     @SuppressWarnings("deprecation")  
    20.     public static void main(String[] args) throws Exception {  
    21.         // 创建连接和频道  
    22.         ConnectionFactory factory = new ConnectionFactory();  
    23.         factory.setHost("192.168.36.102");  
    24.         // 指定用户 密码  
    25.         factory.setUsername("admin");  
    26.         factory.setPassword("admin");  
    27.         // 指定端口  
    28.         factory.setPort(AMQP.PROTOCOL.PORT);  
    29.         Connection connection = factory.newConnection();  
    30.         Channel channel = connection.createChannel();  
    31.           
    32.         //声明转发器和类型headers  
    33.         channel.exchangeDeclare(EXCHANGE_NAME, ExchangeTypes.HEADERS,false,true,null);  
    34.         String message = new Date().toLocaleString() + " : log something";  
    35.           
    36.         Map<String,Object> headers =  new Hashtable<String, Object>();  
    37.         headers.put("aaa", "01234");  
    38.         Builder properties = new BasicProperties.Builder();  
    39.         properties.headers(headers);  
    40.           
    41.         // 指定消息发送到的转发器,绑定键值对headers键值对  
    42.         channel.basicPublish(EXCHANGE_NAME, "",properties.build(),message.getBytes());  
    43.           
    44.         System.out.println("Sent message :'" + message + "'");  
    45.         channel.close();  
    46.         connection.close();  
    47.     }  
    48. }  

    2.消费者Consumer.java

    [java] view plain copy
     
     print?
    1. package cn.slimsmart.rabbitmq.demo.headers;  
    2.   
    3. import java.util.Hashtable;  
    4. import java.util.Map;  
    5.   
    6. import org.springframework.amqp.core.ExchangeTypes;  
    7.   
    8. import com.rabbitmq.client.AMQP;  
    9. import com.rabbitmq.client.Channel;  
    10. import com.rabbitmq.client.Connection;  
    11. import com.rabbitmq.client.ConnectionFactory;  
    12. import com.rabbitmq.client.QueueingConsumer;  
    13.   
    14. public class Consumer {  
    15.     private final static String EXCHANGE_NAME = "header-exchange";  
    16.     private final static String QUEUE_NAME = "header-queue";  
    17.       
    18.     public static void main(String[] args) throws Exception {  
    19.         // 创建连接和频道  
    20.         ConnectionFactory factory = new ConnectionFactory();  
    21.         factory.setHost("192.168.36.102");  
    22.         // 指定用户 密码  
    23.         factory.setUsername("admin");  
    24.         factory.setPassword("admin");  
    25.         // 指定端口  
    26.         factory.setPort(AMQP.PROTOCOL.PORT);  
    27.         Connection connection = factory.newConnection();  
    28.         Channel channel = connection.createChannel();  
    29.           
    30.         //声明转发器和类型headers  
    31.         channel.exchangeDeclare(EXCHANGE_NAME, ExchangeTypes.HEADERS,false,true,null);  
    32.         channel.queueDeclare(QUEUE_NAME,false, false, true,null);  
    33.           
    34.         Map<String, Object> headers = new Hashtable<String, Object>();  
    35.         headers.put("x-match", "any");//all any  
    36.         headers.put("aaa", "01234");  
    37.         headers.put("bbb", "56789");  
    38.         // 为转发器指定队列,设置binding 绑定header键值对  
    39.         channel.queueBind(QUEUE_NAME, EXCHANGE_NAME,"", headers);  
    40.         QueueingConsumer consumer = new QueueingConsumer(channel);  
    41.         // 指定接收者,第二个参数为自动应答,无需手动应答  
    42.         channel.basicConsume(QUEUE_NAME, true, consumer);  
    43.         while (true) {  
    44.             QueueingConsumer.Delivery delivery = consumer.nextDelivery();  
    45.             String message = new String(delivery.getBody());  
    46.             System.out.println(message);  
    47.         }   
    48.     }  
    49. }  

    实例代码:http://download.csdn.net/detail/tianwei7518/8136413

  • 相关阅读:
    Kubernetes组件及网络基础
    mybatis小结-001
    mysql+navicat安装小结
    ibatsi学习总结
    linux 相关的问题
    java 基础 --int 和Integer的区别
    java 接口和抽象类的区别
    java 堆和栈的区别
    springMVC controller配置方式总结
    GC是什么?为什么要有GC
  • 原文地址:https://www.cnblogs.com/telwanggs/p/7124635.html
Copyright © 2011-2022 走看看