4.1 Fanout（广播）
Fanout，英文翻译是扇出，在MQ中叫广播更合适。
在广播模式下，消息发送流程是这样的：
- 可以有多个队列。
- 每个队列都要绑定到Exchange（交换机）。
- 生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无法决定。
- 交换机把消息发送给绑定过的所有队列。
- 订阅队列的消费者都能拿到消息。
1、在消费者中创建一个配置类，声明队列和交换机：
package com.jie.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; /** * @description:广播交换机配置类 * @author: jie * @time: 2022/2/22 11:09 */ @Configuration public class FanoutConfig { /** * @description:创建交换机itcast.fanout * @author: jie * @time: 2022/2/12 19:38 */ @Bean public FanoutExchange fanoutExchange(){ return new FanoutExchange("itcast.fanout"); } /** * @description:创建队列fanout.queuel * @author: jie * @time: 2022/2/12 19:40 */ @Bean public Queue fanoutQueue1(){ return new Queue("fanout.queuel"); } /** * @description:绑定队列fanout.queuel到交换机 * @author: jie * @time: 2022/2/12 19:41 */ @Bean public Binding fanoutBinding1(Queue fanoutQueue1,FanoutExchange fanoutExchange){ return BindingBuilder .bind(fanoutQueue1) .to(fanoutExchange); } // fanout.queue2 /** * @description:创建队列fanout.queue2 * @author: jie * @time: 2022/2/12 19:40 */ @Bean public Queue fanoutQueue2(){ return new Queue("fanout.queue2"); } /** * @description:绑定队列fanout.queue2到交换机 * @author: jie * @time: 2022/2/12 19:41 */ @Bean public Binding fanoutBinding2(Queue fanoutQueue2,FanoutExchange fanoutExchange){ return BindingBuilder .bind(fanoutQueue2) .to(fanoutExchange); } }
运行消费者然后查看浏览器。
2、消息发送
在生产者服务的测试类中添加测试方法：
/** * @description:测试交换机发送消息 * @author: jie * @time: 2022/2/22 11:18 */ @Test public void testFanoutExchange() { //交换机名称 String exchangeName = "itcast.fanout"; //消息 String message = "hello,every one!"; // 发送消息 rabbitTemplate.convertAndSend(exchangeName,"",message); }
3、消息接收
在消费者服务中添加两个方法，作为消费者：
@RabbitListener(queues = "fanout.queuel") public void listenFanoutQueue1(String msg){ System.out.println("消费者接收到fanout.queuel的消息 ：【" + msg + "】"); } @RabbitListener(queues = "fanout.queue2") public void listenFanoutQueue2(String msg){ System.out.println("消费者接收到fanout.queue2的消息 ：【" + msg + "】"); }
4、测试
运行生产者消息发送，启动消费者，查看控制台。
5、小结
交换机的作用是什么？
- 接收生产者发送的消息
- 将消息按照规则路由到与之绑定的队列
- 不能缓存消息，路由失败，消息丢失
- FanoutExchange的会将消息路由到每个绑定的队列
声明队列、交换机、绑定关系的Bean是什么？
- Queue（队列）
- FanoutExchange（广播交换机）
- Binding（绑定）
4.2、Direct（路由）
在Fanout模式中，一条消息，会被所有订阅的队列都消费。但是，在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。
Direct Exchange 会将接收到的消息根据规则路由到指定的Queue，因此称为路由模式（routes）。
在Direct模型下：
- 队列与交换机的绑定，不能是任意绑定了，而是要指定一个
RoutingKey（路由key）- 消息的发送方在 向 Exchange发送消息时，也必须指定消息的
RoutingKey。- Exchange不再把消息交给每一个绑定的队列，而是根据消息的
Routing Key进行判断，只有队列的Routingkey与消息的Routing key完全一致，才会接收到消息
1、消息接收
基于@Bean的方式声明队列和交换机比较麻烦，Spring还提供了基于注解方式来声明。
在消费者r中添加两个消费者，同时基于注解来声明队列和交换机：
@RabbitListener(bindings = @QueueBinding( //队列 value = @Queue(name = "driect.queue1"), //交换机 exchange = @Exchange(name = "itcast.direct",type = ExchangeTypes.DIRECT), key = {"red","blue"} )) public void listenDirectQueue2(String msg){ System.out.println("消费者接收到direct.queue1的消息 ：【" + msg + "】"); } @RabbitListener(bindings = @QueueBinding( //队列 value = @Queue(name = "driect.queue2"), //交换机 exchange = @Exchange(name = "itcast.direct",type = ExchangeTypes.DIRECT), key = {"red","yellow"} )) public void listenDirectQueue1(String msg){ System.out.println("消费者接收到direct.queue2的消息 ：【" + msg + "】"); }
2、消息发送
@Test public void testSendDirectExchange(){ //交换机名称 String exchangeName = "itcast.direct"; //消息 String message = "hello,jie"; // 发送消息 rabbitTemplate.convertAndSend(exchangeName,"blue",message); }
3、总结
描述下Direct交换机与Fanout交换机的差异？
- Fanout交换机将消息路由给每一个与之绑定的队列
- Direct交换机根据RoutingKey判断路由给哪个队列
- 如果多个队列具有相同的RoutingKey，则与Fanout功能类似
基于@RabbitListener注解声明队列和交换机有哪些常见注解？
- @Queue
- @Exchange
4.3 Topic（通配符）
Topic类型的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key的时候使用通配符！
Routingkey一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如：item.insert通配符规则：
#：匹配一个或多个词
*：匹配不多不少恰好1个词举例：
item.#：能够匹配item.spu.insert或者item.spu
item.*：只能匹配item.spu
1、消息发送
在生产者服务的中添加测试方法：
@Test public void testSendTopicExchange(){ //交换机名称 String exchangeName = "itcast.topic"; //消息 String message = "2022冬奥运在中国开始了！"; // 发送消息 rabbitTemplate.convertAndSend(exchangeName,"china.weather",message); }
2、消息接收
在消费者服务中添加方法：
@RabbitListener(bindings = @QueueBinding( value = @Queue(name = "topic.queue1"), exchange = @Exchange(name = "itcast.topic",type = ExchangeTypes.TOPIC), key = "china.#" )) public void listenTopicQueue1(String msg){ System.out.println("消费者接收到topic.queue1的消息 ：【" + msg + "】"); } @RabbitListener(bindings = @QueueBinding( value = @Queue(name = "topic.queue2"), exchange = @Exchange(name = "itcast.topic",type = ExchangeTypes.TOPIC), key = "#.news" )) public void listenTopicQueue2(String msg){ System.out.println("消费者接收到topic.queue2的消息 ：【" + msg + "】"); }
3、小结
- Topic交换机接收的消息RoutingKey必须是多个单词，以
**.**分割- Topic交换机与队列绑定时的bindingKey可以指定通配符
#：代表0个或多个词*：代表1个词
5、消息转换器
Spring会把你发送的消息序列化为字节发送给MQ，接收消息的时候，还会把字节反序列化为Java对象。
默认情况下Spring采用的序列化方式是JDK序列化。众所周知，JDK序列化存在下列问题：
- 数据体积过大
- 有安全漏洞
- 可读性差
我们来测试一下。
5.1 测试默认转换器
@Test public void test6() { Map<String,Object> msg = new HashMap<>(); msg.put("name","zs"); msg.put("age",24); rabbitTemplate.convertAndSend("object.queue",msg); }
我们修改消息发送的代码，发送一个Map对象：
发送消息后查看浏览器：
5.2 配置JSON转换器
显然，JDK序列化方式并不合适。我们希望消息体的体积更小、可读性更高，因此可以使用JSON方式来做序列化和反序列化。
在生产者和消费者两个服务中都引入依赖：
<dependency> <groupId>com.fasterxml.jackson.core</groupId> <artifactId>jackson-databind</artifactId> </dependency>
配置消息转换器。
在启动类中添加一个Bean即可：
@Bean public MessageConverter messageConverter(){ return new Jackson2JsonMessageConverter(); }
查看浏览器
