RabbitMQ-基础
2021-09-09 23:28
253 查看
1. 简介
MQ（Message Queue）消息队列，是基础数据结构中“FIFO（先进先出）”的一种数据结构。
一般用来解决应用解耦，异步消息，流量削峰等问题，实现高性能，高可用，可伸缩和最终一致性架构。
应用解耦
MQ相当于一个中介，生产方通过MQ与消费方交互，它将应用程序进行解耦合。
异步消息
将不需要同步处理的并且耗时长的操作由消息队列通知消息接收方进行异步处理。提高了应用程序的响应时间。
流量削峰
如订单系统，在下单的时候就会往数据库写数据。但是数据库只能支撑每秒1000左右的并发写入，并发量再高就容易宕机。低峰期的时候并发也就100多个，但是在高峰期时候，并发量会突然激增到5000以上，这个时候数据库肯定卡死了。
这时候我们可以使用MQ将消息保存起来，然后系统就可以按照自己的消费能力来消费，比如每秒1000个数据，这样慢慢写入数据库，这样就不会卡死数据库了。
但是使用了MQ之后，限制消费消息的速度为1000，但是这样一来，高峰期产生的数据势必会被积压在MQ中，高峰就被“削”掉了。但是因为消息积压，在高峰期过后的一段时间内，消费消息的速度还是会维持在1000QPS，直到消费完积压的消息,这就叫做“填谷”。
2. RabbitMQ
RabbitMQ是由erlang语言开发，基于AMQP（Advanced Message Queue 高级消息队列协议）协议实现的消息队列。
RabbitMQ 其实是一个消息代理：它接受和转发消息。可以将其视为邮局：当你把 要投递的邮件放入邮箱时，你可以确定邮递员最终会将邮件递送给你的收件人。在这个比喻中，RabbitMQ 是一个邮箱、一个邮局和一个信件载体。 RabbitMQ 和邮局之间的主要区别在于它不处理纸张，而是接受、存储和转发二进制数据块 - 消息。
3. 模式
这里仅介绍了常用的模式，最近看官网又多个模式Publisher Confirms，完了有时间再补充上。
关于官网中提到的第六种模式RPC，由于RPC通信一般不使用RabbitMQ，所以这里也没有讲。
3.1 简单模式
如图所示：只有一个生产者（P）一个队列（红色块）和 一个消费者(C)。
**应用场景：**可以实现对应用程序的解耦，并且可以实现对业务的异步处理。事实上这是mq最基本的功能。
3.2 工作模式
如图所示：一个生产者对应多个消费者。多个消费者功能消费一个队列（负载均衡）。
每个消息只能被其中的一个消费者消费。
**应用场景：**对于 任务过重或任务较多情况使用工作队列可以提高任务处理的速度。
3.3 发布订阅模式
如图所示：在生产者和队列之间多了个交换机（X），此时的交换机类型为：
扇形交换机（Fanout Exchange）。
事实上，简单模式和工作模式也都有自己的Exchange，只不过不用显性的声明，因为默认使用
default Exchange。
即：一个发送到Exchange的消息都会被转发到与该交换机绑定的所有队列上。
每一个消息能被多个消费者都消费。
Fanout Exchange消息路由规则如图所示：
**应用场景：**顾名思义，一个消息想被多个订阅者消费。
3.4 路由模式
如图所示：相比发布订阅模式，Exchange和Queue之间多了个路由关系，此时的交换机类型为：
直连交换机（Direct Exchange）。
队列和交换机不是任意绑定了，而是要指定一个
Routingkey
。生产者在向Exchange发送消息时，也必须指定消息的
RoutingKey
。Exchange不再把消息交给每一个绑定的队列，而是根据消息的
Routing Key
进行判断，只有队列的Routingkey
与消息的Routing key
完全一致，才会接收到消息。
Direct Exchange消息路由规则如图所示：
3.5 通配符模式/主题模式
如图所示：相比路由模式，Exchange和Queue之间不只是通过固定的
RoutingKey进行绑定，还支持通配符的方式，此时的交换机类型为：
主题交换机/通配符交换机（Topic Exchange）。
Topic Exchange消息路由规则如图所示：
3. 安装RabbitMQ
version: '2' services: rabbitmq: hostname: rabbitmq image: rabbitmq:3.8.3-management restart: always environment: # 默认的用户名 - RABBITMQ_DEFAULT_USER=admin # 默认的密码 - RABBITMQ_DEFAULT_PASS=admin123 volumes: - ./data:/var/lib/rabbitmq - ./log:/var/log/rabbitmq/log ports: # rabbit ui 默认端口 - "15672:15672" # Epmd 是 Erlang Port Mapper Daemon 的缩写， # 在 Erlang 集群中相当于 dns 的作用，绑定在4369端口上 - "4369:4369" # rabbit 默认的端口 - "5672:5672" # 25672端口用于节点间和CLI工具通信（Erlang分发服务器端口）， # 并从动态范围分配（默认情况下仅限于单个端口， # 计算方式为AMQP 0-9-1和AMQP 1.0端口+20000）, # 默认情况下通过 RABBITMQ_NODE_PORT 计算是25672 - "25672:25672"
4. 各种模式的简单实现
4.1 项目搭建
4.1.1 引入依赖
我们这里使用
spring-boot-starter-amqp操作RabbitMQ。
<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>2.5.4</version> <relativePath/> <!-- lookup parent from repository --> </parent> <groupId>com.ldx</groupId> <artifactId>rabbitmq</artifactId> <version>0.0.1-SNAPSHOT</version> <name>rabbitmq</name> <description>Demo project for Spring Boot</description> <properties> <java.version>1.8</java.version> </properties> <dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency> <dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> <optional>true</optional> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <scope>test</scope> </dependency> <dependency> <groupId>org.springframework.amqp</groupId> <artifactId>spring-rabbit-test</artifactId> <scope>test</scope> </dependency> </dependencies> </project>
4.1.2 application.yaml
spring: rabbitmq: host: localhost port: 5672 # rabbit 默认的虚拟主机 virtual-host: / # rabbitmq 安装时指定的超管信息 username: admin password: admin123
4.2 简单模式
4.2.1 声明一个简单队列
package com.ldx.rabbitmq.config;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* rabbit 快速开始
*
* @author ludangxin
* @date 2021/8/23
*/
@Configuration
public class RabbitSimpleConfig {
/**
* 设置一个简单的队列
*/
@Bean
public Queue queue() {
return new Queue("helloMQ");
}
}
4.2.2 创建生产者
package com.ldx.rabbitmq.producer;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
/**
* 生产者
*
* @author ludangxin
* @date 2021/8/23
*/
@Component
public class SimpleProducer {
@Autowired
private RabbitTemplate rabbitTemplate;
public void send() {
String context = "helloMQ " + System.currentTimeMillis();
rabbitTemplate.convertAndSend("helloMQ", context);
}
}
4.2.3 创建消费者
package com.ldx.rabbitmq.consumer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
/**
* 消费者
*
* @author ludangxin
* @date 2021/8/23
*/
@Slf4j
@Component
@RabbitListener(queues = {"helloMQ"})
public class SimpleConsumer {
@RabbitHandler
public void process(String hello) {
log.info("Message：{} ", hello);
}
}
4.2.4 创建测试类
package com.ldx.rabbitmq;
import com.ldx.rabbitmq.producer.SimpleProducer;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
@SpringBootTest
public class RabbitMQTest {
@Autowired
private SimpleProducer simpleSender;
@Test
public void hello() throws Exception {
// 每秒发送一条消息
for (int i = 0; i < 10; i++) {
simpleSender.send();
Thread.sleep(1000);
}
}
}
4.2.5 启动测试
启动测试类，输出内容如下：
每秒消费一条消息。
2021-09-08 23:58:01.837 INFO 29956 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.SimpleConsumer : Message：helloMQ 1631116681827 2021-09-08 23:58:02.839 INFO 29956 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.SimpleConsumer : Message：helloMQ 1631116682833 2021-09-08 23:58:03.842 INFO 29956 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.SimpleConsumer : Message：helloMQ 1631116683838 2021-09-08 23:58:04.852 INFO 29956 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.SimpleConsumer : Message：helloMQ 1631116684843 2021-09-08 23:58:05.853 INFO 29956 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.SimpleConsumer : Message：helloMQ 1631116685844 2021-09-08 23:58:06.853 INFO 29956 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.SimpleConsumer : Message：helloMQ 1631116686847 2021-09-08 23:58:07.857 INFO 29956 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.SimpleConsumer : Message：helloMQ 1631116687850 2021-09-08 23:58:08.863 INFO 29956 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.SimpleConsumer : Message：helloMQ 1631116688855 2021-09-08 23:58:09.868 INFO 29956 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.SimpleConsumer : Message：helloMQ 1631116689858 2021-09-08 23:58:10.870 INFO 29956 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.SimpleConsumer : Message：helloMQ 1631116690862
4.2.6 小节
简单模式，顾名思义，很简单，相当于Hello World程序。我们在编写的时候
- 指定了一个
Queue
并且名称为helloMQ
。 - 消息生产者通过SpringBoot 提供的
RabbitTemplate
发送消息，我们在发送时指定了Queue
为helloMQ
且发送了指定内容。 - 消息消费者通过
@RabbitListener
注解监听了指定Queue
为helloMQ
，且使用@RabbitHandler
注解指定消费方法SimpleConsumer::process()
。 - 最后编写测试类循环调用生产者消息发送逻辑，实现了消息的生产与消费。
4.3 工作模式
首先分析：其实工作模式和简单模式相比，仅仅是又一个消费者变成了多个消费者。ok，很好办，我们通过代码再多加一个消费者即可。
4.3.1 添加消费者
package com.ldx.rabbitmq.consumer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
/**
* 消费者
*
* @author ludangxin
* @date 2021/8/23
*/
@Slf4j
@Component
@RabbitListener(queues = {"helloMQ"})
public class SimpleConsumer2 {
@RabbitHandler
public void process(String hello) {
log.info("Message2：{} ", hello);
}
}
4.3.2 启动测试
我们再次执行test方法，查看消息消费情况。
输出日志如下：
SimpleConsumer和
SimpleConsumer2交替消费队列中的消息（消费者之间消费消息是通过轮询的关系）。
2021-09-09 20:24:35.043 INFO 41927 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.SimpleConsumer : Message：helloMQ 1631190275019 2021-09-09 20:24:36.038 INFO 41927 --- [ntContainer#1-1] c.ldx.rabbitmq.consumer.SimpleConsumer2 : Message2：helloMQ 1631190276029 2021-09-09 20:24:37.036 INFO 41927 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.SimpleConsumer : Message：helloMQ 1631190277032 2021-09-09 20:24:38.046 INFO 41927 --- [ntContainer#1-1] c.ldx.rabbitmq.consumer.SimpleConsumer2 : Message2：helloMQ 1631190278036 2021-09-09 20:24:39.049 INFO 41927 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.SimpleConsumer : Message：helloMQ 1631190279041 2021-09-09 20:24:40.049 INFO 41927 --- [ntContainer#1-1] c.ldx.rabbitmq.consumer.SimpleConsumer2 : Message2：helloMQ 1631190280042 2021-09-09 20:24:41.054 INFO 41927 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.SimpleConsumer : Message：helloMQ 1631190281047 2021-09-09 20:24:42.060 INFO 41927 --- [ntContainer#1-1] c.ldx.rabbitmq.consumer.SimpleConsumer2 : Message2：helloMQ 1631190282051 2021-09-09 20:24:43.062 INFO 41927 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.SimpleConsumer : Message：helloMQ 1631190283055 2021-09-09 20:24:44.074 INFO 41927 --- [ntContainer#1-1] c.ldx.rabbitmq.consumer.SimpleConsumer2 : Message2：helloMQ 1631190284057
4.3.3 小节
在一个队列中如果有多个消费者，那么消费者之间是轮询的关系。
4.4 发布订阅模式
首先分析：发布订阅模式其实是将消息先发送给
扇形交换机，交换机再将消息转发给其绑定到此交换机的队列上。
这里，我们声明一个交换机，给交换机绑定两个队列，并且使用两个消费者分别绑定到两个队列上（其实就是为了和
3.3保持一致）。
4.4.1 声明交换机和队列
package com.ldx.rabbitmq.config;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* 扇形交换机配置
*
* @author ludangxin
* @date 2021/9/9
*/
@Configuration
public class RabbitFanoutConfig {
public static final String EXCHANGE_NAME = "FANOUT_EXCHANGE";
public static final String QUEUE_NAME = "FANOUT_QUEUE";
public static final String QUEUE_NAME_1 = "FANOUT_QUEUE_1";
/**
* 1.交换机
*/
@Bean(EXCHANGE_NAME)
public Exchange fanoutExchange() {
return ExchangeBuilder.fanoutExchange(EXCHANGE_NAME).durable(true).build();
}
/**
* 2.Queue 队列
*/
@Bean(QUEUE_NAME)
public Queue fanoutQueue() {
return QueueBuilder.durable(QUEUE_NAME).build();
}
/**
* 2.1 Queue 队列
*/
@Bean(QUEUE_NAME_1)
public Queue fanoutQueue1() {
return QueueBuilder.durable(QUEUE_NAME_1).build();
}
/**
* 3. 队列和交互机绑定关系 Binding
*/
@Bean
public Binding bindFanoutExchange(@Qualifier(QUEUE_NAME) Queue queue, @Qualifier(EXCHANGE_NAME) Exchange exchange) {
// fanout :routing key 默认为 "",指定了别的值也没用
return BindingBuilder.bind(queue).to(exchange).with("").noargs();
}
/**
* 3.1 队列和交互机绑定关系 Binding
*/
@Bean
public Binding bindFanoutExchange1(@Qualifier(QUEUE_NAME_1) Queue queue, @Qualifier(EXCHANGE_NAME) Exchange exchange) {
// fanout :routing key 默认为 "",指定了别的值也没用，我们这里随便写个值，看会不会有影响
return BindingBuilder.bind(queue).to(exchange).with("aaabbb").noargs();
}
}
4.4.2 创建生产者
package com.ldx.rabbitmq.producer;
import com.ldx.rabbitmq.config.RabbitFanoutConfig;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
/**
* 扇形交换机消息生产者
*
* @author ludangxin
* @date 2021/9/9
*/
@Component
public class FanoutProducer {
@Autowired
private RabbitTemplate rabbitTemplate;
public void sendWithFanout() {
rabbitTemplate.convertAndSend(RabbitFanoutConfig.EXCHANGE_NAME, "", "fanout mq hello~~~");
// 指定一个routingKey 看消费方能不能正常接收消息
rabbitTemplate.convertAndSend(RabbitFanoutConfig.EXCHANGE_NAME, "abc", "fanout2 mq hello~~~");
}
}
4.4.3 创建消费者
package com.ldx.rabbitmq.consumer;
import com.ldx.rabbitmq.config.RabbitFanoutConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
/**
* 扇形交换机消息消费者
*
* @author ludangxin
* @date 2021/9/9
*/
@Slf4j
@Component
public class FanoutConsumer {
@RabbitListener(queues = {RabbitFanoutConfig.QUEUE_NAME})
public void process(String message){
log.info("queue === " + message);
}
@RabbitListener(queues = {RabbitFanoutConfig.QUEUE_NAME_1})
public void process1(String message){
log.info("queue1 === " + message);
}
}
4.4.4 创建测试代码
@Autowired
private FanoutProducer producer;
@Test
@SneakyThrows
public void sendWithFanout(){
producer.sendWithFanout();
// 为了阻塞进程，使消费者能正常消费。
System.in.read();
}
4.4.5 启动测试
执行测试方法，输出内容如下：
生产者发送的两条消息，被两个消费者共同消费了。实现了消息的广播。
2021-09-09 21:59:17.538 INFO 43749 --- [ntContainer#1-1] c.ldx.rabbitmq.consumer.FanoutConsumer : queue1 === fanout mq hello~~~ 2021-09-09 21:59:17.538 INFO 43749 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.FanoutConsumer : queue === fanout mq hello~~~ 2021-09-09 21:59:17.539 INFO 43749 --- [ntContainer#1-1] c.ldx.rabbitmq.consumer.FanoutConsumer : queue1 === fanout2 mq hello~~~ 2021-09-09 21:59:17.539 INFO 43749 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.FanoutConsumer : queue === fanout2 mq hello~~~
4.4.6 小节
本节代码中我们创建了一个
fanout Exchange，并且创建了两个队列与其绑定，其中一个队列进行绑定的时候还指定了
routing key，但程序执行时消息正常被消费，说明
fanout Exchange不用指定
routing key。
发布订阅模式与工作队列模式的区别
1、工作队列模式不用定义交换机，而发布/订阅模式需要定义交换机。
2、发布/订阅模式的生产方是面向交换机发送消息，工作队列模式的生产方是面向队列发送消息(底层使用默认交换机)。
3、发布/订阅模式需要设置队列和交换机的绑定，工作队列模式不需要设置，实际上工作队列模式会将队列绑定到默认的交换机 。
4.5 路由模式
首先分析：路由模式其实就是将 发布订阅模式中的
fanout Exchange换成了
direct Exchange从而指定相应的路由规则即可。
4.5.1 声明交换机和队列
package com.ldx.rabbitmq.config;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* 直连交换机配置
*
* @author ludangxin
* @date 2021/9/9
*/
@Configuration
public class RabbitDirectConfig {
public static final String EXCHANGE_NAME = "DIRECT_EXCHANGE";
public static final String QUEUE_NAME_INSERT = "DIRECT_QUEUE_INSERT";
public static final String QUEUE_NAME_UPDATE = "DIRECT_QUEUE_UPDATE";
/**
* 1.交换机
*/
@Bean(EXCHANGE_NAME)
public Exchange bootExchange() {
return ExchangeBuilder.directExchange(EXCHANGE_NAME).durable(true).build();
}
/**
* 2.Queue insert队列
*/
@Bean(QUEUE_NAME_INSERT)
public Queue bootQueueInsert() {
return QueueBuilder.durable(QUEUE_NAME_INSERT).build();
}
/**
* 2.Queue update队列
*/
@Bean(QUEUE_NAME_UPDATE)
public Queue bootQueueUpdate() {
return QueueBuilder.durable(QUEUE_NAME_UPDATE).build();
}
/**
* 3. 绑定insert 队列
* 3. routing key: insert
*/
@Bean
public Binding bindInsertDirectExchange(@Qualifier(QUEUE_NAME_INSERT) Queue queue, @Qualifier(EXCHANGE_NAME) Exchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with("insert").noargs();
}
/**
* 3. 绑定update 队列
* 3. routing key: update
*/
@Bean
public Binding bindUpdateDirectExchange(@Qualifier(QUEUE_NAME_UPDATE) Queue queue, @Qualifier(EXCHANGE_NAME) Exchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with("update").noargs();
}
}
4.5.2 创建生产者
package com.ldx.rabbitmq.producer;
import com.ldx.rabbitmq.config.RabbitDirectConfig;
import com.ldx.rabbitmq.config.RabbitFanoutConfig;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
/**
* 直连交换机消息生产者
*
* @author ludangxin
* @date 2021/9/9
*/
@Component
public class DirectProducer {
@Autowired
private RabbitTemplate rabbitTemplate;
public void sendWithDirect() {
rabbitTemplate.convertAndSend(RabbitDirectConfig.EXCHANGE_NAME, "insert", "diect insert mq hello~~~");
rabbitTemplate.convertAndSend(RabbitDirectConfig.EXCHANGE_NAME, "update", "diect update mq hello~~~");
// 指定一个没有配置的routingKey 看消费方能不能接收消息
rabbitTemplate.convertAndSend(RabbitDirectConfig.EXCHANGE_NAME, "delete", "diect update mq hello~~~");
}
}
4.5.3 创建消息者
package com.ldx.rabbitmq.consumer;
import com.ldx.rabbitmq.config.RabbitDirectConfig;
import com.ldx.rabbitmq.config.RabbitFanoutConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
/**
* 直连交换机消息消费者
*
* @author ludangxin
* @date 2021/9/9
*/
@Slf4j
@Component
public class DirectConsumer {
/**
* @param message message 为springboot 封装的消息存储的实例对象，其对象中不仅封装了生产者发送的消息
* 而且也封装了很多消息的元数据，例如：headers contentType receivedRoutingKey ...
*/
@RabbitListener(queues = {RabbitDirectConfig.QUEUE_NAME_INSERT, RabbitDirectConfig.QUEUE_NAME_UPDATE})
public void directQueue(Message message){
log.info(message.toString());
log.info(new String(message.getBody()));
}
}
4.5.4 创建测试代码
@Autowired
private DirectProducer directProducer;
@Test
@SneakyThrows
public void sendWithDirect() {
directProducer.sendWithDirect();
System.in.read();
}
4.5.5 启动测试
执行测试代码，输出内容如下：
insert 和 update 对应的消息都被正常消费，其中值得注意的是指定
routing key=delete的消息丢失了，因为队列与交换机绑定时根本没有此
routing key，而交换机之所以叫交换机，因为其不存储消息，只是转发消息，其没有持久化消息的能力，所以消息还没有到
queue，然后嗝屁。
2021-09-09 22:38:49.451 INFO 44436 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.DirectConsumer : (Body:'diect insert mq hello~~~' MessageProperties [headers={}, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=DIRECT_EXCHANGE, receivedRoutingKey=insert, deliveryTag=1, consumerTag=amq.ctag-WJmYhQDljkKkM1pFeW99Yg, consumerQueue=DIRECT_QUEUE_INSERT])
2021-09-09 22:38:49.451 INFO 44436 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.DirectConsumer : diect insert mq hello~~~
2021-09-09 22:38:49.452 INFO 44436 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.DirectConsumer : (Body:'diect update mq hello~~~' MessageProperties [headers={}, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=DIRECT_EXCHANGE, receivedRoutingKey=update, deliveryTag=2, consumerTag=amq.ctag-guzpfaF0BdII70w85ywiCg, consumerQueue=DIRECT_QUEUE_UPDATE])
2021-09-09 22:38:49.452 INFO 44436 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.DirectConsumer : diect update mq hello~~~
4.5.6 小节
路由模式特点：
- 队列与交换机的绑定，不能是任意绑定了，而是要指定一个
rutingKey
（路由key）。 - 消息的发送方在 向 Exchange发送消息时，也必须指定消息的
routingKey
。 - Exchange不再把消息交给每一个绑定的队列，而是根据消息的
routing Key
进行判断，只有队列的routingkey
与消息的routing key
完全一致，才会接收到消息。
4.6 主题模式
首先分析：通配符模式其实就是将 路由模式中的
direct Exchange换成了
topic Exchange， 使其不仅可以将
exchange和
queue以
routing key全匹配的方式进行绑定，而且还支持
通配符。
routingkey一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如：
item.insert
通配符规则：
#：匹配一个或多个单词
*：匹配一个单词
举例：
item.#：能够匹配
item.insert.abc或者
item.insert
item.*：只能匹配
item.insert
图解：
- 红色Queue：绑定的是
usa.#
，因此凡是以usa.
开头的routing key
都会被匹配到 - 黄色Queue：绑定的是
#.news
，因此凡是以.news
结尾的routing key
都会被匹配
4.6.1 声明交换机和队列
package com.ldx.rabbitmq.config;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* 主题交换机配置
*
* @author ludangxin
* @date 2021/9/9
*/
@Configuration
public class RabbitTopicConfig {
public static final String EXCHANGE_NAME = "TOPIC_EXCHANGE";
public static final String QUEUE_NAME1 = "TOPIC_QUEUE1";
public static final String QUEUE_NAME2 = "TOPIC_QUEUE2";
/**
* 1.交换机
* topicExchange:通配符，把消息交给符合routing pattern（路由模式） 的队列
*/
@Bean(EXCHANGE_NAME)
public Exchange bootExchange() {
return ExchangeBuilder.topicExchange(EXCHANGE_NAME).durable(true).build();
}
/**
* 2.Queue 队列
*/
@Bean(QUEUE_NAME1)
public Queue bootQueue1() {
return QueueBuilder.durable(QUEUE_NAME1).build();
}
/**
* 2.Queue 队列
*/
@Bean(QUEUE_NAME2)
public Queue bootQueue2() {
return QueueBuilder.durable(QUEUE_NAME2).build();
}
/**
* 3. 队列和交互机绑定关系 Binding
* 匹配 routing key 以 insert 开头的 如 insert.user ; insert.user.log
*/
@Bean
public Binding bindTopicExchange1(@Qualifier(QUEUE_NAME1) Queue queue, @Qualifier(EXCHANGE_NAME) Exchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with("insert.#").noargs();
}
/**
* 3. 队列和交互机绑定关系 Binding
* routing key 中的 * 只能匹配单个单词
* 匹配 routing key 以 update 开头的 如 update.user
* 不能匹配 如 update.user.log
*/
@Bean
public Binding bindTopicExchange2(@Qualifier(QUEUE_NAME1) Queue queue, @Qualifier(EXCHANGE_NAME) Exchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with("update.*").noargs();
}
/**
* 3. 队列和交互机绑定关系 Binding
* routing key 中的 * 只能匹配单个单词
* 匹配 routing key 以 . 分割的
* 不能匹配 如 update.user.log
*/
@Bean
public Binding bindTopicExchange3(@Qualifier(QUEUE_NAME2) Queue queue, @Qualifier(EXCHANGE_NAME) Exchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with("*.*").noargs();
}
}
4.6.2 创建生产者
package com.ldx.rabbitmq.producer;
import com.ldx.rabbitmq.config.RabbitTopicConfig;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
/**
* 主题交换机消息生产者
*
* @author ludangxin
* @date 2021/9/9
*/
@Component
public class TopicProducer {
@Autowired
private RabbitTemplate rabbitTemplate;
public void sendWithTopic() {
rabbitTemplate.convertAndSend(RabbitTopicConfig.EXCHANGE_NAME, "insert.user.log", "topic mq hello~~~ routing is insert.user.lo");
rabbitTemplate.convertAndSend(RabbitTopicConfig.EXCHANGE_NAME, "update.user", "topic mq hello~~~ routing is update.user");
rabbitTemplate.convertAndSend(RabbitTopicConfig.EXCHANGE_NAME, "update.user.log", "topic mq hello~~~ routing is update.user.log");
rabbitTemplate.convertAndSend(RabbitTopicConfig.EXCHANGE_NAME, "delete.user", "topic mq hello~~~ routing is delete.user");
rabbitTemplate.convertAndSend(RabbitTopicConfig.EXCHANGE_NAME, "delete.user.log", "topic mq hello~~~routing is delete.user.log");
}
}
4.6.3 创建消费者
package com.ldx.rabbitmq.consumer;
import com.ldx.rabbitmq.config.RabbitTopicConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
/**
* 主题交换机消息消费者
*
* @author ludangxin
* @date 2021/9/9
*/
@Slf4j
@Component
public class TopicConsumer {
@RabbitListener(queues = {RabbitTopicConfig.QUEUE_NAME1, RabbitTopicConfig.QUEUE_NAME2})
public void topicQueue(Message message){
log.info(message.toString());
log.info(new String(message.getBody()));
}
}
4.6.4 创建测试代码
@Autowired
private TopicProducer topicProducer;
@Test
@SneakyThrows
public void sendWithTopic() {
topicProducer.sendWithTopic();
System.in.read();
}
4.6.5 启动测试
执行测试代码，输入内容如下：
其中符合通配符条件的消息均已消费。
2021-09-09 23:02:57.131 INFO 44812 --- [ntContainer#5-1] com.ldx.rabbitmq.consumer.TopicConsumer : (Body:'topic mq hello~~~ routing is insert.user.lo' MessageProperties [headers={}, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=TOPIC_EXCHANGE, receivedRoutingKey=insert.user.log, deliveryTag=1, consumerTag=amq.ctag-PeBOPjJFHMF3BMW1zDXCvw, consumerQueue=TOPIC_QUEUE1])
2021-09-09 23:02:57.132 INFO 44812 --- [ntContainer#5-1] com.ldx.rabbitmq.consumer.TopicConsumer : topic mq hello~~~ routing is insert.user.lo
2021-09-09 23:02:57.132 INFO 44812 --- [ntContainer#5-1] com.ldx.rabbitmq.consumer.TopicConsumer : (Body:'topic mq hello~~~ routing is update.user' MessageProperties [headers={}, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=TOPIC_EXCHANGE, receivedRoutingKey=update.user, deliveryTag=2, consumerTag=amq.ctag-PeBOPjJFHMF3BMW1zDXCvw, consumerQueue=TOPIC_QUEUE1])
2021-09-09 23:02:57.132 INFO 44812 --- [ntContainer#5-1] com.ldx.rabbitmq.consumer.TopicConsumer : topic mq hello~~~ routing is update.user
2021-09-09 23:02:57.133 INFO 44812 --- [ntContainer#5-1] com.ldx.rabbitmq.consumer.TopicConsumer : (Body:'topic mq hello~~~ routing is update.user' MessageProperties [headers={}, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=TOPIC_EXCHANGE, receivedRoutingKey=update.user, deliveryTag=3, consumerTag=amq.ctag-ocNDmCGDF8-aPxJ4lK1c8g, consumerQueue=TOPIC_QUEUE2])
2021-09-09 23:02:57.133 INFO 44812 --- [ntContainer#5-1] com.ldx.rabbitmq.consumer.TopicConsumer : topic mq hello~~~ routing is update.user
2021-09-09 23:02:57.133 INFO 44812 --- [ntContainer#5-1] com.ldx.rabbitmq.consumer.TopicConsumer : (Body:'topic mq hello~~~ routing is delete.user' MessageProperties [headers={}, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=TOPIC_EXCHANGE, receivedRoutingKey=delete.user, deliveryTag=4, consumerTag=amq.ctag-ocNDmCGDF8-aPxJ4lK1c8g, consumerQueue=TOPIC_QUEUE2])
2021-09-09 23:02:57.133 INFO 44812 --- [ntContainer#5-1] com.ldx.rabbitmq.consumer.TopicConsumer : topic mq hello~~~ routing is delete.user
4.6.6 小节
Topic主题模式可以实现
Publish/Subscribe发布与订阅模式和
Routing路由模式的功能；只是Topic在配置routing key 的时候可以使用通配符，显得更加灵活。
5. 模式总结
RabbitMQ工作模式： 1、简单模式 HelloWorld 一个生产者、一个消费者，不需要设置交换机（使用
default Exchange）。
2、工作队列模式 Work Queue 一个生产者、多个消费者（平均分配消息），不需要设置交换机（使用
default Exchange）。
3、发布订阅模式 Publish/subscribe 需要设置交换机类型为
fanout Exchange，并且交换机和队列进行绑定，当发送消息到交换机后，交换机会将消息发送到绑定的队列。
4、路由模式 Routing 需要设置交换机类型为
direct Exchange，交换机和队列进行绑定，并且指定
routing key，发送消息时也要指定对应的
routing key到交换机，交换机会根据
routing key将消息发送到对应的队列。
5、主题模式 Topic 需要设置交换机类型为
topic Exchange，，交换机和队列进行绑定，并且指定通配符方式的
routing key，发送消息时指定
routing key到交换机后，交换机会根据
routing key规则将消息发送到对应的队列。主题模式比上面四类更灵活。
内容来自用户分享和网络整理，不保证内容的准确性，如有侵权内容，可联系管理员处理 
相关文章推荐
- RabbitMQ基础概念详细介绍
- RabbitMQ基础
- 【Flask基础】1、flask结合RabbitMq实现轻量级的数据接收队列，并使用twisted打印日志
- rabbitmq 基础一
- RabbitMQ 基础教程 Publish/Subscribe（发布/订阅）
- RabbitMQ消息队列入门篇（环境配置+Java实例+基础概念）
- RabbitMQ 基础教程 Routing（消息路由）
- RabbitMQ基础概念详细介绍
- RabbitMQ基础总结
- .NET 云原生架构师训练营（模块二 基础巩固 RabbitMQ 工作队列和交换机）--学习笔记
- RabbitMQ基础知识
- RabbitMQ基础概念详细介绍
- RabbitMQ实战基础
- RabbitMQ基础概念
- 【转】RabbitMQ基础——和——持久化机制
- 消息队列基础 RabbitMQ与AMQP协议详解——超大规模高可用OpenStack核心技术深入解析系列（二）
- RabbitMQ基础概念介绍
- RabbitMQ-从基础到实战（2）— 防止消息丢失
- RabbitMQ基础概念详细介绍
- ##2.基础服务(SQl,RabbitMQ)-- openstack pike