Rabbitmq打怪升級之路（十二）Headers-頭交換機模式

簡書：亞武de小文【原創(chuàng)：轉載請注明出處】

頭交換機模式（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鍵值對任意一個匹配，就可以路由到對應交換機

頭交換機和主題交換機類似，區(qū)別在于：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");

            // 創(chuàng)建一個連接
            Connection connection = factory.newConnection();
            // 創(chuàng)建一個信道
            Channel channel = connection.createChannel();

            String message = "發(fā)送信息-headers交換機";

            //聲明一個Header類型的交換機
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.HEADERS);

            // 生成發(fā)送消息的屬性
            AMQP.BasicProperties props = new AMQP.BasicProperties
                    .Builder()
                    .headers(headerMap)
                    .build();

            // 向交換機發(fā)送消息
            channel.basicPublish(EXCHANGE_NAME, "like.orange.color", null, message.getBytes("UTF-8"));

            LOGGER.info("消息發(fā)送成功:{}", 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交換機模式學習完畢