RabbitMQ交换器

2019-09-28 本文已影响0人 ruoshy

概述

RabbitMQ是实现了高级消息队列协议（AMQP）的开源消息代理软件（亦称面向消息的中间件）。RabbitMQ服务器是用Erlang语言编写的，而集群和故障转移是构建在开放电信平台框架上的。所有主要的编程语言均有与代理接口通讯的客户端库。

交换器

RabbitMQ共有4种交换（Exchange）策略：

Direct （路由模式）
Topic（通配符模式）
Fanout （广播模式）
Header（头模式）

在RabbitMQ中，所有消息生产者提交的消息都会交由Exchange进行分配，Exchange会根据不同的策略将将消息分发到不同的队列中

Direct

DirectExchange的路由策略是将消息绑定到一个DirectExchange上，当一条消息到达DirectExchange时会被转发到与该条routingKey对应的队列上。

Direct 配置如下：

@Configuration
public class RabbitDirectConfig {

    public final static String DIRECT_NAME = "user-direct";

    /**
     * 创建队列 test-queue
     */
    @Bean
    Queue phoneQueue() {
        return new Queue("phone");
    }

    @Bean
    Queue zjsQueue() {
        return new Queue("zjs");
    }

    /**
     * 创建交换策略 direct
     */
    @Bean
    DirectExchange directExchange() {
        return new DirectExchange(DIRECT_NAME, true, false);
    }

    /**
     * 将队列与交换策略进行绑定
     */
    @Bean
    Binding binding1() {
        return BindingBuilder.bind(phoneQueue())
                .to(directExchange()).with("direct");
    }

    @Bean
    Binding binding2() {
        return BindingBuilder.bind(zjsQueue())
                .to(directExchange()).with("direct");
    }
}

Direct 消费者配置如下：

@Component
public class Consumer {

    @RabbitListener(queues = "phone")
    public void phone(String message) {
        System.out.println("phone - " + message);
    }

    @RabbitListener(queues = "zjs")
    public void news(String message) {
        System.out.println("zjs - " + message);
    }
}

使用@RabbitListener指定监听的队列，若存在多个方法同时监听一个队列，会随机发送一条消息给其中一个方法。

Direct 生产者配置如下：

@RunWith(SpringRunner.class)
@SpringBootTest
public class MqApplicationTests {

    @Resource
    private RabbitTemplate template;

    @Test
    public void directPhone() {
        template.convertAndSend("phone", "11111111111");
    }

    @Test
    public void directZjs() {
        template.convertAndSend("zjs", "浙江省");
    }
}

运行结果：

direct

Fanout

FanoutExchange的数据交换策略是把所有到达FanoutExchange的消息转发给所有与他绑定的队列，这种策略下routingKey将不起任何作用。

Fanout 配置如下：

@Configuration
public class RabbitFanoutConfig {

    public final static String FANOUT_NAME = "user-fanout";

    /**
     * 创建队列 test-queue
     */
    @Bean
    Queue phoneQueue() {
        return new Queue("phone");
    }

    @Bean
    Queue zjsQueue() {
        return new Queue("zjs");
    }

    /**
     * 创建交换策略 fanout
     */
    FanoutExchange fanoutExchange() {
        return new FanoutExchange(FANOUT_NAME, true, false);
    }

    /**
     * 将队列与交换策略进行绑定
     */
    @Bean
    Binding binding1() {
        return BindingBuilder.bind(phoneQueue()).to(fanoutExchange());
    }

    @Bean
    Binding binding2() {
        return BindingBuilder.bind(zjsQueue()).to(fanoutExchange());
    }
}

Fanout 消费者配置上同。

Fanout 生产者配置如下：

@RunWith(SpringRunner.class)
@SpringBootTest
public class MqApplicationTests {

    @Resource
    private RabbitTemplate template;

    @Test
    public void fanout() {
        template.convertAndSend(RabbitFanoutConfig.FANOUT_NAME, null, "Hello world！");
    }
}

运行结果：

fanout

Topic

TopicExchange是一种比较灵活的路由策略，他会将所有所有到达TopicExchange的消息根据routingKey分发到一个或多个与他绑定的队列上。

Topic 配置如下：

@Configuration
public class RabbitTopicConfig {

    public final static String TOPIC_NAME = "user_topic";

    /**
     * 创建队列 test-queue
     */
    @Bean
    Queue phoneQueue() {
        return new Queue("phone");
    }

    @Bean
    Queue zjsQueue() {
        return new Queue("zjs");
    }

    /**
     * 创建交换策略 topic
     */
    @Bean
    TopicExchange topicExchange() {
        return new TopicExchange(TOPIC_NAME, true, false);
    }

    /**
     * 将队列与交换策略进行绑定
     */
    @Bean
    Binding binding1() {
        return BindingBuilder.bind(phoneQueue()).to(topicExchange()).with("#.phone.#");
    }

    @Bean
    Binding binding2() {
        return BindingBuilder.bind(zjsQueue()).to(topicExchange()).with("zjs.#");
    }
}

使用通配符作为路由策略，将消息发送到指定的队列中。

Topic 消费者配置上同。

Topic 生产者配置如下：

    @Test
    public void phonTopic() {
        template.convertAndSend(RabbitTopicConfig.TOPIC_NAME, "phone", "11111111111");
    }

运行结果：

topic-1

使用通配符策略：#.phone.#根据routingKey：phone将消息发送到phone队列中。

    @Test
    public void zjsPhoneTopic() {
        template.convertAndSend("zjs.phone", null, "11111111111");
    }

运行结果：

topic-2

使用通配符策略：#.phone.# 和 zjs.#根据routingKey：zjs.phone将消息发送到zjs,phone队列中。

    @Test
    public void zjsPhoneAddressTopic() {
        template.convertAndSend(RabbitTopicConfig.TOPIC_NAME,"zjs.phone.address", "杭州");
    }

运行结果：

topic-3

使用通配符策略：#.phone.# 和 zjs.#根据routingKey：zjs.phone.address将消息发送到zjs,phone队列中。

    @Test
    public void zjsNewTopic() {
        template.convertAndSend(RabbitTopicConfig.TOPIC_NAME, "zjs.new", "新闻");
    }

运行结果：

topic-4

使用通配符策略： zjs.#根据routingKey：zjs.new将消息发送到zjs队列中。

Header

HeaderExchange是一种使用比较少的路由策略，他会根据消息的Header将消息发送到不同的队列上，这种策略与routingKey无关。

Header配置如下：

@Configuration
public class RabbitHeaderConfig {

    public final static String HEADER_NAME = "user_header";

    /**
     * 创建队列 test-queue
     */
    @Bean
    Queue phoneQueue() {
        return new Queue("phone");
    }

    /**
     * 创建交换策略 header
     */
    @Bean
    HeadersExchange topicExchange() {
        return new HeadersExchange(HEADER_NAME, true, false);
    }


    /**
     * 若头存在 name 使用该策略
     */
    @Bean
    Binding binding1() {
        return BindingBuilder.bind(phoneQueue())
                .to(topicExchange())
                .where("name")
                .exists();
    }

    /**
     * 若头存在 name 且值为 ruoshy 使用该策略
     */
    @Bean
    Binding binding2() {
        return BindingBuilder.bind(phoneQueue()).to(topicExchange())
                .where("name")
                .matches("ruoshy");
    }

    /**
     * 若存在 name，age 使用该策略
     */
    @Bean
    Binding binding3() {
        return BindingBuilder.bind(phoneQueue()).to(topicExchange())
                .whereAny("name", "age")
                .exist();
    }

    /**
     * 若头存在 name，age 且值分别为 ruoshy，20 使用该策略
     */
    @Bean
    Binding binding4() {
        Map<String, Object> map = new HashMap<>();
        map.put("name", "ruoshy");
        map.put("age", 20);
        return BindingBuilder.bind(phoneQueue()).to(topicExchange())
                .whereAny(map)
                .match();
    }
}

Header 消费者配置如下：

@Component
public class Consumer {

    @RabbitListener(queues = "phone")
    public void phone(byte[] bytes) {
        System.out.println("phone - " + new String(bytes));
    }
}

Header 生产者配置如下

@RunWith(SpringRunner.class)
@SpringBootTest
public class MqApplicationTests {

    @Resource
    private RabbitTemplate template;

    /**
     * 添加头 name
     */
    @Test
    public void whereExists() {
        Message message = MessageBuilder.withBody("11111111111".getBytes())
                .setHeader("name", null)
                .build();
        template.convertAndSend(RabbitHeaderConfig.HEADER_NAME, null, message);
    }

    /**
     * 添加头 name 值 ruoshy
     */
    @Test
    public void whereMatches() {
        Message message = MessageBuilder.withBody("22222222222".getBytes())
                .setHeader("name", "ruoshy")
                .build();
        template.convertAndSend(RabbitHeaderConfig.HEADER_NAME, null, message);
    }

    /**
     * 添加头 name age
     */
    @Test
    public void whereAnyExist() {
        Message message = MessageBuilder.withBody("33333333333".getBytes())
                .setHeader("name", null)
                .setHeader("age", null)
                .build();
        template.convertAndSend(RabbitHeaderConfig.HEADER_NAME, null, message);
    }

    /**
     * 添加头 name 值 ruoshy
     * 添加头 age 值 20
     */
    @Test
    public void whereAnyMatches() {
        Message message = MessageBuilder.withBody("44444444444".getBytes())
                .setHeader("name", "ruoshy")
                .setHeader("age", 20)
                .build();
        template.convertAndSend(RabbitHeaderConfig.HEADER_NAME, null, message);
    }essage message = MessageBuilder.withBody("44444444444".getBytes())
                .setHeader("name", "ruoshy")
                .setHeader("age", 20)
                .build();
        template.convertAndSend(RabbitHeaderConfig.HEADER_NAME, null, message);
    }
}

运行结果：

header

实例地址

RabbitMQ交换器

概述

交换器

Direct

Fanout

Topic

Header

猜你喜欢

热点阅读