Rabbitmq打怪升级之路（十二）Headers-头交换机模式

简书：亚武de小文 【原创：转载请注明出处】

头交换机模式（Headers）

LengToo上学.png

RabbitMQ有以下几种工作模式 ：

• Work queues
• Publish/Subscribe
• Routing
• Topic
• Headers
• RPC

---

Header

模型图

[亚武de小文]Headers模型图.png

消息header数据里有一个特殊的键x-match，它有两个值：

• all: 默认值。一个传送消息的header里的键值对和交换机的header键值对全部匹配，才可以路由到对应交换机
• any: 一个传送消息的header里的键值对和交换机的header键值对任意一个匹配，就可以路由到对应交换机

头交换机和主题交换机类似，区别在于：Topic路由值是基于路由键，Headers的路由值基于消息的header数据。 主题交换机路由键只有是字符串,而头交换机可以是整型和哈希值。

参考代码

生产者

• Producer.java

  package com.yawu.xiaowen.header;
  
  import com.rabbitmq.client.*;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.util.Map;
  
  /**
   * Header交换机
   * 生产者
   *
   * @author yawu
   * @date 2019.07.01
   */
  public class Producer {
      private static final Logger LOGGER = LoggerFactory.getLogger(Producer.class);
      private static final String EXCHANGE_NAME = "mq_header_exchange";
  
      public static void execute(Map<String, Object> headerMap) {
          try {
              // RabbitMQ建立连接的管理器
              ConnectionFactory factory = new ConnectionFactory();
              // 设置服务器地址
              factory.setHost("127.0.0.1");
              factory.setUsername("guest");
              factory.setPassword("guest");
  
              // 创建一个连接
              Connection connection = factory.newConnection();
              // 创建一个信道
              Channel channel = connection.createChannel();
  
              String message = "发送信息-headers交换机";
  
              //声明一个Header类型的交换机
              channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.HEADERS);
  
              // 生成发送消息的属性
              AMQP.BasicProperties props = new AMQP.BasicProperties
                      .Builder()
                      .headers(headerMap)
                      .build();
  
              // 向交换机发送消息
              channel.basicPublish(EXCHANGE_NAME, "like.orange.color", null, message.getBytes("UTF-8"));
  
              LOGGER.info("消息发送成功:{}", message);
              channel.close();
              connection.close();
  
          } catch (Exception e) {
              e.printStackTrace();
          }
      }
  
  }
  

消费者

• Consumer01.java

  package com.yawu.xiaowen.header;
  
  import com.rabbitmq.client.*;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.io.IOException;
  import java.util.Map;
  
  /**
   * Header交换机
   * 消费者
   *
   * @author yawu
   * @date 2019.07.01
   */
  public class Consumer01 {
      private static final Logger LOGGER = LoggerFactory.getLogger(Producer.class);
      private static final String EXCHANGE_NAME = "mq_header_exchange";
  
      public static void execute(Map<String, Object> myHeaderMap) {
          try {
              ConnectionFactory factory = new ConnectionFactory();
              factory.setHost("127.0.0.1");
              Connection connection = factory.newConnection();
              Channel channel = connection.createChannel();
  
              //声明一个Headers类型的交换机
              channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.HEADERS);
  
              // 声明一个临时队列
              String queue_name = channel.queueDeclare().getQueue();
  
              // 队列绑定时需要指定参数,注意虽然不需要路由键但仍旧不能写成null，需要写成空字符串""
              channel.queueBind(queue_name, EXCHANGE_NAME, "", myHeaderMap);
  
              LOGGER.info("【Consumer01：" + myHeaderMap + "】 等待消息...");
              Consumer consumer = new DefaultConsumer(channel) {
                  @Override
                  public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                      final String message = new String(body, "UTF-8");
                      LOGGER.info("【Consumer01：" + myHeaderMap + "】接收到的消息 '" + properties.getHeaders() + "':'" + message + "'");
                  }
              };
  
              // 队列一确认消息
              channel.basicConsume(queue_name, true, consumer);
  
          } catch (Exception e) {
              LOGGER.error("an exception was occurred , caused by :{}", e.getMessage());
          }
      }
  }
  
  

测试工具

• HearTest.java

  package com.yawu.xiaowen.header;
  
  
  import org.junit.Test;
  
  import java.util.HashMap;
  import java.util.Map;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.Executors;
  
  public class HeaderTest {
  
      private ExecutorService executorService = Executors.newFixedThreadPool(10);
  
      @Test
      public void header() throws InterruptedException {
  
          // 消费者1：绑定 health=Nice,mentality=Good
          executorService.submit(() -> {
              Map<String, Object> headerMap = new HashMap<>();
              headerMap.put("health", "Nice");
              headerMap.put("mentality", "Good");
              headerMap.put("x-match", "all");
              Producer.execute(headerMap);
          });
  
          // 消费者2：绑定  health=Nice,mentality=Bad
          executorService.submit(() -> {
              Map<String, Object> headerMap = new HashMap<>();
              headerMap.put("health", "Nice");
              headerMap.put("mentality", "Bad");
              headerMap.put("x-match", "any");
              Producer.execute(headerMap);
          });
  
          // 消费者3：绑定  health=Terrible,mentality=Good
          executorService.submit(() -> {
              Map<String, Object> headerMap = new HashMap<>();
              headerMap.put("health", "Terrible");
              headerMap.put("mentality", "Good");
              headerMap.put("x-match", "all");
  //            headerMap.put("x-match","any");
              Producer.execute(headerMap);
          });
  
          Thread.sleep(2 * 1000);
          System.out.println("=============消息01===================");
          // 生产者1 ： health=Nice,mentality=Good,x-match=all
          executorService.submit(() -> {
              Map<String, Object> headerMap = new HashMap<>();
              headerMap.put("health", "Nice");
              headerMap.put("mentality", "Good");
  //            headerMap.put("x-match","all");
              Consumer01.execute(headerMap);
          });
  
          Thread.sleep(5 * 100);
          System.out.println("=============消息02===================");
          // 生产者2 ： health=Nice,x-match=any
          executorService.submit(() -> {
              Map<String, Object> headerMap = new HashMap<>();
              headerMap.put("health", "Nice");
  //            headerMap.put("x-match","any");
              Consumer01.execute(headerMap);
          });
  
          Thread.sleep(5 * 100);
          System.out.println("=============消息03===================");
          // 生产者1 ： health=Terrible,mentality=Bad,x-match=all
          executorService.submit(() -> {
              Map<String, Object> headerMap = new HashMap<>();
              headerMap.put("health", "Terrible");
              headerMap.put("mentality", "Bad");
  //            headerMap.put("x-match","all");
              Consumer01.execute(headerMap);
          });
  
          // sleep 10s
          Thread.sleep(10 * 1000);
      }
  }
  

• 运行HeaderTest测试工具，结果如图：

  
  Headers模式测试结果.png

• 至此，Headers交换机模式学习完毕