5.RabbitMQ系列之headers交换器

headers exchange是根据消息header值而不是routing key将消息路由到队列的交换器。

生产者在消息头中以键值对的形式添加一些值，并将其发送到headers exchange,

收到消息后，headers exchange尝试将所有或任何（基于x-match的值）header值与绑定到它的所有队列的值匹配。如果找到匹配，则将消息路由到绑定值匹配的队列，如果未找到匹配，则忽略该消息

1. 完整示例代码

编写HeaderExchangeConfig.java配置文件，已声明headers exchange及其绑定

@Configuration

public class HeaderExchangeConfig {

    @Bean

    public HeadersExchange headers() {

        return new HeadersExchange("headers");

    }

    private static class ConsumerConfig {

        @Bean

        public Queue headersAutoDeleteQueue1() {

            return new AnonymousQueue();

        }

        @Bean

        public Queue headersAutoDeleteQueue2() {

            return new AnonymousQueue();

        }

        @Bean

        public Queue headersAutoDeleteQueue3() {

            return new AnonymousQueue();

        }

        @Bean

        public Binding headersBinding1(HeadersExchange headers, Queue headersAutoDeleteQueue1) {

            Map<String, Object> headerMap = new HashMap<>();

            headerMap.put("h1", "Header1");

            headerMap.put("h2", "Header2");

            /**

             * 匹配到任意一个就发送至队列

             */

            return BindingBuilder.bind(headersAutoDeleteQueue1).to(headers).whereAny(headerMap).match();

        }

        @Bean

        public Binding headersBinding2(HeadersExchange headers, Queue headersAutoDeleteQueue2) {

            Map<String, Object> headerMap = new HashMap<>();

            headerMap.put("h1", "Header1");

            headerMap.put("h2", "Header2");

            /**

             * 全部匹配到才会发送至队列

             */

            return BindingBuilder.bind(headersAutoDeleteQueue2).to(headers).whereAll(headerMap).match();

        }

        @Bean

        public Binding headersBinding3(HeadersExchange headers, Queue headersAutoDeleteQueue3) {

            /**

             * 匹配到任意一个就发送至队列， 此处与headersBinding1一致，为了证明其也有fanout模式的功能

             */

            Map<String, Object> headerMap = new HashMap<>();

            headerMap.put("h1", "Header1");

            headerMap.put("h2", "Header2");

            return BindingBuilder.bind(headersAutoDeleteQueue3).to(headers).whereAny(headerMap).match();

        }

    }

}

编写生产者

@Component

public class HeaderSender {

    private RabbitMessagingTemplate rabbitMessagingTemplate;

    public HeaderSender(RabbitMessagingTemplate rabbitMessagingTemplate) {

        this.rabbitMessagingTemplate = rabbitMessagingTemplate;

    }

    public void send() {

        String msg = "Hello World!";

        Map<String, Object> headerMap = new HashMap<>();

        headerMap.put("h1", "Header1");

        headerMap.put("h3", "Header3");

        rabbitMessagingTemplate.convertAndSend("headers", "", msg, headerMap);

        msg = "My Girl!";

        headerMap.clear();

        headerMap.put("h1", "Header1");

        headerMap.put("h2", "Header2");

        rabbitMessagingTemplate.convertAndSend("headers", "", msg, headerMap);

    }

}

编写消费者

@Component

public class HeaderReceiver {

    @RabbitListener(queues = "#{headersAutoDeleteQueue1.name}")

    public void receive1(String in) {

        System.out.println("临时队列1接收到消息:" + in);

    }

    @RabbitListener(queues = "#{headersAutoDeleteQueue2.name}")

    public void receive2(String in) {

        System.out.println("临时队列2接收到消息:" + in);

    }

    @RabbitListener(queues = "#{headersAutoDeleteQueue3.name}")

    public void receive3(String in) {

        System.out.println("临时队列3接收到消息:" + in);

    }

}

编写测试方法

@SpringBootTest

public class RabbitTest {

    @Autowired

    private HeaderSender headerSender;

    @Test

    public void testHeaderSender() {

        headerSender.send();

    }

}

从结果中不难看出，My Girl消息3个队列都匹配到headers, 而Hello Word消息只有队列1与3匹配到，毕竟他没有h2头啦

欢迎关注公众号算法小生或沈健的技术博客

巴特西

5.RabbitMQ系列之headers交换器

1. 完整示例代码

最新文章

热门文章