1.MQ基础概念
1.1 异步调用概念
MQ是一种异步调用方式，异步调用方式是一种基于消息的通知方式，一般包含三种角色
- 消息发送者：投递消息的人，就是原来的调用方
- 消息Broker：管理、暂存、转发消息，你可以把它理解成微信服务器
- 消息接收者：接收和处理消息的人，就是原来的服务提供方
编辑
在异步调用中，发送者不再直接同步调用接收者的业务接口，而是发送一条消息投递给消息Broker。然后接收者根据自己的需求从消息Broker那里订阅消息。每当发送方发送消息后，接受者都能获取消息并处理。这样，发送消息的人和接收消息的人就完全解耦了。
以余额支付业务为例： 
编辑
除了扣减余额、更新支付流水单状态以外，其它调用逻辑全部取消。而是改为发送一条消息到Broker。而相关的微服务都可以订阅消息通知，一旦消息到达Broker，则会分发给每一个订阅了的微服务，处理各自的业务。
假如产品经理提出了新的需求，比如要在支付成功后更新用户积分。支付代码完全不用变更，而仅仅是让积分服务也订阅消息即可：
编辑
不管后期增加了多少消息订阅者，作为支付服务来讲，执行问扣减余额、更新支付流水状态后，发送消息即可。业务耗时仅仅是这三部分业务耗时，仅仅100ms，大大提高了业务性能。
另外，不管是交易服务、通知服务，还是积分服务，他们的业务与支付关联度低。现在采用了异步调用，解除了耦合，他们即便执行过程中出现了故障，也不会影响到支付服务。
综上，异步调用的优势包括：
- 耦合度更低
- 性能更好
- 业务拓展性强
- 故障隔离，避免级联失败
当然，异步通信也并非完美无缺，它存在下列缺点：
- 完全依赖于Broker的可靠性、安全性和性能
- 架构复杂，后期维护和调试麻烦
1.2 MQ技术选型
消息Broker，目前常见的实现方案就是消息队列（MessageQueue），简称为MQ.
目比较常见的MQ实现：
- ActiveMQ
- RabbitMQ
- RocketMQ
- Kafka
几种常见MQ的对比： 
编辑
追求可用性：Kafka、 RocketMQ 、RabbitMQ
追求可靠性：RabbitMQ、RocketMQ
追求吞吐能力：RocketMQ、Kafka
追求消息低延迟：RabbitMQ、Kafka
据统计，目前国内消息队列使用最多的还是RabbitMQ，再加上其各方面都比较均衡，稳定性较好
2.RabbitMQ安装
基于Docker起RabbitMQ服务：
docker run \ -e RABBITMQ_DEFAULT_USER=lgh \ -e RABBITMQ_DEFAULT_PASS=123456 \ -v mq-plugins:/plugins \ --name mq \ --hostname mq \ -p 15672:15672 \ -p 5672:5672 \ -d \ rabbitmq:3.8-management
可以看到在安装命令中有两个映射的端口：
- 15672：RabbitMQ提供的管理控制台的端口
- 5672：RabbitMQ的消息发送处理接口
访问15672端口进入管理后台：
编辑
3.RabbitMQ基础概念
编辑
其中包含几个概念：
publisher：生产者，也就是发送消息的一方consumer：消费者，也就是消费消息的一方queue：队列，存储消息。生产者投递的消息会暂存在消息队列中，等待消费者处理exchange：交换机，负责消息路由。生产者发送的消息由交换机决定投递到哪个队列。virtual host：虚拟主机，起到数据隔离的作用。每个虚拟主机相互独立，有各自的exchange、queue
上述这些东西都可以在RabbitMQ的管理控制台来管理。
3.1 收发消息
3.1.1 交换机
我们打开Exchanges选项卡，可以看到已经存在很多交换机：
编辑
点击任意交换机，即可进入交换机详情页面。仍然会利用控制台中的publish message 发送一条消息：
编辑
这里是由控制台模拟了生产者发送的消息。由于没有消费者存在，最终消息丢失了，这样说明交换机没有存储消息的能力。
3.1.2 队列
打开Queues选项卡，新建两个个队列
编辑
发送到交换机的消息，只会路由到与其绑定的队列，因此仅仅创建队列是不够的，我们还需要将其与交换机绑定。
点击Exchanges选项卡，点击amq.fanout交换机，进入交换机详情页，然后点击Bindings菜单，在表单中填写要绑定的队列名称：
编辑
发送消息：
编辑
回到Queues页面， 发现队列中已接受消息 
编辑
编辑
这个时候如果有消费者监听了MQ队列，自然就能接收到消息了。
3.2 数据隔离
3.2.1 用户管理
点击Admin选项卡，首先会看到RabbitMQ控制台的用户管理界面：
编辑
这里的用户都是RabbitMQ的管理或运维人员。目前只有安装RabbitMQ时添加的`itheima`这个用户。仔细观察用户表格中的字段，如下：
- Name：lgh，也就是用户名
- Tags：administrator，说明lgh用户是超级管理员，拥有所有权限
- Can access virtual host： /，可以访问的virtual host，这里的`/`是默认的`virtual host`
对于小型企业而言，出于成本考虑，我们通常只会搭建一套MQ集群，公司内的多个不同项目同时使用。这个时候为了避免互相干扰， 我们会利用`virtual host`的隔离特性，将不同项目隔离。一般会做两件事情：
- 给每个项目创建独立的运维账号，将管理权限分离。
- 给每个项目创建不同的`virtual host`，将每个项目的数据隔离。
例如，我们给当前项目创建用户，命名test1
编辑
此时test1用户没有任何`virtual host`的访问权限：
3.2.2 Virtual Host
将当前登录账号切换回test1,然后点击Virtual Hosts菜单，进入virtual host管理页：
可以看到目前只有一个默认的virtual host，名字为 /，可以给我们的项目创建一个单独的`virtual host`，而不是使用默认的`/`。
编辑
编辑
切换virtual host空间后，再次查看发现之前创建的队列在当前账号下已经不可见了，这就是virtual host的隔离效果。
4.SpringAMQP
将来我们开发业务功能的时候，肯定不会在控制台收发消息，而是应该基于编程的方式。由于RabbitMQ采用了AMQP协议，因此它具备跨语言的特性。任何语言只要遵循AMQP协议收发消息，都可以与RabbitMQ交互。并且RabbitMQ官方也提供了各种不同语言的客户端。 但是，RabbitMQ官方提供的Java客户端编码相对复杂，一般生产环境下我们更多会结合Spring来使用。而Spring的官方刚好基于RabbitMQ提供了这样一套消息收发的模板工具：SpringAMQP。并且还基SpringBoot对其实现了自动装配，使用起来非常方便。
SpringAmqp的官方地址： Spring AMQP SpringAMQP提供了三个功能：
- 自动声明队列、交换机及其绑定关系
- 基于注解的监听器模式，异步接收消息
- 封装了RabbitTemplate工具，用于发送消息
4.1 入门案例：
在之前的案例中，我们都是经过交换机发送消息到队列，不过有时候为了测试方便，我们也可以直接向队列发送消息，跳过交换机。在入门案例中，我们就演示这样的简单模型。
引入AMQP依赖：
<!--AMQP依赖，包含RabbitMQ--> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency>
创建队列：
编辑
在appilication.yml中配置MQ：
logging: pattern: dateformat: MM-dd HH:mm:ss:SSS spring: rabbitmq: host: port: 5672 virtual-host: / username: lgh password: listener: simple: prefetch: 1
编写测试类SpringAmqpTest，并利用RabbitTemplate实现消息发送：
@SpringBootTest public class SpringAmqpTest { @Autowired private RabbitTemplate rabbitTemplate; @Test void testSendMessage2Queue() { String queueName = "t1.q"; String msg = "hello, amqp!"; rabbitTemplate.convertAndSend(queueName, msg); } }
在控制台看到队列成功接收到消息：
编辑
监听消息测试：
@Component public class SpringRabbitListener { // 利用RabbitListener来声明要监听的队列信息 // 将来一旦监听的队列中有了消息，就会推送给当前服务，调用当前方法，处理消息。 // 可以看到方法体中接收的就是消息体的内容 @RabbitListener(queues = "t1.q") public void listenSimpleQueueMessage(String msg) throws InterruptedException { System.out.println("spring 消费者接收到消息：【" + msg + "】"); } }
4.2 WorkQueues模型
Work queues，任务模型。简单来说就是让多个消费者绑定到一个队列，共同消费队列中的消息。 
编辑
当消息处理比较耗时的时候，可能生产消息的速度会远远大于消息的消费速度。长此以往，消息就会堆积越来越多，无法及时处理。 此时就可以使用work 模型，多个消费者共同处理消息处理，消息处理的速度就能大大提高了。接下来，我们就来模拟这样的场景。 首先，我们在控制台创建一个新的队列，命名为work.queue：
编辑
发送方测试：
/** * workQueue * 向队列中不停发送消息，模拟消息堆积。 */ @Test public void testWorkQueue() throws InterruptedException { // 队列名称 String queueName = "simple.queue"; // 消息 String message = "hello, message_"; for (int i = 0; i < 50; i++) { // 发送消息，每20毫秒发送一次，相当于每秒发送50条消息 rabbitTemplate.convertAndSend(queueName, message + i); Thread.sleep(20); } }
接收方测试：
要模拟多个消费者绑定同一个队列，我们在SpringRabbitListener中添加2个新的方法：
@RabbitListener(queues = "work.queue") public void listenWorkQueue1(String msg) throws InterruptedException { System.out.println("消费者1接收到消息：【" + msg + "】" + LocalTime.now()); Thread.sleep(20); } @RabbitListener(queues = "work.queue") public void listenWorkQueue2(String msg) throws InterruptedException { System.err.println("消费者2........接收到消息：【" + msg + "】" + LocalTime.now()); Thread.sleep(200); }
注意到这两消费者，都设置了Thead.sleep，模拟任务耗时：
- 消费者1 sleep了20毫秒，相当于每秒钟处理50个消息
- 消费者2 sleep了200毫秒，相当于每秒处理5个消息
可以从执行结果中看到消费者1和消费者2竟然每人消费了25条消息：
- 消费者1很快完成了自己的25条消息。
- 消费者2却在缓慢的处理自己的25条消息。
也就是说消息是平均分配给每个消费者，并没有考虑到消费者的处理能力。导致1个消费者空闲，另一个消费者忙的不可开交。没有充分利用每一个消费者的能力，最终消息处理的耗时远远超过了1秒。这样显然是有问题的。
在spring中有一个简单的配置，可以解决这个问题。我们修改consumer服务的application.yml文件，添加配置：
spring: rabbitmq: listener: simple: prefetch: 1 # 每次只能获取一条消息，处理完成才能获取下一个消息
可以发现，由于消费者1处理速度较快，所以处理了更多的消息；消费者2处理速度较慢，只处理了6条消息。而最终总的执行耗时也在1秒左右，大大提升。 正所谓能者多劳，这样充分利用了每一个消费者的处理能力，可以有效避免消息积压问题。
Work模型的使用：
- 多个消费者绑定到一个队列，同一条消息只会被一个消费者处理
- 通过设置prefetch来控制消费者预取的消息数量
4.3 交换机类型
编辑
可以看到，在订阅模型中，多了一个exchange角色，而且过程略有变化：
- Publisher：生产者，不再发送消息到队列中，而是发给交换机
- Exchange：交换机，一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。
- Queue：消息队列也与以前一样，接收消息、缓存消息。不过队列一定要与交换机绑定。
- Consumer：消费者，与以前一样，订阅队列，没有变化
Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！
交换机的类型有四种：
- Fanout：广播，将消息交给所有绑定到交换机的队列。我们最早在控制台使用的正是Fanout交换机
- Direct：订阅，基于RoutingKey（路由key）发送给订阅了消息的队列
- Topic：通配符订阅，与Direct类似，只不过RoutingKey可以使用通配符
- Headers：头匹配，基于MQ的消息头匹配，用的较少。
4.4 Fanout交换机
在广播模式下，消息发送流程是这样的：
编辑
- 1） 可以有多个队列
- 2） 每个队列都要绑定到Exchange（交换机）
- 3） 生产者发送的消息，只能发送到交换机
- 4） 交换机把消息发送给绑定过的所有队列
- 5） 订阅队列的消费者都能拿到消息
我们的计划是这样的：
编辑
- 创建一个名为
hmall.fanout的交换机，类型是Fanout - 创建两个队列
fanout.queue1和fanout.queue2，绑定到交换机hmall.fanout
在控制台创建队列fanout.queue1:
编辑
在创建一个队列fanout.queue2：
编辑
然后再创建一个交换机：
编辑
然后绑定两个队列到交换机：
编辑
编辑
消息发送：
@Test public void testFanoutExchange() { // 交换机名称 String exchangeName = "hmall.fanout"; // 消息 String message = "hello, everyone!"; rabbitTemplate.convertAndSend(exchangeName, "", message); }
消息接收：
@RabbitListener(queues = "fanout.queue1") public void listenFanoutQueue1(String msg) { System.out.println("消费者1接收到Fanout消息：【" + msg + "】"); } @RabbitListener(queues = "fanout.queue2") public void listenFanoutQueue2(String msg) { System.out.println("消费者2接收到Fanout消息：【" + msg + "】"); }
交换机的作用是什么？
- 接收publisher发送的消息
- 将消息按照规则路由到与之绑定的队列
- 不能缓存消息，路由失败，消息丢失
- FanoutExchange的会将消息路由到每个绑定的队列
4.5 Direct交换机
在Fanout模式中，一条消息，会被所有订阅的队列都消费。但是，在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。
编辑
在Direct模型下：
- 队列与交换机的绑定，不能是任意绑定了，而是要指定一个
RoutingKey（路由key） - 消息的发送方在 向 Exchange发送消息时，也必须指定消息的
RoutingKey。 - Exchange不再把消息交给每一个绑定的队列，而是根据消息的
Routing Key进行判断，只有队列的Routingkey与消息的Routing key完全一致，才会接收到消息
案例需求如图：
编辑
- 声明一个名为
hmall.direct的交换机 - 声明队列
direct.queue1，绑定hmall.direct，bindingKey为blud和red - 声明队列
direct.queue2，绑定hmall.direct，bindingKey为yellow和red - 在
consumer服务中，编写两个消费者方法，分别监听direct.queue1和direct.queue2 - 在publisher中编写测试方法，向
hmall.direct发送消息
声明队列和交换机：
首先在控制台声明两个队列direct.queue1和direct.queue2，这里不再展示过程：
编辑
然后声明一个direct类型的交换机，命名为hmall.direct:
编辑
然后使用red和blue作为key，绑定direct.queue1到hmall.direct：
编辑
编辑
同理，使用red和yellow作为key，绑定direct.queue2到hmall.direct，步骤略，最终结果：
编辑
消息接收：
@RabbitListener(queues = "direct.queue1") public void listenDirectQueue1(String msg) { System.out.println("消费者1接收到direct.queue1的消息：【" + msg + "】"); } @RabbitListener(queues = "direct.queue2") public void listenDirectQueue2(String msg) { System.out.println("消费者2接收到direct.queue2的消息：【" + msg + "】"); }
消息发送：
@Test public void testSendDirectExchange() { // 交换机名称 String exchangeName = "hmall.direct"; // 消息 String message = "红色警报！日本乱排核废水，导致海洋生物变异，惊现哥斯拉！"; // 发送消息 rabbitTemplate.convertAndSend(exchangeName, "red", message); }
由于使用的red这个key，所以两个消费者都收到了消息。
描述下Direct交换机与Fanout交换机的差异？
- Fanout交换机将消息路由给每一个与之绑定的队列
- Direct交换机根据RoutingKey判断路由给哪个队列
- 如果多个队列具有相同的RoutingKey，则与Fanout功能类似
4.6 Topic交换机
Topic类型的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。 只不过Topic类型Exchange可以让队列在绑定BindingKey 的时候使用通配符！
BindingKey 一般都是有一个或多个单词组成，多个单词之间以.分割，例如： item.insert
通配符规则：
#：匹配一个或多个词*：匹配不多不少恰好1个词
举例：
item.#：能够匹配item.spu.insert或者item.spuitem.*：只能匹配item.spu
图示：
编辑
假如此时publisher发送的消息使用的RoutingKey共有四种：
china.news代表有中国的新闻消息；china.weather代表中国的天气消息；japan.news则代表日本新闻japan.weather代表日本的天气消息；
解释：
topic.queue1：绑定的是china.#，凡是以china.开头的routing key都会被匹配到，包括：
china.newschina.weather
topic.queue2：绑定的是#.news，凡是以.news结尾的routing key都会被匹配。包括:
china.newsjapan.news
接下来，我们就按照上图所示，来演示一下Topic交换机的用法。
首先，在控制台按照图示例子创建队列、交换机，并利用通配符绑定队列和交换机。此处步骤略。最终结果如下： 
编辑
消息发送：
/** * topicExchange */ @Test public void testSendTopicExchange() { // 交换机名称 String exchangeName = "hmall.topic"; // 消息 String message = "喜报！孙悟空大战哥斯拉，胜!"; // 发送消息 rabbitTemplate.convertAndSend(exchangeName, "china.news", message); }
消息接收：
@RabbitListener(queues = "topic.queue1") public void listenTopicQueue1(String msg){ System.out.println("消费者1接收到topic.queue1的消息：【" + msg + "】"); } @RabbitListener(queues = "topic.queue2") public void listenTopicQueue2(String msg){ System.out.println("消费者2接收到topic.queue2的消息：【" + msg + "】"); }
描述下Direct交换机与Topic交换机的差异？
- Topic交换机接收的消息RoutingKey必须是多个单词，以
**.**分割 - Topic交换机与队列绑定时的bindingKey可以指定通配符
#：代表0个或多个词*：代表1个词
4.7 声明交换机与队列
在之前我们都是基于RabbitMQ控制台来创建队列、交换机。但是在实际开发时，队列和交换机是程序员定义的，将来项目上线，又要交给运维去创建。那么程序员就需要把程序中运行的所有队列和交换机都写下来，交给运维。在这个过程中是很容易出现错误的。 因此推荐的做法是由程序启动时检查队列和交换机是否存在，如果不存在自动创建。
4.7.1 基本API
SpringAMQP提供了一个Queue类，用来创建队列：
编辑
SpringAMQP还提供了一个Exchange接口，来表示所有不同类型的交换机：
编辑
我们可以自己创建队列和交换机，不过SpringAMQP还提供了ExchangeBuilder来简化这个过程：
编辑
而在绑定队列和交换机时，则需要使用BindingBuilder来创建Binding对象：
编辑
4.7.2 Fanout示例
创建一个类，声明队列和交换机：
import org.springframework.amqp.core.Binding; import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.FanoutExchange; import org.springframework.amqp.core.Queue; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @Configuration public class FanoutConfig { /** * 声明交换机 * @return Fanout类型交换机 */ @Bean public FanoutExchange fanoutExchange(){ return new FanoutExchange("hmall.fanout"); } /** * 第1个队列 */ @Bean public Queue fanoutQueue1(){ return new Queue("fanout.queue1"); } /** * 绑定队列和交换机 */ @Bean public Binding bindingQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange){ return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange); } /** * 第2个队列 */ @Bean public Queue fanoutQueue2(){ return new Queue("fanout.queue2"); } /** * 绑定队列和交换机 */ @Bean public Binding bindingQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange){ return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange); } }
4.7.3 direct示例
import org.springframework.amqp.core.*; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @Configuration public class DirectConfig { /** * 声明交换机 * @return Direct类型交换机 */ @Bean public DirectExchange directExchange(){ return ExchangeBuilder.directExchange("hmall.direct").build(); } /** * 第1个队列 */ @Bean public Queue directQueue1(){ return new Queue("direct.queue1"); } /** * 绑定队列和交换机 */ @Bean public Binding bindingQueue1WithRed(Queue directQueue1, DirectExchange directExchange){ return BindingBuilder.bind(directQueue1).to(directExchange).with("red"); } /** * 绑定队列和交换机 */ @Bean public Binding bindingQueue1WithBlue(Queue directQueue1, DirectExchange directExchange){ return BindingBuilder.bind(directQueue1).to(directExchange).with("blue"); } /** * 第2个队列 */ @Bean public Queue directQueue2(){ return new Queue("direct.queue2"); } /** * 绑定队列和交换机 */ @Bean public Binding bindingQueue2WithRed(Queue directQueue2, DirectExchange directExchange){ return BindingBuilder.bind(directQueue2).to(directExchange).with("red"); } /** * 绑定队列和交换机 */ @Bean public Binding bindingQueue2WithYellow(Queue directQueue2, DirectExchange directExchange){ return BindingBuilder.bind(directQueue2).to(directExchange).with("yellow"); } }
4.7.4 基于注解声明
基于@Bean的方式声明队列和交换机比较麻烦，Spring还提供了基于注解方式来声明。
例如，我们同样声明Direct模式的交换机和队列：
@Slf4j @Component public class MqListener { @RabbitListener(queues = "simple.queue") public void listenSimpleQueue(String msg){ System.out.println("消费者收到了simple.queue的消息：【" + msg +"】"); } @RabbitListener(queues = "work.queue") public void listenWorkQueue1(String msg) throws InterruptedException { System.out.println("消费者1 收到了 work.queue的消息：【" + msg +"】"); Thread.sleep(20); } @RabbitListener(queues = "work.queue") public void listenWorkQueue2(String msg) throws InterruptedException { System.err.println("消费者2 收到了 work.queue的消息...... ：【" + msg +"】"); Thread.sleep(200); } @RabbitListener(queues = "fanout.queue1") public void listenFanoutQueue1(String msg) throws InterruptedException { System.out.println("消费者1 收到了 fanout.queue1的消息：【" + msg +"】"); } @RabbitListener(queues = "fanout.queue2") public void listenFanoutQueue2(String msg) throws InterruptedException { System.out.println("消费者2 收到了 fanout.queue2的消息：【" + msg +"】"); } @RabbitListener(bindings = @QueueBinding( value = @Queue(name = "direct.queue1", durable = "true"), exchange = @Exchange(name = "hmall.direct", type = ExchangeTypes.DIRECT), key = {"red", "blue"} )) public void listenDirectQueue1(String msg) throws InterruptedException { System.out.println("消费者1 收到了 direct.queue1的消息：【" + msg +"】"); } @RabbitListener(bindings = @QueueBinding( value = @Queue(name = "direct.queue2", durable = "true"), exchange = @Exchange(name = "hmall.direct", type = ExchangeTypes.DIRECT), key = {"red", "yellow"} )) public void listenDirectQueue2(String msg) throws InterruptedException { System.out.println("消费者2 收到了 direct.queue2的消息：【" + msg +"】"); } @RabbitListener(queues = "topic.queue1") public void listenTopicQueue1(String msg) throws InterruptedException { System.out.println("消费者1 收到了 topic.queue1的消息：【" + msg +"】"); } @RabbitListener(queues = "topic.queue2") public void listenTopicQueue2(String msg) throws InterruptedException { System.out.println("消费者2 收到了 topic.queue2的消息：【" + msg +"】"); } @RabbitListener(queues = "object.queue") public void listenObject(Map<String, Object> msg) throws InterruptedException { System.out.println("消费者 收到了 object.queue的消息：【" + msg +"】"); } }
第二种写法：
@Component @RabbitListener(queues = "TestDirectQueue") // 监听的队列名称 TestDirectQueue public class RabbitMessageListener { @RabbitHandler public void process(String message) { System.out.println("DirectReceiver m消费者收到消息 : " + message); } }
4.8 消息转换器
Spring的消息发送代码接收的消息体是一个Object：
编辑
当使用 Spring 发送消息时，传输内容通常是 Java 对象，但消息队列只能处理字节数据。因此，需要先将对象序列化为字节发送，接收时再反序列化为对象。Spring 默认使用的是 JDK 的序列化机制，但这种方式格式不友好、体积大、不安全。实际开发中，通常会配置 JSON 转换器，使消息更轻量、可读且更安全。
引入依赖：
<dependency> <groupId>com.fasterxml.jackson.dataformat</groupId> <artifactId>jackson-dataformat-xml</artifactId> <version>2.9.10</version> </dependency>
配置消息转换器：
@Bean public MessageConverter messageConverter(){ // 1.定义消息转换器 Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter(); // 2.配置自动创建消息id，用于识别不同消息，也可以在业务中基于ID判断是否是重复消息 jackson2JsonMessageConverter.setCreateMessageIds(true); return jackson2JsonMessageConverter; }
消息转换器中添加的messageId可以便于我们将来做幂等性判断。
Springboot会将jackson2JsonMessageConverter这个bean对象自动注入到RabbitTemRabbitTemplate中。
消息发送接收接收：
我们在consumer服务中定义一个新的消费者，publisher是用Map发送，那么消费者也一定要用Map接收，格式如下：
@Test public void testSendMap() throws InterruptedException { // 准备消息 Map<String,Object> msg = new HashMap<>(); msg.put("name", "柳岩"); msg.put("age", 21); // 发送消息 rabbitTemplate.convertAndSend("object.queue", msg); }
@RabbitListener(queues = "object.queue") public void listenSimpleQueueMessage(Map<String, Object> msg) throws InterruptedException { System.out.println("消费者接收到object.queue消息：【" + msg + "】"); }
4.9 调用实例
@RestController @RequestMapping("/mq") public class SendMessageController { @Autowired private RabbitTemplate rabbitTemplate; @GetMapping("/sendDirectMessage") public String sendDirectMessage() { // 发送String类型消息到指定交换机和路由键 String message = "你好，这是一条字符串消息"; rabbitTemplate.convertAndSend("TestDirectExchange", "TestDirectRouting", message); return "ok"; } }
