RabbitMQ的交换机（原理及代码实现）

1.交换机类型

• Fanout Exchange（扇形）
• Direct Exchange（直连）
• opic Exchange（主题）
• Headers Exchange（头部）

2.Fanout Exchange

2.1 简介

Fanout 扇形的，散开的； 扇形交换机

投递到所有绑定的队列，不需要路由键，不需要进行路由键的匹配，相当于广播、群发；
如下图所示

P代表provider提供者，X代表exchange交换机，第三部分代表队列

2.2 代码示例

2.2.1 扇形交换机与队列

定义交换机主要由3部分组成，

• 1.定义交换机
• 2.定义队列
• 3.绑定交换机

@Configuration
public class RabbitConfig {
    //rabbitmq三部曲
    //1.定义交换机
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("exchange.fanout");
    }
    //2.定义队列
    //此处定义两个队列
    @Bean
    public Queue queueA(){
        return new Queue("queue.fanout.a");
    }
    @Bean
    public Queue queueB(){
        return new Queue("queue.fanout.b");
    }
    //3.绑定交换机
    @Bean
    public Binding bindingA(FanoutExchange fanoutExchange,Queue queueA){
        //将队列A绑定到扇形交换机
        return BindingBuilder.bind(queueA).to(fanoutExchange);
    }
    @Bean
    public Binding bindingB(FanoutExchange fanoutExchange,Queue queueB){
        //将队列A绑定到扇形交换机
        return BindingBuilder.bind(queueB).to(fanoutExchange);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

2.2.2 模拟provider发送消息

发送消息模拟

@Component
@Slf4j
public class MessageService {
    @Resource
    private RabbitTemplate rabbitTemplate;

    public void senMsg(){
        //定义消息
        String msg="hello world";
        //发消息
        Message message= new Message(msg.getBytes());
        rabbitTemplate.convertAndSend("exchange.fanout","",message);
        log.info("消息发送完毕，发送时间为:{}", new Date());
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

2.2.3 接受扇形交换机信息

3.Direct Exchange

根据路由路由键（Routing Key）精确匹配（一模一样）进行路由消息队列；
与扇形交换机不同的是，直连交换机必须要绑定key

3.2代码示例

@Configuration
public class RabbitConfig {
    //rabbitmq三部曲
    //1.定义交换机
    @Bean
    public DirectExchange directExchange(){
        return ExchangeBuilder.directExchange("exchange.direct").build();
    }
    //2.定义队列
    //此处定义两个队列
    @Bean
    public Queue queueA(){
        return new Queue("queue.direct.a");
    }
    @Bean
    public Queue queueB(){
        return new Queue("queue.direct.b");
    }
    //3.绑定交换机
    @Bean
    public Binding bindingA(DirectExchange directExchange,Queue queueA){
        //将队列A绑定到扇形交换机
        return BindingBuilder.bind(queueA).to(directExchange).with("error");
    }
    @Bean
    public Binding bindingB1(DirectExchange directExchange,Queue queueB){
        //将队列A绑定到扇形交换机
        return BindingBuilder.bind(queueB).to(directExchange).with("error");
    }
    @Bean
    public Binding bindingB2(DirectExchange directExchange,Queue queueB){
        //将队列A绑定到扇形交换机
        return BindingBuilder.bind(queueB).to(directExchange).with("info");
    }
    @Bean
    public Binding bindingB3(DirectExchange directExchange,Queue queueB){
        //将队列A绑定到扇形交换机
        return BindingBuilder.bind(queueB).to(directExchange).with("warning");
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

4. Topic Exchange

通配符匹配，相当于模糊匹配；

#匹配多个单词，用来表示任意数量（零个或多个）单词
*匹配一个单词（必须有一个，而且只有一个），用.隔开的为一个单词

5.Headers Exchange

基于消息内容中的headers属性进行匹配；

用的比较少