利用java代码操作rabbitMQ收发消息

美丽的神话 · 4 天前

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>
<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");
}
}