SpringCloud MQ介绍与使用

古城微笑少年丶 2024-05-23 18:58 41阅读 0赞

![168c1abc67f4402ebfc99ff077a3e4da.gif][]

> 哈喽~大家好，这篇来看看SpringCloud MQ介绍与使用。
> 
> ?个人主页：[个人主页][Link 1]
> 
> ? 系列专栏：[【微服务】][Link 2]
> 
> ?与这篇相关的文章：
> 
> <table> 
>  <tbody> 
>   <tr> 
>    <td>SpringCloud Sentinel 使用</td> 
>    <td><a href="https://blog.csdn.net/aasd23/article/details/128032516?spm=1001.2014.3001.5501" title="SpringCloud Sentinel 使用_程序猿追的博客-CSDN博客_springcloud使用sentinel" rel="nofollow">SpringCloud Sentinel 使用_程序猿追的博客-CSDN博客_springcloud使用sentinel</a></td> 
>   </tr> 
>   <tr> 
>    <td>将Nacos注册到springboot使用以及Feign实现服务调用</td> 
>    <td><a href="https://blog.csdn.net/aasd23/article/details/127961919?spm=1001.2014.3001.5502" title="将Nacos注册到springboot使用以及Feign实现服务调用_程序猿追的博客-CSDN博客_nacos springboot 服务调用" rel="nofollow">将Nacos注册到springboot使用以及Feign实现服务调用_程序猿追的博客-CSDN博客_nacos springboot 服务调用</a></td> 
>   </tr> 
>   <tr> 
>    <td>微服务介绍与 SpringCloud Eureka</td> 
>    <td><a href="https://blog.csdn.net/aasd23/article/details/127913922?spm=1001.2014.3001.5502" title="微服务介绍与 SpringCloud Eureka_程序猿追的博客-CSDN博客" rel="nofollow">微服务介绍与 SpringCloud Eureka_程序猿追的博客-CSDN博客</a></td> 
>   </tr> 
>  </tbody> 
> </table>

**目录**

一、前言

1、什么是MQ？

2、为什么要用 MQ

2.1、流量消峰

2.2、应用解耦

2.3、异步处理

3、同步和异步通讯

4、异步调用则可以避免上述问题

2、运行暂停的容器

3、MQ的基本结构

2、consumer实现

3、效果

四、SpringAMQP

1、AMQP

2、Spring AMQP

五、Basic Queue 简单队列模型

1、父工程导入依赖

2、消息接收

六、WorkQueue

1、消息发送

2、消息接收

3、测试

4、总结

七、发布/订阅

1、Fanout

2、Direct

2.1、基于注解声明队列和交换机

2.2、消息发送

5、Topic

5.1、消息发送

5.2、消息接收

5.3、消息转换器

5.4、测试默认转换器

5.5、配置JSON转换器

--------------------

## 一、前言 ##

### 1、什么是MQ？ ###

> **MQ 全称为Message Queue，即消息队列。**“消息队列”是在消息的传输过程中保存消息的容器。它是典型的：生产者、消费者模型。生产者不断向消息队列中生产消息，消费者不断的从队列中获取消息。因为消息的生产和消费都是异步的，而且只关心消息的发送和接收，没有业务逻辑的侵入，这样就实现了生产者和消费者的解耦。

### 2、为什么要用 MQ ###

#### 2.1、流量消峰 ####

> 简单来说： 就是在访问量剧增的情况下，但是应用仍然不能停，比如“双十一”下单的人多，但是淘宝这个应用仍然要运行，所以就可以使用消息中间件采用队列的形式减少突然访问的压力

#### 2.2、应用解耦 ####

> 以电商应用为例，应用中有**订单系统、库存系统、物流系统、支付系统**。用户创建订单后，如果耦合调用库存系统、物流系统、支付系统，任何一个子系统出了故障，都会造成下单操作异常。当转变成基于消息队列的方式后，系统间调用的问题会减少很多，比如物流系统因为发生故障，需要几分钟来修复。在这几分钟的时间里，物流系统要处理的内存被缓存在消息队列中，用户的下单操作可以正常完成。当物流系统恢复后，继续处理订单信息即可，中间用户感受不到物流系统的故障，提升系统的可用性。

#### 2.3、异步处理 ####

> 有些服务间调用是异步的，例如 A 调用 B，B 需要花费很长时间执行，但是 A 需要知道 B 什么时候可以执行完，以前一般有两种方式，A 过一段时间去调用 B 的查询 api 查询。或者 A 提供一个 callback api， B 执行完之后调用 api 通知 A 服务。 这两种方式都不是很优雅，使用消息总线，可以很方便解决这个问题，A 调用 B 服务后，只需要监听 B 处理完成的消息，当 B 处理完成后，会发送一条消息给 MQ，MQ 会将此消息转发给 A 服务。这样 A 服务既不用循环调用 B 的查询 api，也不用提供 callback api。同样 B 服务也不用做这些操作。A 服务还能及时的得到异步处理成功的消息。

### 3、同步和异步通讯 ###

> **微服务间通讯有同步和异步两种方式：**
> 
> **同步通讯：**就像打电话，需要实时响应。
> 
> **异步通讯：**就像发邮件，不需要马上回复。
> 
> 两种方式各有优劣，打电话可以立即得到响应，但是你却不能跟多个人同时通话。发送邮件可以同时与多个人收发邮件，但是往往响应会有延迟。
> 
> 之前学习的Feign调用就属于同步方式，虽然调用可以实时得到结果，但存在问题，例如：
> 
> **耦合度高：**每次加入新的需求，都要修改原来的代码
> 
> **性能下降：**调用者需要等待服务提供者响应，如果调用链过长则响应时间等于每次调用的时间之和。
> 
> **资源浪费：**调用链中的每个服务在等待响应过程中，不能释放请求占用的资源，高并发场景下会极度浪费系统资源
> 
> **级联失败：**如果服务提供者出现问题，所有调用方都会跟着出问题，如同多米诺骨牌一样，迅速导致整个微服务群故障

### 4、异步调用则可以避免上述问题 ###

> 我们以购买商品为例，用户支付后需要调用订单服务完成订单状态修改，调用物流服务，从仓库分配响应的库存并准备发货。
> 
> 在事件模式中，支付服务是事件**发布者（publisher）**，在支付完成后只需要发布一个支付成功的事件（event），事件中带上订单id。
> 
> 订单服务和物流服务是事件**订阅者（Consumer）**，订阅支付成功的事件，监听到事件后完成自己业务即可。

> 为了解除事件发布者与订阅者之间的耦合，两者并不是直接通信，而是有一个中间人（Broker）。发布者发布事件到Broker，不关心谁来订阅事件。订阅者从Broker订阅事件，不关心谁发来的消息。

![7d3d8653bf6c4fa3b3026ce900cb9590.png][]

**Broker** 是一个像数据总线一样的东西，所有的服务要接收数据和发送数据都发到这个总线上，这个总线就像协议一样，让服务间的通讯变得标准和可控。

**好处**

> **吞吐量提升：**无需等待订阅者处理完成，响应更快速
> 
> **故障隔离：**服务没有直接调用，不存在级联失败问题 ，调用间没有阻塞，不会造成无效的资源占用 ，耦合度极低，每个服务都可以灵活插拔，可替换
> 
> **流量削峰：**不管发布事件的流量波动多大，都由Broker接收，订阅者可以按照自己的速度去处理事件
> 
> **缺点：** 架构复杂了，业务没有明显的流程线，不好管理 ，需要依赖于Broker的可靠、安全、性能

**二、安装RabbitMQ**

**1、在线拉取**

docker pull rabbitmq:3-management

上传到虚拟机中后，使用命令加载镜像即可：

docker load -i mq.tar

执行下面的命令来运行MQ容器：

docker run \
     -e RABBITMQ_DEFAULT_USER=itcast \
     -e RABBITMQ_DEFAULT_PASS=123321 \
     --name mq \
     --hostname mq1 \
     -p 15672:15672 \
     -p 5672:5672 \
     -d \
     rabbitmq:3-management

### 2、运行暂停的容器 ###

docker start XXX # xxx就是id

使用ip + 端口号开运行 rabbitmq:3-management

![088ddc68803a45978e0c103f94a8f8c8.png][]

### 3、MQ的基本结构 ###

![40dee56b8d3941dea2485f2a05687752.png][]

> RabbitMQ中的一些角色名词：
> 
> **publisher：**生产者
> 
> **consumer：**消费者
> 
> **exchange：**交换机，负责消息路由
> 
> **queue：**队列，存储消息
> 
> **virtualHost：**虚拟主机，隔离不同租户的exchange、queue、消息的隔离

**三、导入Demo工程**

结构如下

![8e9eecfabcc2488695917acc164ea9df.png][]

包括三部分

> **mq-demo：**父工程，管理项目依赖
> 
> **publisher：**消息的发送者
> 
> **consumer：**消息的消费者

官方的HelloWorld是基于最基础的消息队列模型来实现的，只包括三个角色：

> **publisher：**消息发布者，将消息发送到队列queue
> 
> **queue：**消息队列，负责接受并缓存消息
> 
> **consumer：**订阅队列，处理队列中的消息

**1、publisher 代码实现**

package cn.itcast.mq.helloworld;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    import org.junit.Test;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    public class PublisherTest {
        @Test
        public void testSendMessage() throws IOException, TimeoutException {
            // 1.建立连接
            ConnectionFactory factory = new ConnectionFactory();
            // 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码
            factory.setHost("192.168.71.132");  // 这是虚拟机的ip，这里不同
            factory.setPort(5672);
            factory.setVirtualHost("/");
            factory.setUsername("itcast");
            factory.setPassword("123456");
            // 1.2.建立连接
            Connection connection = factory.newConnection();
    
            // 2.创建通道Channel
            Channel channel = connection.createChannel();
    
            // 3.创建队列
            String queueName = "simple.queue";
            channel.queueDeclare(queueName, false, false, false, null);
    
            // 4.发送消息
            String message = "hello, rabbitmq!";
            channel.basicPublish("", queueName, null, message.getBytes());
            System.out.println("发送消息成功：【" + message + "】");
    
            // 5.关闭通道和连接
            channel.close();
            connection.close();
    
        }
    }

### 2、consumer实现 ###

package cn.itcast.mq.helloworld;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    public class ConsumerTest {
    
        public static void main(String[] args) throws IOException, TimeoutException {
            // 1.建立连接
            ConnectionFactory factory = new ConnectionFactory();
            // 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码
            factory.setHost("192.168.71.132");
            factory.setPort(5672);
            factory.setVirtualHost("/");
            factory.setUsername("itcast");
            factory.setPassword("123456");
            // 1.2.建立连接
            Connection connection = factory.newConnection();
    
            // 2.创建通道Channel
            Channel channel = connection.createChannel();
    
            // 3.创建队列
            String queueName = "simple.queue";
            channel.queueDeclare(queueName, false, false, false, null);
    
            // 4.订阅消息
            channel.basicConsume(queueName, true, new DefaultConsumer(channel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope,
                                           AMQP.BasicProperties properties, byte[] body) throws IOException {
                    // 5.处理消息
                    String message = new String(body);
                    System.out.println("接收到消息：【" + message + "】");
                }
            });
            System.out.println("等待接收消息。。。。");
        }
    }

**先运行consumer，后运行publisher**

### 3、效果 ###

![85bb9e4906c3477fb0f5b46003a85aa8.png][]

## 四、SpringAMQP ##

> SpringAMQP是基于RabbitMQ封装的一套模板，并且还利用SpringBoot对其实现了自动装配，使用起来非常方便。

SpringAmqp的官方地址：[Spring AMQP][]

### 1、AMQP ###

> Advanced Message Queuing ProtocoL, 是用于在应用程序之间传递业务消息的开放标准。该协议与语言和平台无关，更符合微服务中独立性的要求。

### 2、Spring AMQP ###

> Spring AMQP是基于AMQP协议定义的一套API规范，提供了模板来发送和接收消息。包含两部分，其中spring- amqp是基础抽象，spring- rabbit是底层的默认实现。

SpringAMQP提供了三个功能：

> 自动声明队列、交换机及其绑定关系
> 
> 基于注解的监听器模式，异步接收消息
> 
> 封装了RabbitTemplate工具，用于发送消息

## 五、Basic Queue 简单队列模型 ##

### 1、父工程导入依赖 ###

<dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>

publisher服务的application.yml中添加配置

logging:
      pattern:
        dateformat: MM-dd HH:mm:ss:SSS
    spring:
      rabbitmq:
        host: 192.168.71.132
        port: 5672
        username: itcast
        password: 123456
        virtual-host: /

publisher服务中编写测试类SpringAmqpTest，并利用RabbitTemplate实现消息发送：

@Autowired
    private RabbitTemplate rabbitTemplate;
    
    //    SimpleQueue 测试
    @Test
    public void test1() {
        String queueName = "simple.queue";
        String message = "hello, spring amqp?";
        rabbitTemplate.convertAndSend(queueName, message);
    }

### 2、消息接收 ###

首先配置MQ地址，在consumer服务的application.yml中添加配置：

logging:
      pattern:
        dateformat: MM-dd HH:mm:ss:SSS
    spring:
      rabbitmq:
        host: 192.168.71.132
        port: 5672
        username: itcast
        password: 123456
        virtual-host: /

然后在consumer`服务`的中新建一个类SpringRabbitListener，代码如下：

public class SpringRabbitListener {
    
        @RabbitListener(queues = "simple.queue")
        public void listenSimpleQueueMessage(String msg) throws InterruptedException {
            System.out.println("spring 消费者接收到消息：【" + msg + "】");
        }
    }

效果差不过，他会在RabbitMQ队列里面注册一个叫simple.queue，然后consumer从里面拿出来（拿出来了，就删除了）

![4594c56779964bebaf8108f86993f00c.png][]

## 六、WorkQueue ##

Work queues，也被称为（Task queues），任务模型。简单来说就是**让多个消费者绑定到一个队列，共同消费队列中的消息**。

![b8291eb81c384c11ab79beb404dc0bb4.png][]

> 当消息处理比较耗时的时候，可能生产消息的速度会远远大于消息的消费速度。长此以往，消息就会堆积越来越多，无法及时处理。
> 
> 此时就可以使用work 模型，多个消费者共同处理消息处理，速度就能大大提高了。

### 1、消息发送 ###

**这次我们循环发送，模拟大量消息堆积现象。**

在publisher服务中的SpringAmqpTest类中添加一个测试方法：

/**
         * workQueue
         * 向队列中不停发送消息，模拟消息堆积。
         */
    @Test
    public void testWorkQueue() throws InterruptedException {
        // 队列名称
        String queueName = "simple.queue";
        // 消息
        String message = "hello, message_";
        for (int i = 0; i < 50; i++) {
            // 发送消息
            rabbitTemplate.convertAndSend(queueName, message + i);
            Thread.sleep(20);
        }
    }

### 2、消息接收 ###

要模拟多个消费者绑定同一个队列，我们在consumer服务的SpringRabbitListener中添加2个新的方法：

@RabbitListener(queues = "simple.queue")
    public void listenWorkQueue1(String msg) throws InterruptedException {
        System.out.println("消费者1接收到消息：【" + msg + "】" + LocalTime.now());
        Thread.sleep(20);
    }
    
    @RabbitListener(queues = "simple.queue")
    public void listenWorkQueue2(String msg) throws InterruptedException {
        System.err.println("消费者2........接收到消息：【" + msg + "】" + LocalTime.now());
        Thread.sleep(200);
    }

注意到这个消费者sleep了1000秒，模拟任务耗时。

### 3、测试 ###

> 启动ConsumerApplication后，在执行publisher服务中刚刚编写的发送测试方法testWorkQueue。
> 
> 可以看到消费者1很快完成了自己的25条消息。消费者2却在缓慢的处理自己的25条消息。
> 
> 也就是说消息是平均分配给每个消费者，并没有考虑到消费者的处理能力。这样显然是有问题的。

在spring中有一个简单的配置，可以解决这个问题。我们修改consumer服务的application.yml文件，添加配置：

logging:
      pattern:
        dateformat: MM-dd HH:mm:ss:SSS
    spring:
      rabbitmq:
        host: 192.168.71.132
        port: 5672
        username: itcast
        password: 123456
        virtual-host: /
        listener:
          simple:
            prefetch: 1 # 每次只能获取一条消息，处理完成才能获取下一个消息

### 4、总结 ###

> Work模型的使用
> 
>  *  多个消费者绑定到一个队列，同一条消息只会被一个消费者处理
>  *  通过设置prefetch来控制消费者预取的消息数量

## 七、发布/订阅 ##

发布订阅的模型如图

![e5f60081e24a488e8d1a0f7a42f8964a.png][]

可以看到，在订阅模型中，多了一个exchange角色，而且过程略有变化：

**Publisher：**生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给X（交换机

> **Exchange：**交换机，图中的X。一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。Exchange有以下3种类型：（**Fanout：**广播，将消息交给所有绑定到交换机的队列。**Direct：**定向，把消息交给符合指定routing key 的队列。**Topic：**通配符，把消息交给符合routing pattern（路由模式） 的队列）
> 
> **Consumer：**消费者，与以前一样，订阅队列，没有变化
> 
> **Queue：**消息队列也与以前一样，接收消息、缓存消息

**Exchange（交换机）只负责转发消息，不具备存储消息的能力**，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！

### 1、Fanout ###

Fanout，英文翻译是扇出，我觉得在MQ中叫广播更合适。

> 在广播模式下，消息发送流程是这样的：
> 
> 1） 可以有多个队列
> 
> 2） 每个队列都要绑定到Exchange（交换机）
> 
> 3） 生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无法决定
> 
> 4） 交换机把消息发送给绑定过的所有队列
> 
> 5） 订阅队列的消费者都能拿到消息

我们的计划是这样的：

> 创建一个交换机 itcast.fanout，类型是Fanout
> 
> 创建两个队列fanout.queue1和fanout.queue2，绑定到交换机itcast.fanout

![e69d90405d1a42309b75de9acb629816.png][]

**简称：在交换机里面的队列我都要通知到**

在consumer中创建一个类，声明队列和交换机

package cn.itcast.mq.config;
    
    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("itcast.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);
        }
    }

在publisher服务的SpringAmqpTest类中添加测试方法：

//    FanoutQueue 测试
        @Test
        public void test3() throws InterruptedException {
            // 交换机名称
            String queueName = "itcast.fanout";
            // 消息
            String message = "hello, message____";
    
            // 发送消息
            rabbitTemplate.convertAndSend(queueName, "", message);
    
        }

在consumer服务的SpringRabbitListener中添加两个方法，作为消费者：

@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的会将消息路由到每个绑定的队列

### 2、Direct ###

> 在Fanout模式中，一条消息，会被所有订阅的队列都消费。但是，在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。

![28a892fa623f41f59d2caf029d3f1ebe.png][]

**在Direct模型下**

> 队列与交换机的绑定，不能是任意绑定了，而是要指定一个`RoutingKey`（路由key）
> 
> 消息的发送方在 向 Exchange发送消息时，也必须指定消息的 `RoutingKey`。
> 
> Exchange不再把消息交给每一个绑定的队列，而是根据消息的`Routing Key`进行判断，只有队列的`Routingkey`与消息的 `Routing key`完全一致，才会接收到消息

**案例需求如下**：

> 1.  利用@RabbitListener声明Exchange、Queue、RoutingKey
> 2.  在consumer服务中，编写两个消费者方法，分别监听direct.queue1和direct.queue2
> 3.  在publisher中编写测试方法，向itcast. direct发送消息

![f483491df8454ae3a584334b7314ffca.png][]

#### 2.1、基于注解声明队列和交换机 ####

基于@Bean的方式声明队列和交换机比较麻烦，Spring还提供了基于注解方式来声明。

在consumer的SpringRabbitListener中添加两个消费者，同时基于注解来声明队列和交换机：

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "direct.queue1"),
        exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),
        key = {"red", "blue"}
    ))
    public void listenDirectQueue1(String msg){
        System.out.println("消费者接收到direct.queue1的消息：【" + msg + "】");
    }
    
    @RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "direct.queue2"),
        exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),
        key = {"red", "yellow"}
    ))
    public void listenDirectQueue2(String msg){
        System.out.println("消费者接收到direct.queue2的消息：【" + msg + "】");
    }

#### 2.2、消息发送 ####

在publisher服务的SpringAmqpTest类中添加测试方法：

@Test
    public void testSendDirectExchange() {
        // 交换机名称
        String exchangeName = "itcast.direct";
        // 消息
        String message = "红色警报！日本乱排核废水，导致海洋生物变异，惊现哥斯拉！";
        // 发送消息
        rabbitTemplate.convertAndSend(exchangeName, "red", message);
    }

**总结**

**描述下Direct交换机与Fanout交换机的差异？**

> Fanout交换机将消息路由给每一个与之绑定的队列
> 
> Direct交换机根据RoutingKey判断路由给哪个队列
> 
> 如果多个队列具有相同的RoutingKey，则与Fanout功能类似

### 5、Topic ###

**说明**

> `Topic`类型的`Exchange`与`Direct`相比，都是可以根据`RoutingKey`把消息路由到不同的队列。只不过`Topic`类型`Exchange`可以让队列在绑定`Routing key` 的时候使用通配符！

> `Routingkey` 一般都是有一个或多个单词组成，多个单词之间以”.”分割。
> 
> `#`：匹配一个或多个词
> 
> `*`：匹配不多不少恰好1个词

要求

> 1.  并利用@RabbitListener声明Exchange、Queue、RoutingKey
> 2.  在consumer服务中，编写两个消费者方法，分别监听topic.queue1和topic.queue2
> 3.  在publisher中编写测试方法，向itcast. topic发送消息

#### 5.1、消息发送 ####

在publisher服务的SpringAmqpTest类中添加测试方法：

/**
         * topicExchange
         */
    @Test
    public void testSendTopicExchange() {
        // 交换机名称
        String exchangeName = "itcast.topic";
        // 消息
        String message = "喜报！孙悟空大战哥斯拉，胜!";
        // 发送消息
        rabbitTemplate.convertAndSend(exchangeName, "china.news", message);
    }

#### 5.2、消息接收 ####

在consumer服务的SpringRabbitListener中添加方法：

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "topic.queue1"),
        exchange = @Exchange(name = "itcast.topic", type = ExchangeTypes.TOPIC),
        key = "china.#"
    ))
    public void listenTopicQueue1(String msg){
        System.out.println("消费者接收到topic.queue1的消息：【" + msg + "】");
    }
    
    @RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "topic.queue2"),
        exchange = @Exchange(name = "itcast.topic", type = ExchangeTypes.TOPIC),
        key = "#.news"
    ))
    public void listenTopicQueue2(String msg){
        System.out.println("消费者接收到topic.queue2的消息：【" + msg + "】");
    }

**总结**

**描述下Direct交换机与Topic交换机的差异？**

>  *  Topic交换机接收的消息RoutingKey必须是多个单词，以 `**.**` 分割
>  *  Topic交换机与队列绑定时的bindingKey可以指定通配符
>  *  `#`：代表0个或多个词
>  *  `*`：代表1个词

#### 5.3、消息转换器 ####

之前说过，Spring会把你发送的消息序列化为字节发送给MQ，接收消息的时候，还会把字节反序列化为Java对象。

![f85f829eefda44c4afd341bbbaf744db.png][]

> 只不过，默认情况下Spring采用的序列化方式是JDK序列化。众所周知，JDK序列化存在下列问题：
> 
> 数据体积过大 、有安全漏洞 、可读性差

我们来测试一下。

#### 5.4、测试默认转换器 ####

我们修改消息发送的代码，发送一个Map对象：

//    发送Map 测试
        @Test
        public void test6() throws InterruptedException {
    
            Map<String, Object> msg = new HashMap<>();
            msg.put("name","程序猿追");
            msg.put("age",20);
    
            // 发送消息
            rabbitTemplate.convertAndSend("object.queue",msg);
    
        }

停止consumer服务

发送消息后查看控制台：

![7ab7009b8a2347708d7bc6bea87a30e2.png][]

#### 5.5、配置JSON转换器 ####

显然，JDK序列化方式并不合适。我们希望消息体的体积更小、可读性更高，因此可以使用JSON方式来做序列化和反序列化。

在父项目引入依赖：

<dependency>
                <groupId>com.fasterxml.jackson.core</groupId>
                <artifactId>jackson-databind</artifactId>
            </dependency>

**配置消息转换器**

在启动类中添加一个Bean即可：

@Bean
    public MessageConverter jsonMessageConverter(){
        return new Jackson2JsonMessageConverter();
    }

从队列拉取(Consumer)同样也是要加上

**拉取代码**

@RabbitListener(queues = "object.queue")
    public void objectQueue(Map<String, Object> msg) {
        System.out.println("消费者Object1: " + msg );
    }

![51ab5a10a1c94748a35eb26978a07cbc.png][]

> **不积跬步无以至千里，趁年轻，使劲拼，给未来的自己一个交代！向着明天更好的自己前进吧！**
> 
> ![d5358fcd619443fe98143f2fd5fef762.gif][]

[168c1abc67f4402ebfc99ff077a3e4da.gif]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/23/9e828afd0fc34bdabc9b718cc15c1960.gif
[Link 1]: https://blog.csdn.net/aasd23?spm=1000.2115.3001.5343
[Link 2]: https://blog.csdn.net/aasd23/category_11867778.html?spm=1001.2014.3001.5482
[7d3d8653bf6c4fa3b3026ce900cb9590.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/23/59eb95531d5344fea6a26e29fc1b4892.png
[088ddc68803a45978e0c103f94a8f8c8.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/23/21d09d6f53f4476d96a93ab609a20406.png
[40dee56b8d3941dea2485f2a05687752.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/23/d52515026b334d47aab4797618c1a896.png
[8e9eecfabcc2488695917acc164ea9df.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/23/955cd1439a164ac7bdcd5b29194769a8.png
[85bb9e4906c3477fb0f5b46003a85aa8.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/23/cfb56795dbfd446787a7a372c2f1ad3f.png
[Spring AMQP]: https://spring.io/projects/spring-amqp
[4594c56779964bebaf8108f86993f00c.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/23/b824ae2bde304d86b92093b14f77f2a3.png
[b8291eb81c384c11ab79beb404dc0bb4.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/23/5c5d3aa7cda74ff593ceaa1cf6f1725c.png
[e5f60081e24a488e8d1a0f7a42f8964a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/23/287c7808c2d44cecae6a4edba40a757f.png
[e69d90405d1a42309b75de9acb629816.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/23/5d8283b36f324f388b4caa2d274b0c18.png
[28a892fa623f41f59d2caf029d3f1ebe.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/23/eab80ce1027d4d159cd284f93b45c524.png
[f483491df8454ae3a584334b7314ffca.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/23/a7b4dcfd38d94fb2aad435d0e8a116e4.png
[f85f829eefda44c4afd341bbbaf744db.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/23/caa5096a71d948949445cb43026199dc.png
[7ab7009b8a2347708d7bc6bea87a30e2.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/23/6bcfbfaf2aa24fb8a3d3b9de36e2ebbb.png
[51ab5a10a1c94748a35eb26978a07cbc.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/23/800d3d3ccb1446f28fb5493ca34fe811.png
[d5358fcd619443fe98143f2fd5fef762.gif]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/23/11e7274484524de7a5dcddbcbdad4ca1.gif