中间件系列七 RabbitMQ之header exchange(头交换机)用法

雨点打透心脏的1/2处 2022-06-01 11:42 225阅读 0赞

## 1. 概述 ##

header exchange(头交换机)和主题交换机有点相似，但是不同于主题交换机的路由是基于路由键，头交换机的路由值基于消息的header数据。  
主题交换机路由键只有是字符串,而头交换机可以是整型和哈希值  
header Exchange类型用的比较少，但还是知道一点好

## 2. 本文实现功能说明： ##

用到的队列说明：

队列A：绑定交换机参数是：format=pdf,type=report,x-match=all，
    队列B: 绑定交换机参数是：format=pdf,type=log,x-match=any，
    队列C：绑定交换机参数是：format=zip,type=report,x-match=all，

测试场景：

消息1发送交换机的头参数是：format=pdf,type=reprot则消息传送到队列A
    消息2发送交换机的头参数是：format=pdf则消息传送到队列A和队列B
    消息3发送交换机的头参数是：format=zip,type=log则消息没有匹配队列，此消息会被丢弃

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

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

## 3. 生产者代码 ##

主要业务逻辑如下：
    1. 配置连接工厂
    2. 建立TCP连接
    3. 在TCP连接的基础上创建通道
    4. 声明一个headers交换机 
    5. 设置要发送消息的headers值（此值由外部传入）
    6. 发送消息

第4，5，6步代码如下：

// 声明一个headers交换机
    channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.HEADERS);
    String message = "headers-" + System.currentTimeMillis();
    
    // 生成发送消息的属性
    AMQP.BasicProperties props = new AMQP.BasicProperties
            .Builder()
            .headers(headers)
            .build();
    
    // 发送消息，并配置消息
    channel.basicPublish(EXCHANGE_NAME, "", props, message.getBytes("UTF-8"));

**完整代码**  
发送者代码: [HeaderSend.java][]

## 4. 消费者代码 ##

主要业务逻辑如下：

1. 配置连接工厂
    2. 建立TCP连接
    3. 在TCP连接的基础上创建通道
    4. 声明一个headers交换机 
    5. 声明一个临时队列
    6. 将队列绑定到指定交换机上，并设置header的参数（此值由外部传入）
    7. 接收消息并处理

第4，5，6，7步代码如下：

// 声明一个headers交换机
    channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.HEADERS);
    
    // 声明一个临时队列
    String queueName = channel.queueDeclare().getQueue();
    // 将队列绑定到指定交换机上
    channel.queueBind(queueName, EXCHANGE_NAME, "", myHeaders);
    
    System.out.println(" [HeaderRecv ["+ myHeaders +"]] Waiting for messages.");
    
    Consumer consumer = new DefaultConsumer(channel) {
        @Override
        public void handleDelivery(String consumerTag, Envelope envelope,
                                   AMQP.BasicProperties properties, byte[] body) throws IOException {
            String message = new String(body, "UTF-8");
            System.out.println(" [HeaderRecv ["+ myHeaders +"] ] Received '" + properties.getHeaders() + "':'" + message + "'");
        }
    };
    // 接收消息
    channel.basicConsume(queueName, true, consumer);

**完整代码**  
消费者代码：[HeaderRecv.java][]

## 5. 测试： ##

BasicTest  
启动上文例子中的3个消费者，再发送3个测试消息。结果符合上文的预期

@Test
    public void header() throws InterruptedException {
    
        // 消费者1：绑定 format=pdf,type=report
        executorService.submit(() -> { Map<String,Object> headers = new HashMap(); headers.put("format","pdf"); headers.put("type","report"); headers.put("x-match","all"); HeaderRecv.execute(rabbitmq_host, rabbitmq_user, rabbitmq_pwd, headers); }); // 消费者2：绑定 format=pdf,type=log executorService.submit(() -> { Map<String,Object> headers = new HashMap(); headers.put("format","pdf"); headers.put("type","log"); headers.put("x-match","any"); HeaderRecv.execute(rabbitmq_host, rabbitmq_user, rabbitmq_pwd, headers); }); // 消费者3：绑定 format=zip,type=report executorService.submit(() -> { Map<String,Object> headers = new HashMap(); headers.put("format","zip"); headers.put("type","report"); headers.put("x-match","all"); // headers.put("x-match","any"); HeaderRecv.execute(rabbitmq_host, rabbitmq_user, rabbitmq_pwd, headers); }); Thread.sleep(2* 1000); System.out.println("=============消息1==================="); // 生产者1 ： format=pdf,type=reprot,x-match=all executorService.submit(() -> { Map<String,Object> headers = new HashMap(); headers.put("format","pdf"); headers.put("type","report"); // headers.put("x-match","all"); HeaderSend.execute(rabbitmq_host, rabbitmq_user, rabbitmq_pwd, headers); }); Thread.sleep(5* 100); System.out.println("=============消息2==================="); // 生产者2 ： format=pdf,x-match=any executorService.submit(() -> { Map<String,Object> headers = new HashMap(); headers.put("format","pdf"); // headers.put("x-match","any"); HeaderSend.execute(rabbitmq_host, rabbitmq_user, rabbitmq_pwd, headers); }); Thread.sleep(5* 100); System.out.println("=============消息3==================="); // 生产者3 ： format=zip,type=log,x-match=all executorService.submit(() -> { Map<String,Object> headers = new HashMap(); headers.put("format","zip"); headers.put("type","log"); // headers.put("x-match","all"); HeaderSend.execute(rabbitmq_host, rabbitmq_user, rabbitmq_pwd, headers); }); // sleep 10s Thread.sleep(10 * 1000); }

输出结果

[HeaderRecv [{x-match=any, format=pdf, type=log}]] Waiting for messages.
     [HeaderRecv [{x-match=all, format=pdf, type=report}]] Waiting for messages.
     [HeaderRecv [{x-match=all, format=zip, type=report}]] Waiting for messages.
    =============消息1，被2个消费者接受===================
     [HeaderSend] Sent '{format=pdf, type=report}':'headers-1516257699815'
     [HeaderRecv [{x-match=all, format=pdf, type=report}] ] Received '{format=pdf, type=report}':'headers-1516257699815'
     [HeaderRecv [{x-match=any, format=pdf, type=log}] ] Received '{format=pdf, type=report}':'headers-1516257699815'
    =============消息2，被1个消费者接受===================
     [HeaderSend] Sent '{format=pdf}':'headers-1516257700323'
     [HeaderRecv [{x-match=any, format=pdf, type=log}] ] Received '{format=pdf}':'headers-1516257700323'
    =============消息3，被1个消费者接受===================
     [HeaderSend] Sent '{format=zip, type=log}':'headers-1516257700817'
     [HeaderRecv [{x-match=any, format=pdf, type=log}] ] Received '{format=zip, type=log}':'headers-1516257700817'

## 6. 代码 ##

上文的详细代码主要如下：  
发送者代码: [HeaderSend.java][]  
消费者代码：[HeaderRecv.java][]  
测试代码：[BasicTest.java][]的方法 header()  
所有的详细代码见[github代码，请尽量使用tag v0.11，不要使用master，因为master一直在变，不能保证文章中代码和github上的代码一直相同][github_tag v0.11_master_master_github]

[HeaderSend.java]: https://github.com/hryou0922/spring_boot/blob/v0.11/rabbitmq/src/main/java/com/hry/spring/rabbitmq/basic/header/HeaderSend.java
[HeaderRecv.java]: https://github.com/hryou0922/spring_boot/blob/v0.11/rabbitmq/src/main/java/com/hry/spring/rabbitmq/basic/header/HeaderRecv.java
[BasicTest.java]: https://github.com/hryou0922/spring_boot/blob/v0.11/rabbitmq/src/test/java/com/hry/spring/rabbitmq/basic/BasicTest.java
[github_tag v0.11_master_master_github]: https://github.com/hryou0922/spring_boot/tree/v0.11/rabbitmq/src/main/java/com/hry/spring/rabbitmq/basic/header