JAVA 8
Spring Boot 2.5.3
RabbitMQ 3.6.10, Erlang 20.2.2（单机）
---
授人以渔：
1、Spring Boot Reference Documentation
This document is also available as Multi-page HTML, Single page HTML and PDF.
有PDF版本哦，下载下来！
有PDF版本哦，下载下来！
当前最新版 RabbitMQ（官网）：
RabbitMQ 3.9.4 release at 17 Aug 2021
目录
试验3：多个Listener接收DirectExchange转发到队列queue1的消息
消息队列（MQ）是一种重要的中间件，可以实现 1）进程、应用间通信；2）异步任务处理，可以很好地实现 程序的低耦合。
AMQP，即Advanced Message Queuing Protocol，一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计（百度百科）。
AMQP是一种 线路级协议（wire-level protocol）——遵循这个协议的实现 之间是可以无缝互操作的。
RabbitMQ是使用 Erlang语言实现了 AMQP协议。
Spring Boot中可以方便地集成RabbitMQ，依赖 spring-boot-starter-amqp 包即可：
本文介绍 S.B. 集成 RabbitMQ 及其基础使用场景。
补充：
RabbitMQ启动后，默认使用下面端口：
| 5672 | 服务器端口，S.B.中配置 | 常用 |
| 15672 | Web服务器端口，可以对RabbitMQ服务器进行管理 | 常用 |
| 25672 | 用于节点间和CLI工具通信（Erlang分发服务器端口），并从动态范围分配 | 不常用 |
| 4369 | EPMD默认端口 | 不常用 |
注，请看 参考文档（原来有这么多）。
依赖 spring-boot-starter-amqp 包后，Spring容器中会存在一下相关Bean：
rabbitConnectionFactory
rabbitTemplateConfigurer
rabbitTemplate # 用过，发送消息
amqpAdmin # 尚未用过
rabbitMessagingTemplate
simpleRabbitListenerContainerFactoryConfigurer
rabbitListenerContainerFactory
directRabbitListenerContainerFactoryConfigurer
RabbitAutoConfiguration
RabbitProperties
RabbitAnnotationDrivenConfiguration
RabbitAnnotationDrivenConfiguration$EnableRabbitConfiguration
RabbitAutoConfiguration
RabbitAutoConfiguration$RabbitConnectionFactoryCreator
RabbitAutoConfiguration$RabbitTemplateConfiguration
RabbitAutoConfiguration$MessagingTemplateConfigurationrabbitTemplate 是其中 （自己）最常用（重要）的，用来发送消息。
public class RabbitTemplate extends RabbitAccessor // NOSONAR type line count
implements BeanFactoryAware, RabbitOperations, ChannelAwareMessageListener,
ListenerContainerAware, PublisherCallbackChannel.Listener, BeanNameAware, DisposableBean {
}
其下的 convertAndSend(...) 是用来发送消息的：
注，convertSendAndReceive(...) 是 和 Basic RPC pattern 相关，还没用过（不懂）。
还有一些 send(...) 函数，也是用来发送消息（和 上面的 convertAndSend(...) 有什么区别呢？）：
RabbitMQ关键概念：
消息队列（Queue）
交换机（Exchange）
绑定（Binding）
路由键值（routingKey）
频道（channel）
---
消息发送到 交换机，交换机 根据 routingKey 转发到不同队列，再由 队列监听者（Spring 中 @RabbitListener 定义） 接收&处理；
其中，队列需要绑定到交换机，其中，每个队列都会绑定到 默认交换机——Direct类型。
默认交换机：AMQP default
Queue、Exchange、Binding都继承了 AbstractDeclarable 类：
public abstract class AbstractExchange extends AbstractDeclarable implements Exchange {
}
public class Queue extends AbstractDeclarable implements Cloneable {
}
public class Binding extends AbstractDeclarable {
}
public abstract class AbstractExchange extends AbstractDeclarable implements Exchange {
}
AbstractExchange 下 有几个子类——Exchange，使用RabbitMQ时会经常用到：
RabbitMQ支持多种消息传输模式，点对点、广播、发布-订阅 等都支持，这需要使用不同的交换机来实现。
下面是一章经典的图：
首先，定义队列queue1，定义队列监听器（@RabbitListener），然后，调用rabbitTemplate.convertAndSend 发送消息。
配置文件：
#
# RabbitMQ
# mylinux 为 虚拟机，hosts中配置（单机模式）
spring.rabbitmq.host=mylinux
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=admin注，上面的配置 可以考虑试验 spring.rabbitmq.addresses 替代——一个搞定。
spring.rabbitmq.addresses=amqp://admin:admin@mylinux:5672试验1源码
# 文件RabbitConfig.java
@Configuration
public class RabbitConfig {
public final static String QUEUE_1 = "queue1";
@Bean
public Queue queue1() {
Queue queue1 = new Queue(QUEUE_1);
return queue1;
}
}
# 文件RabbitmqHelloApplication.java （主类，部分）
@Component
@Order(1)
@Slf4j
class TestRunner1 implements CommandLineRunner {
@Autowired
private RabbitTemplate rabbitTemplate;
@Override
public void run(String... args) throws Exception {
/**
* 每隔2秒发送1条消息
* 看看 是否到 queue1、是否能监听到
*/
for (int i=0; i<10; i++) {
log.info("发送msg-{}", i);
// 默认交换机 AMQP default，默认的 routingKey=""，，怎么接收并处理？TODO
// rabbitTemplate.convertAndSend("Simple msg-" + i + ", @" + new Date());
// 默认交换机，routingKey=queue1，转发到 queue队列
rabbitTemplate.convertAndSend(RabbitConfig.QUEUE_1, "Simple msg-" + i + ", @" + new Date());
TimeUnit.SECONDS.sleep(1L);
}
}
}
@Component
@Slf4j
class RabbitListener1 {
/**
* 接收队列queue1 的消息，默认绑定到 默认交换机（AMQP default）
* @author ben
* @date 2021-08-23 22:31:58 CST
* @param content
*/
@RabbitListener(queues = {RabbitConfig.QUEUE_1})
public void processMessage(String content) {
log.info("接收消息：content={}", content);
}
}试验1测试结果：
日志1
2021-08-23 23:40:14.380 INFO 13700 --- [ main] org.lib.rabbitmqhello.TestRunner1 : 发送msg-0
2021-08-23 23:40:14.401 INFO 13700 --- [ntContainer#0-1] org.lib.rabbitmqhello.RabbitListener1 : 接收消息：content=Simple msg-0, @Mon Aug 23 23:40:14 CST 2021
2021-08-23 23:40:15.389 INFO 13700 --- [ main] org.lib.rabbitmqhello.TestRunner1 : 发送msg-1
2021-08-23 23:40:15.393 INFO 13700 --- [ntContainer#0-1] org.lib.rabbitmqhello.RabbitListener1 : 接收消息：content=Simple msg-1, @Mon Aug 23 23:40:15 CST 2021
2021-08-23 23:40:16.390 INFO 13700 --- [ main] org.lib.rabbitmqhello.TestRunner1 : 发送msg-2
2021-08-23 23:40:16.392 INFO 13700 --- [ntContainer#0-1] org.lib.rabbitmqhello.RabbitListener1 : 接收消息：content=Simple msg-2, @Mon Aug 23 23:40:16 CST 2021
2021-08-23 23:40:17.390 INFO 13700 --- [ main] org.lib.rabbitmqhello.TestRunner1 : 发送msg-3
2021-08-23 23:40:17.394 INFO 13700 --- [ntContainer#0-1] org.lib.rabbitmqhello.RabbitListener1 : 接收消息：content=Simple msg-3, @Mon Aug 23 23:40:17 CST 2021
2021-08-23 23:40:18.391 INFO 13700 --- [ main] org.lib.rabbitmqhello.TestRunner1 : 发送msg-4
2021-08-23 23:40:18.401 INFO 13700 --- [ntContainer#0-1] org.lib.rabbitmqhello.RabbitListener1 : 接收消息：content=Simple msg-4, @Mon Aug 23 23:40:18 CST 2021
2021-08-23 23:40:19.392 INFO 13700 --- [ main] org.lib.rabbitmqhello.TestRunner1 : 发送msg-5
2021-08-23 23:40:19.397 INFO 13700 --- [ntContainer#0-1] org.lib.rabbitmqhello.RabbitListener1 : 接收消息：content=Simple msg-5, @Mon Aug 23 23:40:19 CST 2021
2021-08-23 23:40:20.394 INFO 13700 --- [ main] org.lib.rabbitmqhello.TestRunner1 : 发送msg-6
2021-08-23 23:40:20.401 INFO 13700 --- [ntContainer#0-1] org.lib.rabbitmqhello.RabbitListener1 : 接收消息：content=Simple msg-6, @Mon Aug 23 23:40:20 CST 2021
2021-08-23 23:40:21.395 INFO 13700 --- [ main] org.lib.rabbitmqhello.TestRunner1 : 发送msg-7
2021-08-23 23:40:21.399 INFO 13700 --- [ntContainer#0-1] org.lib.rabbitmqhello.RabbitListener1 : 接收消息：content=Simple msg-7, @Mon Aug 23 23:40:21 CST 2021
2021-08-23 23:40:22.397 INFO 13700 --- [ main] org.lib.rabbitmqhello.TestRunner1 : 发送msg-8
2021-08-23 23:40:22.401 INFO 13700 --- [ntContainer#0-1] org.lib.rabbitmqhello.RabbitListener1 : 接收消息：content=Simple msg-8, @Mon Aug 23 23:40:22 CST 2021
2021-08-23 23:40:23.398 INFO 13700 --- [ main] org.lib.rabbitmqhello.TestRunner1 : 发送msg-9
2021-08-23 23:40:23.407 INFO 13700 --- [ntContainer#0-1] org.lib.rabbitmqhello.RabbitListener1 : 接收消息：content=Simple msg-9, @Mon Aug 23 23:40:23 CST 2021本试验试验的是 默认交换机——AMQP default。
在开始发送时，使用了1个参数的 convertAndSend，结果，发送成功，但监听器没反应。原来，此时使用了 默认的routingKey=""，队列没有收到。
改为 上面的 2个参数的convertAndSend 后，监听器才收到了发送的消息——routingKey为 队列名。
RabbitConfig 里面定义的队列是必须的，否则，启动会发生异常：
o.s.a.r.listener.BlockingQueueConsumer : Failed to declare queue: queue1
o.s.a.r.listener.BlockingQueueConsumer : Queue declaration failed; retries left=3
org.springframework.amqp.rabbit.listener.BlockingQueueConsumer$DeclarationException: Failed to declare queue(s):[queue1]小结：默认交换机
1、发送使用默认routingKey，无法进入队列，对于的Listener监听不到；
2、发送使用 队列名 作为 routingKey，Listener接收成功；
在 试验1 的基础上，定义 DirectExchange类型directExchange1——名称ex1、Binding类型binding1——绑定queue1和ex1，发送时，发送到 交换机ex1。
源码改动：
试验2源码
# RabbitConfig.java
@Configuration
public class RabbitConfig {
public final static String QUEUE_1 = "queue1";
public final static String EXCHANGE_1 = "ex1";
@Bean
public Queue queue1() {
Queue queue1 = new Queue(QUEUE_1);
return queue1;
}
@Bean
public DirectExchange directExchange1() {
return new DirectExchange(EXCHANGE_1);
}
@Bean
public Binding binding1(Queue queue1, DirectExchange directExchange1) {
// 绑定队列queue1到交换机ex1，使用队列名作为 routingKey
return BindingBuilder
.bind(queue1)
.to(directExchange1)
.withQueueName();
}
}
# RabbitmqHelloApplication.java
# 消息多了 default 前缀
rabbitTemplate.convertAndSend(RabbitConfig.QUEUE_1, "default-Simple msg-" + i + ", @" + new Date());
# TestRunner1#run 中增加下面的发送语句
// ex1交换机=>queue1
rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_1, RabbitConfig.QUEUE_1, "ex1-Simple msg-" + i + ", @" + new Date());注，保留了 发送消息到 默认交换机。
测试结果：
日志2
2021-08-23 23:56:05.294 INFO 13500 --- [ main] org.lib.rabbitmqhello.TestRunner1 : 发送msg-0
2021-08-23 23:56:05.316 INFO 13500 --- [ntContainer#0-1] org.lib.rabbitmqhello.RabbitListener1 : 接收消息：content=default-Simple msg-0, @Mon Aug 23 23:56:05 CST 2021
2021-08-23 23:56:05.317 INFO 13500 --- [ntContainer#0-1] org.lib.rabbitmqhello.RabbitListener1 : 接收消息：content=ex1-Simple msg-0, @Mon Aug 23 23:56:05 CST 2021
2021-08-23 23:56:06.303 INFO 13500 --- [ main] org.lib.rabbitmqhello.TestRunner1 : 发送msg-1
2021-08-23 23:56:06.313 INFO 13500 --- [ntContainer#0-1] org.lib.rabbitmqhello.RabbitListener1 : 接收消息：content=default-Simple msg-1, @Mon Aug 23 23:56:06 CST 2021
2021-08-23 23:56:06.315 INFO 13500 --- [ntContainer#0-1] org.lib.rabbitmqhello.RabbitListener1 : 接收消息：content=ex1-Simple msg-1, @Mon Aug 23 23:56:06 CST 2021
2021-08-23 23:56:07.306 INFO 13500 --- [ main] org.lib.rabbitmqhello.TestRunner1 : 发送msg-2
2021-08-23 23:56:07.311 INFO 13500 --- [ntContainer#0-1] org.lib.rabbitmqhello.RabbitListener1 : 接收消息：content=default-Simple msg-2, @Mon Aug 23 23:56:07 CST 2021
2021-08-23 23:56:07.311 INFO 13500 --- [ntContainer#0-1] org.lib.rabbitmqhello.RabbitListener1 : 接收消息：content=ex1-Simple msg-2, @Mon Aug 23 23:56:07 CST 2021
2021-08-23 23:56:08.307 INFO 13500 --- [ main] org.lib.rabbitmqhello.TestRunner1 : 发送msg-3
2021-08-23 23:56:08.316 INFO 13500 --- [ntContainer#0-1] org.lib.rabbitmqhello.RabbitListener1 : 接收消息：content=default-Simple msg-3, @Mon Aug 23 23:56:08 CST 2021
2021-08-23 23:56:08.317 INFO 13500 --- [ntContainer#0-1] org.lib.rabbitmqhello.RabbitListener1 : 接收消息：content=ex1-Simple msg-3, @Mon Aug 23 23:56:08 CST 2021
2021-08-23 23:56:09.310 INFO 13500 --- [ main] org.lib.rabbitmqhello.TestRunner1 : 发送msg-4
2021-08-23 23:56:09.315 INFO 13500 --- [ntContainer#0-1] org.lib.rabbitmqhello.RabbitListener1 : 接收消息：content=default-Simple msg-4, @Mon Aug 23 23:56:09 CST 2021
2021-08-23 23:56:09.316 INFO 13500 --- [ntContainer#0-1] org.lib.rabbitmqhello.RabbitListener1 : 接收消息：content=ex1-Simple msg-4, @Mon Aug 23 23:56:09 CST 2021
2021-08-23 23:56:10.311 INFO 13500 --- [ main] org.lib.rabbitmqhello.TestRunner1 : 发送msg-5
2021-08-23 23:56:10.318 INFO 13500 --- [ntContainer#0-1] org.lib.rabbitmqhello.RabbitListener1 : 接收消息：content=default-Simple msg-5, @Mon Aug 23 23:56:10 CST 2021
2021-08-23 23:56:10.320 INFO 13500 --- [ntContainer#0-1] org.lib.rabbitmqhello.RabbitListener1 : 接收消息：content=ex1-Simple msg-5, @Mon Aug 23 23:56:10 CST 2021
2021-08-23 23:56:11.312 INFO 13500 --- [ main] org.lib.rabbitmqhello.TestRunner1 : 发送msg-6
2021-08-23 23:56:11.315 INFO 13500 --- [ntContainer#0-1] org.lib.rabbitmqhello.RabbitListener1 : 接收消息：content=default-Simple msg-6, @Mon Aug 23 23:56:11 CST 2021
2021-08-23 23:56:11.315 INFO 13500 --- [ntContainer#0-1] org.lib.rabbitmqhello.RabbitListener1 : 接收消息：content=ex1-Simple msg-6, @Mon Aug 23 23:56:11 CST 2021
2021-08-23 23:56:12.312 INFO 13500 --- [ main] org.lib.rabbitmqhello.TestRunner1 : 发送msg-7
2021-08-23 23:56:12.314 INFO 13500 --- [ntContainer#0-1] org.lib.rabbitmqhello.RabbitListener1 : 接收消息：content=default-Simple msg-7, @Mon Aug 23 23:56:12 CST 2021
2021-08-23 23:56:12.315 INFO 13500 --- [ntContainer#0-1] org.lib.rabbitmqhello.RabbitListener1 : 接收消息：content=ex1-Simple msg-7, @Mon Aug 23 23:56:12 CST 2021
2021-08-23 23:56:13.313 INFO 13500 --- [ main] org.lib.rabbitmqhello.TestRunner1 : 发送msg-8
2021-08-23 23:56:13.318 INFO 13500 --- [ntContainer#0-1] org.lib.rabbitmqhello.RabbitListener1 : 接收消息：content=default-Simple msg-8, @Mon Aug 23 23:56:13 CST 2021
2021-08-23 23:56:13.318 INFO 13500 --- [ntContainer#0-1] org.lib.rabbitmqhello.RabbitListener1 : 接收消息：content=ex1-Simple msg-8, @Mon Aug 23 23:56:13 CST 2021
2021-08-23 23:56:14.315 INFO 13500 --- [ main] org.lib.rabbitmqhello.TestRunner1 : 发送msg-9
2021-08-23 23:56:14.319 INFO 13500 --- [ntContainer#0-1] org.lib.rabbitmqhello.RabbitListener1 : 接收消息：content=default-Simple msg-9, @Mon Aug 23 23:56:14 CST 2021
2021-08-23 23:56:14.324 INFO 13500 --- [ntContainer#0-1] org.lib.rabbitmqhello.RabbitListener1 : 接收消息：content=ex1-Simple msg-9, @Mon Aug 23 23:56:14 CST 2021
使用队列名作为 routingKey，来自 默认交换机、自定义交换机ext1 的消息都收到了。
Web管理页面变化：
监听器收到了 2条消息！
根据 Web管理页面得到的信息：队列 无法和 默认交换机，因此，如果按照上面的方式 发送消息，那么，默认交换机会转发到 queue1 队列。
Default exchange
The default exchange is implicitly bound to every queue, with a routing key equal to the queue name.
It is not possible to explicitly bind to, or unbind from the default exchange.
It also cannot be deleted.试验3：多个Listener接收DirectExchange转发到队列queue1的消息
注，取消上面的 TestRunner1 的消息发送，改为调用接口。
测试源码：
# 调用接口 /try1/send 发送消息
@RestController
@RequestMapping(value="/try1")
@Slf4j
public class Try1Controller {
@Autowired
private RabbitTemplate rabbitTemplate;
/**
* 使用 交换机ext1 发送消息到 队列queue1
* @author ben
* @date 2021-08-27 21:24:19 CST
* @return
*/
@GetMapping(value="/send")
public boolean send() {
IntStream.range(0, 10).forEach(i->{
// ex1交换机=>queue1
rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_1, RabbitConfig.QUEUE_1,
"Try1Controller-ex1-Simple msg-" + i + ", @" + new Date());
try {
TimeUnit.SECONDS.sleep(1L);
} catch (InterruptedException e) {
log.info("异常：{}", e.getMessage());
}
});
return true;
}
}
# 两个监听器
@Component
@Slf4j
class RabbitListener1 {
/**
* Listener1
* @author ben
* @date 2021-08-23 22:31:58 CST
* @param content
*/
@RabbitListener(queues = {RabbitConfig.QUEUE_1})
public void processMessage1(String content) {
log.info("接收消息：processMessage1-content={}", content);
}
/**
* Listener2
* @author ben
* @date 2021-08-27 21:26:30 CST
* @param content
*/
@RabbitListener(queues = {RabbitConfig.QUEUE_1})
public void processMessage2(String content) {
log.info("接收消息：processMessage2-content={}", content);
}
}一个队列，两个监听器。来自博客园
执行结果：
调用接口 /try1/send，共发送10条消息，两个监听器都收到了，分别5条——实现了均衡。
一个Fanout交换机，两个队列，绑定。
配置：
@Configuration
public class RabbitConfig2 {
public final static String QUEUE_21 = "queue21";
public final static String QUEUE_22 = "queue22";
public final static String FAN_EXCHANGE_2 = "fanout2";
@Bean
public Queue queue21() {
Queue queue1 = new Queue(QUEUE_21);
return queue1;
}
@Bean
public Queue queue22() {
Queue queue1 = new Queue(QUEUE_22);
return queue1;
}
@Bean
public FanoutExchange FanoutExchange() {
// 最简单的构造函数
return new FanoutExchange(FAN_EXCHANGE_2);
}
@Bean
public Binding binding21(Queue queue21, FanoutExchange FanoutExchange) {
return BindingBuilder
.bind(queue21)
// to后面 没有 withXXX 函数，to的结果即返回 Binding对象
.to(FanoutExchange);
}
@Bean
public Binding binding22(Queue queue22, FanoutExchange FanoutExchange) {
return BindingBuilder
.bind(queue22)
// to后面 没有 withXXX 函数，to的结果即返回 Binding对象
.to(FanoutExchange);
}
}发送接口：
/**
* 发送消息
* @author ben
* @date 2021-08-27 21:24:19 CST
* @return
*/
@GetMapping(value="/send")
public boolean send() {
IntStream.range(0, 10).forEach(i->{
rabbitTemplate.convertAndSend(RabbitConfig2.FAN_EXCHANGE_2, RabbitConfig2.QUEUE_21,
"Try2Controller-Simple msg-" + i + ", @" + new Date());
try {
TimeUnit.SECONDS.sleep(1L);
} catch (InterruptedException e) {
log.info("异常：{}", e.getMessage());
}
});
return true;
}监听器：来自博客园
@Component
@Slf4j
public class Try2Listener {
/**
* Listener1
* @author ben
* @date 2021-08-23 22:31:58 CST
* @param content
*/
@RabbitListener(queues = {RabbitConfig2.QUEUE_21})
public void processMessage1(String content) {
log.info("接收消息：队列{}-processMessage1-content={}",
RabbitConfig2.QUEUE_21, content);
}
/**
* Listener2
* @author ben
* @date 2021-08-27 21:26:30 CST
* @param content
*/
@RabbitListener(queues = {RabbitConfig2.QUEUE_21})
public void processMessage2(String content) {
log.info("接收消息：队列{}-processMessage2-content={}",
RabbitConfig2.QUEUE_21, content);
}
/**
* Listener3
* @author ben
* @date 2021-08-27 21:26:30 CST
* @param content
*/
@RabbitListener(queues = {RabbitConfig2.QUEUE_22})
public void processMessage3(String content) {
log.info("接收消息：队列{}-processMessage3-content={}", RabbitConfig2.QUEUE_22, content);
}
}QUEUE_21 两个监听器，QUEUE_22 一个监听器。
启动后Web管理器展示：
测试结果：
调用接口 /try2/send，
QUEUE_21 两个监听器，均衡地接收并处理了所有消息；来自博客园
QUEUE_22 一个监听器，独自处理了所有消息。
虽然发送时使用了 routingKey 参数，但是，绑定时没有使用 routingKey。来自博客园
FanoutExchange会把收到的消息 发送到所有与其绑定的队列，这就是 实现广播机制的原理吧。
根据主题（routingKey）将消息发送到不同的队列，可以使用通配符。
配置：
@Configuration
public class RabbitConfig3 {
public final static String QUEUE_31 = "queue31";
public final static String QUEUE_32 = "queue32";
public final static String TOPIC_EXCHANGE_3 = "topic3";
@Bean
public Queue queue31() {
return new Queue(QUEUE_31);
}
@Bean
public Queue queue32() {
return new Queue(QUEUE_32);
}
@Bean
public TopicExchange topicExchange() {
// 最简单的构造函数
return new TopicExchange(TOPIC_EXCHANGE_3);
}
@Bean
public Binding binding31(Queue queue31, TopicExchange topicExchange) {
return BindingBuilder
.bind(queue31)
.to(topicExchange)
// routingKey 为 QUEUE_31——队列名
.with(QUEUE_31);
}
@Bean
public Binding binding32(Queue queue32, TopicExchange topicExchange) {
return BindingBuilder
.bind(queue32)
.to(topicExchange)
// routingKey 为 *，通配符，所有消息
.with("*");
}
}测试接口：/try3/send
总计发送 30条，分别使用不同的 routingKey。来自博客园
@GetMapping(value="/send")
public boolean send() {
IntStream.range(0, 10).forEach(i->{
// routingKey 为队列31的名称
rabbitTemplate.convertAndSend(RabbitConfig3.TOPIC_EXCHANGE_3, RabbitConfig3.QUEUE_31,
"Try3Controller-Simple msg 1-" + i + ", @" + new Date());
// routingKey 为队列32的名称
rabbitTemplate.convertAndSend(RabbitConfig3.TOPIC_EXCHANGE_3, RabbitConfig3.QUEUE_32,
"Try3Controller-Simple msg 2-" + i + ", @" + new Date());
// routingKey 为topicXyz
rabbitTemplate.convertAndSend(RabbitConfig3.TOPIC_EXCHANGE_3, "topicXyz",
"Try3Controller-Simple msg 3-" + i + ", @" + new Date());
try {
TimeUnit.SECONDS.sleep(1L);
} catch (InterruptedException e) {
log.info("异常：{}", e.getMessage());
}
});
return true;
}监听器：两个，分别监听 QUEUE_31、QUEUE_32。
@RabbitListener(queues = {RabbitConfig3.QUEUE_31})
public void processMessage31(String content) {
log.info("接收消息：队列{}-processMessage31-content={}",
RabbitConfig3.QUEUE_31, content);
}
@RabbitListener(queues = {RabbitConfig3.QUEUE_32})
public void processMessage32(String content) {
log.info("接收消息：队列{}-processMessage32-content={}", RabbitConfig3.QUEUE_32, content);
}执行结果（部分）：
两个监听器，总计收到 40条，
其中，processMessage31 收到了10条，只是 routingKey 为 队列名 queue31 的，来自博客园
processMessage32 收到30条——所有——因为使用了 通配符 “*”。
疑问：
都是通过 routingKey 转发消息到队列，DirectExchange 和 TopicExchange 的区别是什么呢？
原来，TopicExchange 也叫 通配符交换机，可以使用 通配符 来转发各种消息，而 DirectExchange 则只是根据 routingKey 的字面值 来转发消息到 队列。
待验证，TODO
在前面的试验中，@RabbitListener 接收消息都没有做更多工作，直接就获取了 消息内容。
其实，这时的消息是 自动确认的。
怎么做到 手动确认消息呢？
@RabbitListener有一个属性——ackMode，将其配置为 MANUAL 即开启 此监听器的手动确认功能（参考类 AcknowledgeMode）。
注，当然，应该还有 全局配置的方式，暂不清楚。
本节在 试验3-试验DirectExchange 的基础上继续测试：一个交换机、一个队列、两个RabbitListener 。来自博客园
6-1：配置其中一个为 手动确认
// 手动确认
@RabbitListener(queues = {RabbitConfig.QUEUE_1}, ackMode="MANUAL")
public void processMessage1(String content) {
log.info("接收消息：processMessage1-content={}", content);
}
// 自动确认
@RabbitListener(queues = {RabbitConfig.QUEUE_1})
public void processMessage2(String content) {
log.info("接收消息：processMessage2-content={}", content);
}执行消息发送接口，初看日志，消息被两个监听器 均衡地消费了。
实际上呢？queue1队列里面还剩5个待消费。
关闭应用，此时，应用没有立即停止，而是输出下面的内容：来自博客园
注意，上图的日志不是 唯一的情景，有时候，队列中的消息 还会被 自动确认的 监听器给消费了——非确定性的！
那么，怎么执行手动确认呢？需要改造被 @RabbitListener 注解的函数。
如下：增加了参数 channel（com.rabbitmq.client.Channel），tag——必须使用 @Header！
@RabbitListener(queues = {RabbitConfig.QUEUE_1}, ackMode="MANUAL")
// public void processMessage1(String content) {
public void processMessage1(String content, Channel channel,
// @Header 必须！否则，异常！
@Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
log.info("接收消息：processMessage1-content={}", content);
log.info("手动确认消息：tag={}, channelNumber={}, channel", tag, channel.getChannelNumber(), channel);
// 确认消息
channel.basicAck(tag, false);
}执行结果：
手动确认成功，RabbitMQ的Web管理器中，也没有剩余消息了。来自博客园
6-2：配置所有@RabbitListener 手动确认
可以看到，上面的 channelNumber 等于 1，那么，将两个 @RabbitListener 都配置为 手动确认的时候如何？
如下：
@RabbitListener(queues = {RabbitConfig.QUEUE_1}, ackMode="MANUAL")
// public void processMessage1(String content) {
public void processMessage1(String content, Channel channel,
// @Header 必须！否则，异常！
@Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
log.info("接收消息1：content={}", content);
log.info("手动确认消息1：tag={}, channelNumber={}, channel", tag, channel.getChannelNumber(), channel);
// 确认消息
channel.basicAck(tag, false);
}
@RabbitListener(queues = {RabbitConfig.QUEUE_1})
// public void processMessage2(String content) {
public void processMessage2(String content, Channel channel,
// @Header 必须！否则，异常！
@Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
log.info("接收消息2：content={}", content);
log.info("手动确认消息2：tag={}, channelNumber={}, channel", tag, channel.getChannelNumber(), channel);
// 确认消息
channel.basicAck(tag, false);
}执行结果：
消息接收成功了。
注意，
channelNumber 有 1、2 两个了，对应两个 @RabbitListener；来自博客园
而且，各个@RabbitListener中的 tag值 是按照顺序增长的。
6-3：启动2个应用，共4个 @RabbitListener
Web监控平台可以看到：
执行消息发送。
执行结果：
一个处理了4条、一个处理了6条，总计10条。来自博客园
关闭其中一个应用，再测试：
剩下的一个应用处理了所有消息。
》》》全文完《《《
RabbitMQ还有更多内容可以探究的：
1、消息持久化，默认的交换机、队列都是持久化的——durable=true；什么时候不需要持久化呢？
2、可以用来做延迟消息队列——消息发送了，不立即处理，指定时间到了后再处理，怎么实现呢？
3、更多高级的配置，spring.rabbitmq.*；
4、RabbitMQ服务器优化等；
5、集群架构，看了一篇博文，居然介绍了4种！
6、会用之后，是不是要探究下原理呢？
……
1、RabbitMQ学习笔记：4369、5672、15672、25672默认端口号修改
2、RabbitMQ三种Exchange模式(fanout,direct,topic)简介
4、