使用java代码操作rabbitMQ收发消息
SpringAMQP
将来我们开发业务功能的时候，肯定不会在控制台收发消息，而是应该基于编程的方式。由于RabbitMQ采用了AMQP协议，因此它具备跨语言的特性。任何语言只要遵循AMQP协议收发消息，都可以与RabbitMQ交互。并且RabbitMQ官方也提供了各种不同语言的客户端。
但是，RabbitMQ官方提供的Java客户端编码相对复杂，一般生产环境下我们更多会结合Spring来使用。而Spring的官方刚好基于RabbitMQ提供了这样一套消息收发的模板工具：SpringAMQP。并且还基于SpringBoot对其实现了自动装配，使用起来非常方便。
SpringAmqp的官方地址：
Spring AMQP
SpringAMQP提供了三个功能：
- 自动声明队列、交换机及其绑定关系
- 基于注解的监听器模式，异步接收消息
- 封装了RabbitTemplate工具，用于发送消息
快速入门
别忘了在我们的项目中，引入spring amqp的依赖。
<dependencies>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</dependency>
<!--AMQP依赖，包含RabbitMQ-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<!--单元测试-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
</dependency>
</dependencies>在之前的案例中，我们都是经过交换机发送消息到队列，不过有时候为了测试方便，我们也可以直接向队列发送消息，跳过交换机。
在入门案例中，我们就演示这样的简单模型，如图：
也就是
- publisher直接发送消息到队列
- 消费者监听并处理队列中的消息
注意：这种模式一般测试使用，很少在生产中使用。
为了方便测试，我们在rabbitMQ控制台，创建名为 simple.queue 的队列。
添加队列后查看
接下来，我们就可以利用Java代码收发消息了。
消息发送
在我们的项目application.yml 中 添加关于rabbitmq的配置信息。
spring:
rabbitmq:
host: 123.56.247.70 # 你的虚拟机IP
port: 5672 # rabbitMQ端口
virtual-host: /sde # 虚拟机名称
username: sundaoen # 用户名
password: 8888888888 # 密码
编写测试类
在我们项目的publisher中创建测试类，并且利用 RabbitTemplate 发送消息。
@SpringBootTest
public class TestSendMessage {
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
public void testSimpleQueue(){
// 1 队列名称
String queueName = "simple.queue";
// 2 消息
String message = "hello simple.queue";
// 3 发送消息
rabbitTemplate.convertAndSend(queueName,message);
}
}打开控制台，可以看到消息已经发送到队列中：
看看消息内容
接下来，我们再来实现消息接收。
消息接收
同样的道理，也是先配置MQ地址。在application.yml 中
spring:
rabbitmq:
host: 123.56.247.70 # 你的虚拟机IP
port: 5672 # rabbitMQ端口
virtual-host: /sde # 虚拟机名称
username: sundaoen # 用户名
password: 8888888888 # 密码在consumer服务中编写监听器类，并利用@RabbitListener实现消息的接收。
@Slf4j
@Component
public class SimpleQueueListener {
/*
利用@RabbitListener注解，可以监听到对应队列的消息
一旦监听的队列有消息，就会回调当前方法，在方法中接收消息并消费处理消息
*/
@RabbitListener(queues = "simple.queue")
public void listenerSimpleQueue(String msg){
System.out.println("SpringRabbitListener 监听到 simple.queue 队列中的消息是：" + msg);
}
}
WorkQueue模型
Work queues，任务模型。简单来说就是让多个消费者绑定到一个队列，共同消费队列中的消息。
当消息处理比较耗时的时候，可能生产消息的速度会远远大于消息的消费速度。长此以往，消息就会堆积越来越多，无法及时处理。
此时就可以使用work 模型，多个消费者共同处理消息处理，消息处理的速度就能大大提高了。
接下来，我们就来模拟这样的场景。
首先，我们在控制台创建一个新的队列，命名为work.queue：
添加后的效果
消息发送
这次我们循环发送，模拟大量消息堆积现象。
在publisher服务中的WorkQueueSendTest类中添加一个测试方法：
@SpringBootTest
public class WorkQueueSendTest {
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
public void testSendWorkQueue() throws InterruptedException {
// 1 队列名称
String queueName = "work.queue";
// 2 消息
String message = "hello work.queue-";
// 3 发送消息
for (int i = 1; i <= 50; i++) {
// 每隔20毫秒发送一条消息，相当于一秒发送50条消息。
rabbitTemplate.convertAndSend(queueName,message + i);
Thread.sleep(20);
}
}
}可以看到在work.queue 队列中有50条消息。
消息接收
要模拟多个消费者绑定同一个队列，我们在consumer服务中的，listener包中。新增WorkQueueListener类并添加2个新的方法：
@Slf4j
@Component
public class WorkQueueListener {
/*
实现两个消费 work.queue的监听消费消息的方法；
一个方法消费后沉睡 20毫秒；一个消息消费后沉睡200毫秒；
*/
@RabbitListener(queues = "work.queue")
public void listenerWorkQueue1(String msg){
System.out.println("消费者1接收到消息" + msg +" 时间："+ LocalDateTime.now());
try {
Thread.sleep(20); // 沉睡20毫秒 1秒是1000毫秒等于1秒处理50条消息
} catch (InterruptedException e) {
e.printStackTrace();
}
}
@RabbitListener(queues = "work.queue")
public void listenerWorkQueue2(String msg){
System.out.println("***消费者2接收到消息" + msg +" 时间："+ LocalDateTime.now());
try {
Thread.sleep(200); // 沉睡200毫秒 1秒是1000毫秒等于1秒处理5条消息
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}注意到这两消费者，都设置了Thead.sleep，模拟任务耗时：
- 消费者1 sleep了20毫秒，相当于每秒钟处理50个消息
- 消费者2 sleep了200毫秒，相当于每秒处理5个消息
测试看结果
消费者1接收到消息hello work.queue-1 时间：2025-02-05T14:53:00.928905400
***消费者2接收到消息hello work.queue-2 时间：2025-02-05T14:53:00.947629900
消费者1接收到消息hello work.queue-3 时间：2025-02-05T14:53:00.977764800
消费者1接收到消息hello work.queue-5 时间：2025-02-05T14:53:01.039608
消费者1接收到消息hello work.queue-7 时间：2025-02-05T14:53:01.101242200
消费者1接收到消息hello work.queue-9 时间：2025-02-05T14:53:01.160396600
***消费者2接收到消息hello work.queue-4 时间：2025-02-05T14:53:01.161396900
消费者1接收到消息hello work.queue-11 时间：2025-02-05T14:53:01.231704200
消费者1接收到消息hello work.queue-13 时间：2025-02-05T14:53:01.281879300
消费者1接收到消息hello work.queue-15 时间：2025-02-05T14:53:01.347333400
***消费者2接收到消息hello work.queue-6 时间：2025-02-05T14:53:01.376528100
消费者1接收到消息hello work.queue-17 时间：2025-02-05T14:53:01.407569700
消费者1接收到消息hello work.queue-19 时间：2025-02-05T14:53:01.464497900
消费者1接收到消息hello work.queue-21 时间：2025-02-05T14:53:01.525121200
消费者1接收到消息hello work.queue-23 时间：2025-02-05T14:53:01.587589500
***消费者2接收到消息hello work.queue-8 时间：2025-02-05T14:53:01.589591300
消费者1接收到消息hello work.queue-25 时间：2025-02-05T14:53:01.647549500
消费者1接收到消息hello work.queue-27 时间：2025-02-05T14:53:01.709757900
消费者1接收到消息hello work.queue-29 时间：2025-02-05T14:53:01.768879300
***消费者2接收到消息hello work.queue-10 时间：2025-02-05T14:53:01.801437800
消费者1接收到消息hello work.queue-31 时间：2025-02-05T14:53:01.829539900
消费者1接收到消息hello work.queue-33 时间：2025-02-05T14:53:01.895907400
消费者1接收到消息hello work.queue-35 时间：2025-02-05T14:53:01.950810
消费者1接收到消息hello work.queue-37 时间：2025-02-05T14:53:02.011575
***消费者2接收到消息hello work.queue-12 时间：2025-02-05T14:53:02.014526300
消费者1接收到消息hello work.queue-39 时间：2025-02-05T14:53:02.073814400
消费者1接收到消息hello work.queue-41 时间：2025-02-05T14:53:02.142812400
消费者1接收到消息hello work.queue-43 时间：2025-02-05T14:53:02.199522100
***消费者2接收到消息hello work.queue-14 时间：2025-02-05T14:53:02.228114600
消费者1接收到消息hello work.queue-45 时间：2025-02-05T14:53:02.255591100
消费者1接收到消息hello work.queue-47 时间：2025-02-05T14:53:02.315954800
消费者1接收到消息hello work.queue-49 时间：2025-02-05T14:53:02.377632900
***消费者2接收到消息hello work.queue-16 时间：2025-02-05T14:53:02.440855300
***消费者2接收到消息hello work.queue-18 时间：2025-02-05T14:53:02.654015100
***消费者2接收到消息hello work.queue-20 时间：2025-02-05T14:53:02.867783300
***消费者2接收到消息hello work.queue-22 时间：2025-02-05T14:53:03.080905400
***消费者2接收到消息hello work.queue-24 时间：2025-02-05T14:53:03.296731200
***消费者2接收到消息hello work.queue-26 时间：2025-02-05T14:53:03.512099400
***消费者2接收到消息hello work.queue-28 时间：2025-02-05T14:53:03.725353500
***消费者2接收到消息hello work.queue-30 时间：2025-02-05T14:53:03.939706400
***消费者2接收到消息hello work.queue-32 时间：2025-02-05T14:53:04.152588100
***消费者2接收到消息hello work.queue-34 时间：2025-02-05T14:53:04.367337200
***消费者2接收到消息hello work.queue-36 时间：2025-02-05T14:53:04.581549200
***消费者2接收到消息hello work.queue-38 时间：2025-02-05T14:53:04.793774100
***消费者2接收到消息hello work.queue-40 时间：2025-02-05T14:53:05.006103400
***消费者2接收到消息hello work.queue-42 时间：2025-02-05T14:53:05.220121400
***消费者2接收到消息hello work.queue-44 时间：2025-02-05T14:53:05.433498300
***消费者2接收到消息hello work.queue-46 时间：2025-02-05T14:53:05.645486500
***消费者2接收到消息hello work.queue-48 时间：2025-02-05T14:53:05.856447600
***消费者2接收到消息hello work.queue-50 时间：2025-02-05T14:53:06.065771700可以看到消费者1和消费者2竟然每人消费了25条消息：
- 消费者1很快完成了自己的25条消息
- 消费者2却在缓慢的处理自己的25条消息。
也就是说消息是平均分配给每个消费者，并没有考虑到消费者的处理能力。导致1个消费者空闲，另一个消费者忙的不可开交。没有充分利用每一个消费者的能力，最终消息处理的耗时远远超过了1秒。这样显然是有问题的。
能者多劳
更改一下我们的配置文件，就好了。更改的是consumer消费者服务 application.yml 配置文件。
spring:
rabbitmq:
listener:
simple:
prefetch: 1 # 每次只能获取一条消息，处理完成才能获取下一个消息
重启项目，再次测试看结果。
消费者1接收到消息hello work.queue-1 时间：2025-02-05T16:19:40.610672600
***消费者2接收到消息hello work.queue-2 时间：2025-02-05T16:19:40.635078900
消费者1接收到消息hello work.queue-3 时间：2025-02-05T16:19:40.668399800
消费者1接收到消息hello work.queue-4 时间：2025-02-05T16:19:40.733468200
消费者1接收到消息hello work.queue-5 时间：2025-02-05T16:19:40.789432700
消费者1接收到消息hello work.queue-6 时间：2025-02-05T16:19:40.849740
***消费者2接收到消息hello work.queue-7 时间：2025-02-05T16:19:40.865255600
消费者1接收到消息hello work.queue-8 时间：2025-02-05T16:19:40.915186600
消费者1接收到消息hello work.queue-9 时间：2025-02-05T16:19:40.975302
消费者1接收到消息hello work.queue-10 时间：2025-02-05T16:19:41.035238100
消费者1接收到消息hello work.queue-11 时间：2025-02-05T16:19:41.098149900
***消费者2接收到消息hello work.queue-12 时间：2025-02-05T16:19:41.110162300
消费者1接收到消息hello work.queue-13 时间：2025-02-05T16:19:41.158752
消费者1接收到消息hello work.queue-14 时间：2025-02-05T16:19:41.214050800
消费者1接收到消息hello work.queue-15 时间：2025-02-05T16:19:41.275456500
消费者1接收到消息hello work.queue-16 时间：2025-02-05T16:19:41.338280900
***消费者2接收到消息hello work.queue-17 时间：2025-02-05T16:19:41.354040400
消费者1接收到消息hello work.queue-18 时间：2025-02-05T16:19:41.397333900
消费者1接收到消息hello work.queue-19 时间：2025-02-05T16:19:41.459536100
消费者1接收到消息hello work.queue-20 时间：2025-02-05T16:19:41.522984800
消费者1接收到消息hello work.queue-21 时间：2025-02-05T16:19:41.589369900
***消费者2接收到消息hello work.queue-22 时间：2025-02-05T16:19:41.595472400
消费者1接收到消息hello work.queue-23 时间：2025-02-05T16:19:41.639076100
消费者1接收到消息hello work.queue-24 时间：2025-02-05T16:19:41.702762100
消费者1接收到消息hello work.queue-25 时间：2025-02-05T16:19:41.761438700
消费者1接收到消息hello work.queue-26 时间：2025-02-05T16:19:41.823348300
***消费者2接收到消息hello work.queue-27 时间：2025-02-05T16:19:41.836398700
消费者1接收到消息hello work.queue-28 时间：2025-02-05T16:19:41.894946600
消费者1接收到消息hello work.queue-29 时间：2025-02-05T16:19:41.962451900
消费者1接收到消息hello work.queue-30 时间：2025-02-05T16:19:42.020227900
***消费者2接收到消息hello work.queue-31 时间：2025-02-05T16:19:42.066749100
消费者1接收到消息hello work.queue-32 时间：2025-02-05T16:19:42.080599800
消费者1接收到消息hello work.queue-33 时间：2025-02-05T16:19:42.143280700
消费者1接收到消息hello work.queue-34 时间：2025-02-05T16:19:42.204272700
消费者1接收到消息hello work.queue-35 时间：2025-02-05T16:19:42.270407300
***消费者2接收到消息hello work.queue-36 时间：2025-02-05T16:19:42.309818400
消费者1接收到消息hello work.queue-37 时间：2025-02-05T16:19:42.332003100
消费者1接收到消息hello work.queue-38 时间：2025-02-05T16:19:42.391974600
消费者1接收到消息hello work.queue-39 时间：2025-02-05T16:19:42.454012300
消费者1接收到消息hello work.queue-40 时间：2025-02-05T16:19:42.509398500
***消费者2接收到消息hello work.queue-41 时间：2025-02-05T16:19:42.555230800
消费者1接收到消息hello work.queue-42 时间：2025-02-05T16:19:42.570220
消费者1接收到消息hello work.queue-43 时间：2025-02-05T16:19:42.629378200
消费者1接收到消息hello work.queue-44 时间：2025-02-05T16:19:42.690519600
消费者1接收到消息hello work.queue-45 时间：2025-02-05T16:19:42.756214500
***消费者2接收到消息hello work.queue-46 时间：2025-02-05T16:19:42.797371400
消费者1接收到消息hello work.queue-47 时间：2025-02-05T16:19:42.813034800
消费者1接收到消息hello work.queue-48 时间：2025-02-05T16:19:42.876228100
消费者1接收到消息hello work.queue-49 时间：2025-02-05T16:19:42.939391
消费者1接收到消息hello work.queue-50 时间：2025-02-05T16:19:42.998590500可以发现，由于消费者1处理速度较快，所以处理了更多的消息；消费者2处理速度较慢，只处理了7条消息。而最终总的执行耗时也在1秒左右，大大提升。
正所谓能者多劳，这样充分利用了每一个消费者的处理能力，可以有效避免消息积压问题。
总结
Work模型的使用：
- 多个消费者绑定到一个队列，同一条消息只会被一个消费者处理
- 通过设置prefetch来控制消费者预取的消息数量
交换机类型
在之前的两个测试案例中，都没有交换机Exchange，生产者直接发送消息到队列。而一旦引入交换机，消息发送的模式会有很大变化：
可以看到，在订阅模型中，多了一个exchange角色，而且过程略有变化：
- Publisher：生产者，不再发送消息到队列中，而是发给交换机
- Exchange：交换机，一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。
- Queue：消息队列也与以前一样，接收消息、缓存消息。不过队列一定要与交换机绑定。
- Consumer：消费者，与以前一样，订阅队列，没有变化
Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！
交换机的类型有四种：
- Fanout：广播，将消息交给所有绑定到交换机的队列。我们最早在控制台使用的正是Fanout交换机
- Direct：订阅，基于RoutingKey（路由key）发送给订阅了消息的队列
- Topic：通配符订阅，与Direct类似，只不过RoutingKey可以使用通配符
- Headers：头匹配，基于MQ的消息头匹配，用的较少
文档中，我们讲解前面的三种交换机模式。
Fanout交换机
Fanout，英文翻译是扇出，我觉得在MQ中叫广播更合适。
在广播模式下，消息发送流程是这样的：
- 1） 可以有多个队列
- 2） 每个队列都要绑定到Exchange（交换机）
- 3） 生产者发送的消息，只能发送到交换机
- 4） 交换机把消息发送给绑定过的所有队列
- 5） 订阅队列的消费者都能拿到消息
我们的计划是这样的：
- 创建一个名为test.fanout的交换机，类型是Fanout
- 创建两个队列fanout.queue1和fanout.queue2，绑定到交换机test.fanout
声明交换机和队列
在控制台创建 fanout.queue1 和 fanout.queue2 两个队列。
然后在创建一个交换机
绑定两个队列到交换机
消息发送
在publisher服务的FanoutExchangeTest类中添加测试方法：
@SpringBootTest
public class FanoutExchangeTest {
@Autowired
private RabbitTemplate rabbitTemplate;
/*
测试 fanout exchange；
向 test.fanout 交换机发送消息，消息内容为 hello everyone!，会发送到所有绑定到该交换机的队列
*/
@Test
public void testSendFanoutExchange(){
// 1 交换机名称
String exchangeName = "test.fanout";
// 2 消息
String msg = "hello everyone！";
// 3 发送消息
rabbitTemplate.convertAndSend(exchangeName,"",msg);
}
}注意：上述的 convertAndSend 方法的第2个参数：路由key 因为没有绑定，所以可以指定为空
看看rabbitMQ的控制台
消息接收
在consumer服务中添加FanoutQueueListener类，并新增两个方法，监听队列中的消息 作为消费者。
@Slf4j
@Component
public class FanoutQueueListener {
/**
* 监听fanout.queue1队列
*/
@RabbitListener(queues = "fanout.queue1")
public void listenFanoutQueue1(String msg){
System.out.println("【消费者1】 接收到消息：" + msg);
}
/**
* 监听fanout.queue2队列
*/
@RabbitListener(queues = "fanout.queue2")
public void listenFanoutQueue2(String msg){
System.out.println("【消费者2】 接收到消息：" + msg);
}
}
总结
交换机的作用是什么？
- 接收publisher发送的消息
- 将消息按照规则路由到与之绑定的队列
- 不能缓存消息，路由失败，消息丢失
- FanoutExchange的会将消息路由到每个绑定的队列
Direct交换机
在Fanout模式中，一条消息，会被所有订阅的队列都消费。但是，在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。
在Direct模型下：
- 队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）
- 消息的发送方在向 Exchange发送消息时，也必须指定消息的 RoutingKey。
- Exchange不再把消息交给每一个绑定的队列，而是根据消息的Routing Key进行判断，只有队列的Routingkey与消息的 Routing key完全一致，才会接收到消息。
案例需求如图：
- 声明一个名为test.direct的交换机
- 声明队列direct.queue1，绑定hmall.direct，bindingKey为blud和red
- 声明队列direct.queue2，绑定hmall.direct，bindingKey为yellow和red
- 在consumer服务中，编写两个消费者方法，分别监听direct.queue1和direct.queue2
- 在publisher中编写测试方法，向test.direct发送消息
声明队列和交换机
首先在控制台声明两个队列direct.queue1和direct.queue2，这里不再展示过程：
然后声明一个direct类型的交换机，命名为test.direct:
然后使用red和blue作为key，绑定direct.queue1到test.direct：
绑定diretc.queue2
看看最后的绑定关系
消息发送
在publish服务中，新增 DirectExchangeTest 类发送消息。
@SpringBootTest
public class DirectExchangeTest {
@Autowired
private RabbitTemplate rabbitTemplate;
/*
测试 direct exchange；
向 test.direct 交换机发送消息，会根据路由key发送到所有绑定到该交换机的队列
*/
@Test
public void testSendDirectExchange(){
// 1 交换机
String exchangeName = "test.direct";
// 2 消息
String msg = "这是一条消息，并且路由key是red 红色。";
// 3 发送消息 路由key为red
rabbitTemplate.convertAndSend(exchangeName,"red",msg);
//改变下消息
msg = "这是一条消息，并且路由key是blue 蓝色。";
rabbitTemplate.convertAndSend(exchangeName,"blue",msg);
}
}看看rabbitMQ控制台，查看消息是否成功发送。
消息接收
在consumer服务中，添加 DirectQueueListener 类，并在里面编写两个方法。
@Slf4j
@Component
public class DirectQueueListener {
/**
* 监听direct.queue1队列
*/
@RabbitListener(queues = "direct.queue1")
public void listenDirectQueue1(String msg){
log.info("【消费者1】接收到消息：{}",msg);
}
/**
* 监听direct.queue2队列
* @param msg
*/
@RabbitListener(queues = "direct.queue2")
public void listenDirectQueue2(String msg){
log.info("【消费者2】接收到消息：{}",msg);
}
}
由于 test.redirect 交换机绑定的两个队列的路由key有red；所以指定了路由key为red的消息能被两个消费者都收到。
而路由key为 blue 的队列只有direct.queue1；所以只有监听这个队列的 消费者1 能够接收到消息：
总结
描述下Direct交换机与Fanout交换机的差异？
- Fanout交换机将消息路由给每一个与之绑定的队列
- Direct交换机根据RoutingKey判断路由给哪个队列
- 如果多个队列具有相同的RoutingKey，则与Fanout功能类似
Topic交换机
Topic类型交换机
Topic类型的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。
只不过Topic类型Exchange可以让队列在绑定RoutingKey 的时候使用通配符！
RoutingKey 一般都是有一个或多个单词组成，多个单词之间以.分割，例如： item.insert
通配符规则：
- #：匹配一个或多个词
- *：匹配不多不少恰好1个词
举例：
- item.#：能够匹配item.spu.insert 或者 item.spu
- item.*：只能匹配item.spu
图示：
假如此时publisher发送的消息使用的RoutingKey共有四种：
- china.news代表有中国的新闻消息；
- china.weather 代表中国的天气消息；
- japan.news 则代表日本新闻
- japan.weather 代表日本的天气消息；
解释：
- topic.queue1：绑定的是china.# ，凡是以 china.开头的routing key 都会被匹配到，包括：
-
- china.news
- china.weather
- topic.queue2：绑定的是#.news ，凡是以 .news结尾的 routing key 都会被匹配。包括:
-
- china.news
- japan.news
接下来，我们就按照上图所示，来演示一下Topic交换机的用法。
首先，在控制台按照图示例子创建队列、交换机，并利用通配符绑定队列和交换机。此处步骤略。最终结果如下：
创建交换机和队列
创建test.topic 交换机
看看效果
给test.topic 交换机绑定两个队列
消息发送
在consumer服务中，新增 TopicExchangeTest类 发送消息。
@SpringBootTest
public class TopicExchangeTest {
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
public void testSendTopicExchange(){
// 1 交换机
String exchangeName = "test.topic";
// 2 消息
String msg = "我是TopicExchange交换机的消息,路由key是 china.news";
// 3 发送路由key为 china.news 的消息
rabbitTemplate.convertAndSend(exchangeName,"china.news",msg);
}
}
消息接收
在consumer服务中，添加 TopicExchangeListener 类，编写两个方法监听消息。
@Slf4j
@Component
public class TopicExchangeListener {
/**
* 监听topic.queue1队列
*/
@RabbitListener(queues = "topic.queue1")
public void listenTopicQueue1(String msg) {
log.info("【消费者1】监听到消息：{}", msg);
}
/**
* 监听topic.queue2队列
*/
@RabbitListener(queues ="topic.queue2")
public void listenTopicQueue2(String msg) {
log.info("【消费者2】监听到消息：{}", msg);
}
}
总结
描述下Direct交换机与Topic交换机的差异？
- Topic交换机接收的消息RoutingKey必须是多个单词，以 . 分割
- Topic交换机与队列绑定时的RoutingKey可以指定通配符
- #：代表0个或多个词
- *：代表1个词
代码声明交换机和队列
在之前我们都是基于RabbitMQ控制台来创建队列、交换机。但是在实际开发时，队列和交换机是程序员定义的，将来项目上线，又要交给运维去创建。那么程序员就需要把程序中运行的所有队列和交换机都写下来，交给运维。在这个过程中是很容易出现错误的。
因此推荐的做法是由程序启动时检查队列和交换机是否存在，如果不存在自动创建。
基本API
SpringAMQP提供了一个Queue类，用来创建队列：
SpringAMQP还提供了一个Exchange接口，来表示所有不同类型的交换机：
我们可以自己创建队列和交换机，不过SpringAMQP还提供了ExchangeBuilder来简化这个过程：
而在绑定队列和交换机时，则需要使用BindingBuilder来创建Binding对象：
把之前创建的队列和交换机删除
删除后的队列
删除后的交换机
Ideal控制台报错
这是因为我们的队列和交换机都删除了，里面写的 RabbitListener 还在监听队列中的消息，但是队列没有了，所以报错。
fanout示例
在consumer服务中，新建config包。并创建FanoutConfig 类 在里面编写代码，创建test.fanout 交换机和fanout.queue1 和fanout.queue2 队列。 并启动consumer服务
@Configuration
public class FanoutConfig {
// 声明 Fanout 类型的交换机
@Bean
public FanoutExchange fanoutExchange(){
return new FanoutExchange("test.fanout");
}
//声明队列，名称为 fanout.queue1
@Bean
public Queue fanoutQueue1(){
return new Queue("fanout.queue1");
}
//绑定队列和交换机
@Bean
public Binding fanoutBinding1(FanoutExchange fanoutExchange,Queue fanoutQueue1){
return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
}
//声明队列，名称为 fanout.queue2
@Bean
public Queue fanoutQueue2(){
return new Queue("fanout.queue2");
}
//绑定队列和交换机
@Bean
public Binding fanoutBinding2(FanoutExchange fanoutExchange,Queue fanoutQueue2){
return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
}
}看看rbbitMQ控制台效果
看看交换机
Direct示例
在consumer 服务中的 config包中，新建 DirectConfig 类，编写代码创建交换机和队列。direct模式由于要绑定多个key，会比较麻烦一点，因为每一个key都要写一个binding方法。
@Configuration
public class DirectConfig {
//声明 test.direct 交换机
@Bean
public DirectExchange directExchange(){
return new DirectExchange("test.direct");
}
//声明 direct.queue1 队列
@Bean
public Queue directQueue1(){
return new Queue("direct.queue1");
}
//绑定 direct.queue1 队列到 test.direct 交换机上 路由key是 red
@Bean
public Binding directBindingQueue1Red(DirectExchange directExchange,Queue directQueue1){
return BindingBuilder.bind(directQueue1).to(directExchange).with("red");
}
//绑定 direct.queue1 队列到 test.direct 交换机上 路由key是 blue
@Bean
public Binding directBindingQueue1Blue(DirectExchange directExchange,Queue directQueue1){
return BindingBuilder.bind(directQueue1).to(directExchange).with("blue");
}
//声明 direct.queue2 队列
@Bean
public Queue directQueue2(){
return new Queue("direct.queue2");
}
//绑定 direct.queue2 队列到 test.direct 交换机上 路由key是 red
@Bean
public Binding directBindingQueue2Red(DirectExchange directExchange, Queue directQueue2){
return BindingBuilder.bind(directQueue2).to(directExchange).with("red");
}
//绑定 direct.queue2 队列到 test.direct 交换机上 路由key是 yellow
@Bean
public Binding directBindingQueue2Yellow(DirectExchange directExchange,Queue directQueue2){
return BindingBuilder.bind(directQueue2).to(directExchange).with("yellow");
}
}看看rabbitMQ控制台
看看交换机和绑定关系
Topic示例
在consumer 服务中的config包里面，新创建TopicConfig类，编写代码创建交换机和队列。
@Configuration
public class TopicConfig {
//声明 test.topic 交换机
@Bean
public TopicExchange topicExchange(){
return new TopicExchange("test.topic");
}
//声明 topic.queue1 队列
@Bean
public Queue topicQueue1(){
return new Queue("topic.queue1");
}
//绑定队列和交换机 路由key是 china.#
@Bean
public Binding topicBinding1(TopicExchange topicExchange,Queue topicQueue1){
return BindingBuilder.bind(topicQueue1).to(topicExchange).with("china.#");
}
//声明 topic.queue2 队列
@Bean
public Queue topicQueue2(){
return new Queue("topic.queue2");
}
//绑定队列和交换机 路由key是 #.news
@Bean
public Binding topicBinding2(TopicExchange topicExchange,Queue topicQueue2){
return BindingBuilder.bind(topicQueue2).to(topicExchange).with("#.news");
}
}看看控制台效果
交换机
基于注解声明
基于@Bean的方式声明队列和交换机比较麻烦，Spring还提供了基于注解方式来声明。不过是在消息监听的时候基于注解的方式来声明。
例如，我们同样声明Direct模式的交换机和队列；用注解的方式声明下。
先把之前创建的 交换机和队列删除。
删除后的效果
Fanout示例
@Configuration
public class FanoutRabbitListener {
// 监听fanout.queue1 队列的消息
@RabbitListener(bindings = @QueueBinding(
value = @Queue("fanout.queue1"),
exchange = @Exchange(value = "test.fanout",type = ExchangeTypes.FANOUT),
key = ""
))
public void listenFanoutQueue1(String msg){
System.out.println("【消费者1】 监听到消息" + msg);
}
// 监听fanout.queue2 队列的消息
@RabbitListener(bindings = @QueueBinding(
value = @Queue("fanout.queue2"),
exchange = @Exchange(value = "test.fanout",type = ExchangeTypes.FANOUT),
key = ""
))
public void listenFanoutQueue2(String msg){
System.out.println("【消费者2】 监听到消息" + msg);
}
}启动consumer服务看效果
交换机和绑定关系
Direct示例
新建 DirectRabbitListener 类，并在里面编写代码进行测试。
@Configuration
public class DirectRabbitListener {
// 声明 direct.queue1
@RabbitListener(bindings = @QueueBinding(
value = @Queue("direct.queue1"),
exchange = @Exchange(value = "test.direct",type = ExchangeTypes.DIRECT),
key = {"red","blue"}
))
public void listenDirectQueue1(String msg){
System.out.println("【消费者1】 接收到消息：" + msg);
}
// 声明direct.queue2
@RabbitListener(bindings =@QueueBinding(
value = @Queue("direct.queue2"),
exchange = @Exchange(value = "test.direct",type = ExchangeTypes.DIRECT),
key = {"red","yellow"}
))
public void listenDirectQueue2(String msg){
System.out.println("【消费者2】 接收到消息：" + msg);
}
}看看效果
交换机和绑定关系
Topic示例
在consumer服务中的 config包里面，创建TopicRabbitListener类。编写代码进行测试
@Configuration
public class TopicRabbitListener {
//声明topic.queue1 队列
@RabbitListener(bindings = @QueueBinding(
value = @Queue("topic.queue1"),
exchange = @Exchange(value = "test.topic",type = ExchangeTypes.TOPIC),
key = {"china.#"}
))
public void listenTopicQueue1(String msg){
System.out.println("【消费者1】接收到消息："+msg);
}
//声明 topic.queue2 队列
@RabbitListener(bindings = @QueueBinding(
value = @Queue("topic.queue2"),
exchange = @Exchange(value = "test.topic",type = ExchangeTypes.TOPIC),
key = {"#.news"}
))
public void listenTopicQueue2(String msg){
System.out.println("【消费者2】接收到消息："+msg);
}
}看看效果
交换机和绑定关系
