RabbitMQ消息队列(六)：SpringBoot整合之通配符模式

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-AeZQrNHS-1660220618697)(E:\typora\image\image-20220811183652501.png)]

路由模式是在订阅模式的基础上加上了路由键key，交换机只分发消息到绑定在交换机上面指定路由键的队列。

1. 环境准备

参考：https://blog.csdn.net/fatestranger/article/details/126284433?spm=1001.2014.3001.5501

交换机&队列：

public class TopicConstants {public static final String TOPIC_EXCHANGE_NAME = "amq.topic";public static final String TV_TOPIC_QUEUE_NAME = "topic.TV";public static final String PHONE_TOPIC_QUEUE_NAME = "topic.TV.phone";public static final String PAPER_TOPIC_QUEUE_NAME = "topic.TV.phone.paper";}

2. 生产者

配置交换机和绑定的队列：

RabbitMQRoutingConfig

@Configuration
public class RabbitMQRoutingConfig {//1、声明注册direct式交换机@Beanpublic DirectExchange directExchange() {return new DirectExchange(DIRECT_EXCHANGE_NAME, true, false);}/*2. 声明队列*/@Beanpublic Queue TVFanoutQueue() {/*Queue构造函数参数说明1. 队列名2. 是否持久化 true：持久化 false：不持久化*/return new Queue(TV_FANOUT_QUEUE_NAME, true);}@Beanpublic Queue phoneFanoutQueue() {/*Queue构造函数参数说明1. 队列名2. 是否持久化 true：持久化 false：不持久化*/return new Queue(PHONE_FANOUT_QUEUE_NAME, true);}//@Beanpublic Queue paperFanoutQueue() {/*Queue构造函数参数说明1. 队列名2. 是否持久化 true：持久化 false：不持久化*/return new Queue(PAPER_FANOUT_QUEUE_NAME, true);}//3、完成交换机和队列的绑定关系@Beanpublic Binding TVDirectBinding() {return BindingBuilder.bind(TVFanoutQueue()).to(directExchange()).with("TV");}@Beanpublic Binding phoneDirectBinding() {return BindingBuilder.bind(phoneFanoutQueue()).to(directExchange()).with("phone");}@Beanpublic Binding paperDirectBinding() {return BindingBuilder.bind(paperFanoutQueue()).to(directExchange()).with("paper");}}

模拟产生消息：

DirectController

@RestController
@Slf4j
public class DirectController {@Autowiredprivate RabbitTemplate rabbitTemplate;@GetMapping("/directQueue")public String sendMsg(@RequestParam String message, @RequestParam String routeKey) {log.info("message：{}, routeKey: {}", message, routeKey);rabbitTemplate.convertAndSend(DIRECT_EXCHANGE_NAME, routeKey, message);//返回消息return "OK!";}
}

3.消费者

三个消费者：

RoutingReceiver

@Component
@Slf4j
public class RoutingReceiver {@Autowiredprivate MqReceiveService mqReceiveService;@RabbitListener(queues = "TV-fanout-queue")public void processTV(String msg) {log.info("队列：TV-fanout-queue，接收msg: {} ", msg);//消息消费存储MqReceive mqReceive = new MqReceive();mqReceive.setType("direct");mqReceive.setReceive("TV-fanout-queue");mqReceive.setContent(msg);mqReceiveService.insetMsg(mqReceive);}@RabbitListener(queues = "phone-fanout-queue")public void processPhone(String msg) {log.info("队列：phone-fanout-queue，接收msg: {} ", msg);//消息消费存储MqReceive mqReceive = new MqReceive();mqReceive.setType("direct");mqReceive.setReceive("phone-fanout-queue");mqReceive.setContent(msg);mqReceiveService.insetMsg(mqReceive);}@RabbitListener(queues = "paper-fanout-queue")public void processPaper(String msg) {log.info("队列：paper-fanout-queue，接收msg: {} ", msg);//消息消费存储MqReceive mqReceive = new MqReceive();mqReceive.setType("direct");mqReceive.setReceive("paper-fanout-queue");mqReceive.setContent(msg);mqReceiveService.insetMsg(mqReceive);}}

测试：
http://localhost:8888/topicQueue?message=topic&routeKey=topic.1

http://localhost:8888/topicQueue?message=topic&routeKey=topic.TV.1

http://localhost:8888/topicQueue?message=topic&routeKey=topic.1.1

4. 模式总结

简单模式 HelloWorld 一个生产者、一个消费者，不需要设置交换机（使用默认的交换机）；
工作队列模式 Work Queue 一个生产者、多个消费者（竞争关系），不需要设置交换机（使用默认的交换机）；
发布订阅模式 Publish/subscribe 需要设置类型为 fanout 的交换机，并且交换机和队列进行绑定，当发送消息到交换机后，交换机会将消息发送到绑定的队列；
路由模式 Routing 需要设置类型为 direct 的交换机，交换机和队列进行绑定，并且指定 routing key，当发送消息到交换机后，交换机会根据 routingkey 将消息发送到对应的队列；
通配符模式 Topic 需要设置类型为 topic 的交换机，交换机和队列进行绑定，并且指定通配符方式的 routing key，当发送消息到交换机后，交换机会根据 routing key 将消息发送到对应的队列。