微服务框架 SpringCloud微服务架构 16 SpringAMQP 16.6 FanoutExchange

微服务框架

【SpringCloud+RabbitMQ+Docker+Redis+搜索+分布式，系统详解springcloud微服务技术栈课程|黑马程序员Java微服务】

SpringCloud微服务架构

16 SpringAMQP

16.6 FanoutExchange

16.6.1 发布订阅 - Fanout Exchange

Fanout Exchange 会将接收到的消息广播到每一个跟其绑定的queue

16.6.2 直接开干

案例：利用SpringAMQP演示FanoutExchange的使用

示意图如下：

实现思路如下：

1. 在consumer服务中，利用代码声明队列、交换机，并将两者绑定

步骤1：在consumer服务声明Exchange、Queue、Binding

SpringAMQP提供了声明交换机、队列、绑定关系的API，例如：

在consumer服务创建一个类，添加@Configuration注解，并声明FanoutExchange、Queue和绑定关系对象Binding，代码如下：

package cn.itcast.mq.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * ClassName: FanoutConfig
 * date: 2022/10/31 13:23
 *
 * @author DingJiaxiong
 */

@Configuration
public class FanoutConfig {

    //交换机 itcast.fanout
    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange("itcast.fanout");
    }

    //队列1 fanout.queue1
    @Bean
    public Queue fanoutQueue1() {
        return new Queue("fanout.queue1");
    }

    //绑定队列1到交换机
    @Bean
    public Binding fanoutBinding1(Queue fanoutQueue1, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
    }

    //队列2 fanout.queue2
    @Bean
    public Queue fanoutQueue2() {
        return new Queue("fanout.queue2");
    }

    //绑定队列2到交换机
    @Bean
    public Binding fanoutBinding2(Queue fanoutQueue2, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(fanoutQueue2).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
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

OK， 重启下项目

查看RabbitMQ 控制台

交换机

两个队列

查看交换机详情信息

意思就是fanout.queue1 和 fanout.queue2 和交换机已经绑上了

2. 在consumer服务中，编写两个消费者方法，分别监听fanout.queue1和fanout.queue2

@RabbitListener(queues = "fanout.queue1")
public void listenerFanoutQueue1(String msg) {
    System.out.println("消费者接收到fanout.queue1的消息：【" + msg + "】");
}

@RabbitListener(queues = "fanout.queue2")
public void listenerFanoutQueue2(String msg) {
    System.out.println("消费者接收到fanout.queue2的消息：【" + msg + "】");
}
1
2
3
4
5
6
7
8
9

OK，重启服务

3. 在publisher中编写测试方法，向itcast.fanout发送消息

@Test
public void testSendFanoutExchange(){
    //交换机名称
    String exchangeName = "itcast.fanout";
    //消息
    String message = "Hello,every one!";

    //发送消息
    rabbitTemplate.convertAndSend(exchangeName,"",message);
}
1
2
3
4
5
6
7
8
9
10

直接运行这个测试方法

OK，消息发送成功

看看消费者那边的日志

效果很明显，两个消费者都收到了

改一下，没写好。区别一下两个消费者

重启服务，再来一次

OK，效果很明显【这样就实现了一次发送，多个队列都能收到了】

回顾一下

步骤1：在consumer服务声明Exchange、Queue、Binding

在consumer服务创建一个类，添加@Configuration注解，并声明FanoutExchange、Queue和绑定关系对象Binding，代码如下：

步骤2：在consumer服务声明两个消费者

在consumer服务的SpringRabbitListener类中，添加两个方法，分别监听fanout.queue1和fanout.queue2：

步骤3：在publisher服务发送消息到FanoutExchange

在publisher服务的SpringAmqpTest类中添加测试方法：

16.6.3 总结

交换机的作用是什么？

• 接收publisher发送的消息
• 将消息按照规则路由到与之绑定的队列
• 不能缓存消息，路由失败，消息丢失
• FanoutExchange的会将消息路由到每个绑定的队列

声明队列、交换机、绑定关系的Bean是什么？

• Queue
• FanoutExchange
• Binding