使用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 # 你的虚拟机IPport: 5672          # rabbitMQ端口virtual-host: /sde  # 虚拟机名称username: sundaoen  # 用户名password: 8888888888       # 密码

编写测试类

在我们项目的publisher中创建测试类，并且利用 RabbitTemplate 发送消息。

@SpringBootTest
public class TestSendMessage {@Autowiredprivate RabbitTemplate rabbitTemplate;@Testpublic 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 # 你的虚拟机IPport: 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 {@Autowiredprivate RabbitTemplate rabbitTemplate;@Testpublic 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 {@Autowiredprivate RabbitTemplate rabbitTemplate;/*测试 fanout exchange；向 test.fanout 交换机发送消息，消息内容为 hello everyone!，会发送到所有绑定到该交换机的队列*/@Testpublic 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完全一致，才会接收到消息。

案例需求如图：

1. 声明一个名为test.direct的交换机
2. 声明队列direct.queue1，绑定hmall.direct，bindingKey为blud和red
3. 声明队列direct.queue2，绑定hmall.direct，bindingKey为yellow和red
4. 在consumer服务中，编写两个消费者方法，分别监听direct.queue1和direct.queue2
5. 在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 {@Autowiredprivate RabbitTemplate rabbitTemplate;/*测试 direct exchange；向 test.direct 交换机发送消息，会根据路由key发送到所有绑定到该交换机的队列*/@Testpublic void testSendDirectExchange(){// 1 交换机String exchangeName = "test.direct";// 2 消息String msg = "这是一条消息，并且路由key是red 红色。";// 3 发送消息 路由key为redrabbitTemplate.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 {@Autowiredprivate RabbitTemplate rabbitTemplate;@Testpublic 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 类型的交换机@Beanpublic FanoutExchange fanoutExchange(){return new FanoutExchange("test.fanout");}//声明队列，名称为 fanout.queue1@Beanpublic Queue fanoutQueue1(){return new Queue("fanout.queue1");}//绑定队列和交换机@Beanpublic Binding fanoutBinding1(FanoutExchange fanoutExchange,Queue fanoutQueue1){return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);}//声明队列，名称为 fanout.queue2@Beanpublic Queue fanoutQueue2(){return new Queue("fanout.queue2");}//绑定队列和交换机@Beanpublic 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 交换机@Beanpublic DirectExchange directExchange(){return new DirectExchange("test.direct");}//声明 direct.queue1 队列@Beanpublic Queue directQueue1(){return new Queue("direct.queue1");}//绑定 direct.queue1 队列到 test.direct 交换机上 路由key是 red@Beanpublic Binding directBindingQueue1Red(DirectExchange directExchange,Queue directQueue1){return BindingBuilder.bind(directQueue1).to(directExchange).with("red");}//绑定 direct.queue1 队列到 test.direct 交换机上 路由key是 blue@Beanpublic Binding directBindingQueue1Blue(DirectExchange directExchange,Queue directQueue1){return BindingBuilder.bind(directQueue1).to(directExchange).with("blue");}//声明 direct.queue2 队列@Beanpublic Queue directQueue2(){return new Queue("direct.queue2");}//绑定 direct.queue2 队列到 test.direct 交换机上 路由key是 red@Beanpublic Binding directBindingQueue2Red(DirectExchange directExchange, Queue directQueue2){return BindingBuilder.bind(directQueue2).to(directExchange).with("red");}//绑定 direct.queue2 队列到 test.direct 交换机上 路由key是 yellow@Beanpublic 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 交换机@Beanpublic TopicExchange topicExchange(){return new TopicExchange("test.topic");}//声明 topic.queue1 队列@Beanpublic Queue topicQueue1(){return new Queue("topic.queue1");}//绑定队列和交换机 路由key是 china.#@Beanpublic Binding topicBinding1(TopicExchange topicExchange,Queue topicQueue1){return BindingBuilder.bind(topicQueue1).to(topicExchange).with("china.#");}//声明 topic.queue2 队列@Beanpublic Queue topicQueue2(){return new Queue("topic.queue2");}//绑定队列和交换机 路由key是 #.news@Beanpublic 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);}
}

看看效果

交换机和绑定关系