• MQ消息队列--RabbitMQ的简单入门，在SpringBoot环境

    一、初识MQ

    MQ，中文意思是消息队列（MessageQueue），字面来看就是存放消息的队列。也就是事件驱动架构中的Broker。

    1.1常见的几种MQ
    • ActiveMQ
    • RabbitMQ
    • RocketMQ
    • Kafka
    1.2几种MQ的对比

    在这里插入图片描述

    1.3MQ的基本结构

    在这里插入图片描述

    • Publisher：消息的提供者
    • exchange：交换机，负责将消息传递给队列，不具备存储功能
    • queue：队列，用来暂时存放消息的，结构就是队列先进先出，消息取出后就在队列中消失，且不可回溯。
    • consumer：消费者，将消息从queue中取出
    • virtualHost：虚拟主机，隔离不同租户的exchange、queue、消息的隔离
    1.4.RabbitMQ的消息模型
    • 基本消息队列（BasicQueue）
    • 工作消息队列（WorkQueue）
    • 发布订阅（Publisher、Subscribe），根据类型不同又分为三种：
      Fanout Exchange：广播
      Direct Exchange：路由
      Topic Exchange：主题

    2.小案列使用官方提供的方式

    2.1Publisher实现
    • 建立连接
    • 创建Channel
    • 声明队列
    • 发送消息
    • 关闭连接和channel
      代码实现：
    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.150.101");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("itcast");
        factory.setPassword("123321");
        // 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();

    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    2.2 consumer的实现

    代码思路：

    • 建立连接
    • 创建Channel
    • 声明队列
    • 订阅消息

    代码实现

    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.150.101");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("itcast");
        factory.setPassword("123321");
        // 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("等待接收消息。。。。");
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41

    3.使用SpirngAMQP

    文章中Consumer的监听类怎么使用，因为监听类注册成了配置类所以，我们只需运行Consumer这个服务即可。将接收队列消息启动，在去启动Publisher的测试类向队列中写数据即可，在consumer的控制台查看接收到的消息

    3.1什么是AMQP

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

    3.2什么是SpringAMQP

    SpringAMQP是基于RabbitMQ封装的一套模板，并且还利用SpringBoot对其实现了自动装配，使用起来非常方便。Spring AMQP是基于AMQP协议定义的一套API规范，提供了模板来发送和接收消息。包含两部分，其中spring-amqp是基础抽象，spring-rabbit是底层的默认实现。
    SpringAmqp的官方地址：https://spring.io/projects/spring-amqp

    3.2.1 Spirng AMQP提供的三个功能
    • 自动声明队列、交换机及其绑定关系
    • 基于注解的监听器模式，异步接收消息
    • 封装了RabbitTemplate工具，用于发送消息
    3.3 使用Spring AMQP完成Basic Queue的实现

    在父工程中导入依赖坐标

            
        <dependency>
            <groupId>org.springframework.bootgroupId>
            <artifactId>spring-boot-starter-amqpartifactId>
        dependency>

    • 1
    • 2
    • 3
    • 4
    • 5

    在YMAL配置文件配置连接RabbitMQ的信息

    spring:
  rabbitmq:
    host: 192.168.***.105
    port: 5672
    virtual-host: / #虚拟主机，不要怀疑这台虚拟机就叫“/”
    username: chen
    password: ******

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    3.3.1编写Publisher的消息发送测试类：
     @Autowired
    RabbitTemplate rabbitTemplate;
    
 @Test
    public void testSpringAMQP() {
        String queryName = "queryChen";
        String message = "hello chen !!";
//        创建一个RabbitAdmin用于操作队列
        RabbitAdmin rabbitAdmin = new RabbitAdmin(rabbitTemplate);
//        声明一个队列
        rabbitAdmin.declareQueue(new Queue(queryName));
//       向队列里面发送消息
        rabbitTemplate.convertAndSend(queryName, message);
    }

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    3.3.2编写consumer的一个监听类

    前置工作：
    1.需要导入依赖，如果父工程有就不用导入了
    2.编写配置文件和Publisher的配置文件内容一样
    然后在consumer服务的cn.itcast.mq.listener包中新建一个类SpringRabbitListener，代码如下：

        @RabbitListener(queues = "queryChen")
    public void listenSimpleQueueMessage(String msg) throws InterruptedException{
        System.out.println("收到消息："+msg);
    }

    • 1
    • 2
    • 3
    • 4
    3.4使用Spring AMQP实现一个Work queues

    意思就是让多个消费者绑定到一个队列中，共同消费队列中的消息。当消息处理比较耗时的时候，可能生产消息的速度会远远大于消息的消费速度。长此以往，消息就会堆积越来越多，无法及时处理。此时就可以使用work 模型，多个消费者共同处理消息处理，速度就能大大提高了。
    在这里插入图片描述

    3.4.1介绍一个声明消息队列的方法之一：配置类

    使用配置类的方式让spring在创建时就将队列创建好，并将队列和交换机绑定好
    以Fanout Exchange（后面会说）为例：

    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 {
    //    声明交换机
    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange("chen.fanout");
    }

    //    声明队列1
    @Bean
    public Queue fanoutQueue1() {
        return new Queue("fanout.queue1");
    }

    //    声明队列2
    @Bean
    public Queue fanoutQueue2() {
        return new Queue("fanout.queue2");
    }

    //    声明绑定关系,绑定关系按照方法名称进行注入
//    绑定队列1到交换机
    @Bean
    public Binding fanoutBinding1(Queue fanoutQueue1, FanoutExchange fanoutExchange) {
        return BindingBuilder
                .bind(fanoutQueue1)
                .to(fanoutExchange);
    }

    //    绑定队列2到交换机
    @Bean
    public Binding fanoutBinding2(Queue fanoutQueue2, FanoutExchange fanoutExchange) {
        return BindingBuilder
                .bind(fanoutQueue2)
                .to(fanoutExchange);
    }

}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    3.4.2在publisher的测试类中想队列添加多条消息

    目的：50条数据，其中一个消费者处理消息速度快1秒钟处理50个，另个一慢，1秒钟处理10个，想让两个消费者共同完成这50个消息的处理。

       /**
     * @return void
     * @author chenqingxu
     * @description 演示发送50条消息，让两个消费者获取
     */
    @Test
    public void testSpringWorkAMQP() throws Exception {
        String queryName = "queryChen2";
        String message = "hello chen__";
//        创建一个RabbitAdmin用于操作队列
        RabbitAdmin rabbitAdmin = new RabbitAdmin(rabbitTemplate);
//        声明一个队列
        rabbitAdmin.declareQueue(new Queue(queryName));
        for (int i = 0; i < 50; i++) {
//       向队列里面发送消息
            rabbitTemplate.convertAndSend(queryName, message + i);
            Thread.sleep(20);
        }
    }

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    3.4.3 在consumer中创建连个消费者与同一个队列绑定
        @RabbitListener(queues = "queryChen2")
    public void listenWorkQueueMessage(String msg) throws InterruptedException {
        System.out.println("消费者1收到消息：" + msg + "[" + LocalTime.now() + "]");
        Thread.sleep(20);
    }

    @RabbitListener(queues = "queryChen2")
    public void listenWorkQueueMessage2(String msg) throws InterruptedException {
        System.err.println("消费者2...........收到消息：" + msg + "[" + LocalTime.now() + "]");
        Thread.sleep(200);
    }


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12

    知识点补充：此处并没有像我们想的那样，快的消费者处理的比慢的消费者多，而是两个消费者平均处理了队列中的消息。
    原因：因为有一个消息预取机制，两个消费者在处理消息的同时，队列会按默认的轮询策略一直给消费者发送，预取机制没有上限，如果只有两个消费者这样就形成了平分队列中消息的情况。
    解决：让处理快的消费者多处里就需要限制他们的预拉取，在consumer的配置文件中设置：

    spring:
  rabbitmq:
    listener:
      simple:
        prefetch: 1 # 每次只能获取一条消息，处理完成才能获取下一个消息

    • 1
    • 2
    • 3
    • 4
    • 5

    这样处理，速度快的消费者就可以处理的多。

    3.5 订阅/发布

    模型结构如下：
    在这里插入图片描述消息的发送过程发生了改变：

    • Publisher：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给X（交换机）
    • Exchange：交换机，图中的X。一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。Exchange有以下3种类型：
      • Fanout：广播，将消息交给所有绑定到交换机的队列
      • Direct：定向，把消息交给符合指定routing key 的队列
      • Topic：通配符，把消息交给符合routing pattern（路由模式） 的队列
    • Consumer：消费者，与以前一样，订阅队列，没有变化
    • Queue：消息队列也与以前一样，接收消息、缓存消息。
      Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失
    3.6 Fanout Exchange的使用

    模型：
    在这里插入图片描述在广播模式下，消息发送流程是这样的：

    • 1） 可以有多个队列
    • 2） 每个队列都要绑定到Exchange（交换机）
    • 3） 生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无 法决定
    • 4） 交换机把消息发送给绑定过的所有队列
    • 5） 订阅队列的消费者都能拿到消息
    3.6.1演示
    • 创建一个交换机chen.fanout，类型是Fanout
    • 创建两个队列fanout.queue1，fanout.queue2 绑定到交换机chen.fanout
      1.在consumer类中声明交换和队列并进行绑定
      在springboot启动时就会自动在rabbitMq中创建队列和交换机并完成绑定
    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 {
    //    声明交换机
    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange("chen.fanout");
    }

    //    声明队列1
    @Bean
    public Queue fanoutQueue1() {
        return new Queue("fanout.queue1");
    }

    //    声明队列2
    @Bean
    public Queue fanoutQueue2() {
        return new Queue("fanout.queue2");
    }

    //    声明绑定关系,绑定关系按照方法名称和属性进行依赖注入
//    绑定队列1到交换机
    @Bean
    public Binding fanoutBinding1(Queue fanoutQueue1, FanoutExchange fanoutExchange) {
        return BindingBuilder
                .bind(fanoutQueue1)
                .to(fanoutExchange);
    }

    //    绑定队列2到交换机
    @Bean
    public Binding fanoutBinding2(Queue fanoutQueue2, FanoutExchange fanoutExchange) {
        return BindingBuilder
                .bind(fanoutQueue2)
                .to(fanoutExchange);
    }

}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48

    2.Fanout Exchange的使用，在consumer中编写接收消息功能

       /**
     * @param msg
     * @return void
     * @author chenqingxu
     * @description 队列交换机的使用，消费者1处理速度快，消费者2处理速度慢
     */
    @RabbitListener(queues = "fanout.queue1")
    public void listenFanoutQueueMessage1(String msg) throws InterruptedException {
        System.err.println("消费者1...........收到消息：" + msg + "[" + LocalTime.now() + "]");
        Thread.sleep(200);
    }

    @RabbitListener(queues = "fanout.queue2")
    public void listenFanoutQueueMessage2(String msg) throws InterruptedException {
        System.err.println("消费者2...........收到消息：" + msg + "[" + LocalTime.now() + "]");
        Thread.sleep(200);
    }

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17

    3.Fanout Exchange的使用，在publisher测试类中中编写消息发送功能

     /**
     * @return void
     * @author chenqingxu
     * @description 将消息发送给交换机，有交换机将消息放入队列，不在需要指定队列的名字
     */
    @Test
    public void testSentFanoutExchange() {
//        准备交换机名称
        String FanoutExchageName = "chen.fanout";
//        需要发送的消息
        String message = "hello everyone !";
//        发送消息
        rabbitTemplate.convertAndSend(FanoutExchageName, "", message);
    }

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14

    总结：

    交换机的作用是什么？
    • 接收publisher发送的消息
    • 将消息按照规则路由到与之绑定的队列
    • 不能缓存消息，路由失败，消息丢失
    • FanoutExchange的会将消息路由到每个绑定的队列

    声明队列、交换机、绑定关系的Bean是什么？

    • Queue
    • FanoutExchange
    • Binding
    3.7 Direct Exchange的使用

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

    • 队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）
    • 消息的发送方在 向 Exchange发送消息时，也必须指定消息的 RoutingKey。
    • Exchange不再把消息交给每一个绑定的队列，而是根据消息的Routing Key进行判断，只有队列的Routingkey与消息的 Routing key完全一致，才会接收到消息
    • 注意：如果两个队列绑定两个一样的rountingkey，就可以模拟Fanout Exchange
    3.7.1创建交换机和队列并将二者绑定的方式二：注解

    在消费者端Consumer中设置的

    @RabbitListener(bindings = @QueueBinding(
//          value声明队列
            value = @Queue(name="direct.queue1"),
            //exchange声明交换机，name交换机名字，type类型
            exchange = @Exchange(name = "chen.direct",type = "direct"),
            // 指定交换机向指定了bindingkey的队列发消息
            key = {"red","blue"}
    ))

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    3.7.2 演示

    1. 利用@RabbitListener声明Exchange、Queue、RoutingKey

    2. 在consumer服务中，编写两个消费者方法，分别监听direct.queue1和direct.queue2

    3. 在publisher中编写测试方法，向chen. direct发送消息
      Consumer服务：

        /**
     * @param msg
     * @return void
     * @author chenqingxu
     * @description 测试DirectExchange模式的交换机, 在@RabbitListener注解中声明对列以及交换机，如果Rabbit中没有，spring会帮我们创建
     * 注意，当消息中的routingkey为red时，两个队列都可收到消息
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name="direct.queue1"),
            exchange = @Exchange(name = "chen.direct",type = "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 = "chen.direct",type = "direct"),
            key = {"red","yellow"}
    ))
    public void listenDirectQueue2(String msg) {
        System.out.println("消费者收到direct.queue2的消息：" + msg);
    }

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24

    Publisher在测试类中实现发送消息功能

    /**
     * @return void
     * @author chenqingxu
     * @description 将消息发送给direct交换机，指定绑定的key
     */
    @Test
    public void testSentDirectExchange() {
//        准备交换机名称
        String FanoutExchageName = "chen.direct";
//        需要发送的消息
//         String message = "hello,blue";
//        String message = "hello,yellow";
        String message = "hello,red";
//        发送消息,routingKey就是direct中和交换机绑定的队列的key
        String[] split = message.split(",");
        //split[1]就是routingkey我是为了测试方便
        rabbitTemplate.convertAndSend(FanoutExchageName, split[1], message);
    }

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    总结：
    描述下Direct交换机与Fanout交换机的差异？
    • Fanout交换机将消息路由给每一个与之绑定的队列
    • Direct交换机根据RoutingKey判断路由给哪个队列
    • 如果多个队列具有相同的RoutingKey，则与Fanout功能类似
    基于@RabbitListener注解声明队列和交换机有哪些常见注解？
    • @Queue
    • @Exchange
    3.8Topic Exchange的使用

    Topic类型的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符！
    Routingkey 一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert

    通配符规则：

    #：匹配一个或多个词或0个

    *：匹配不多不少恰好1个词
    举例：

    item.#：能够匹配item.spu.insert 或者 item.spu

    item.*：只能匹配item.spu

    模型：
    在这里插入图片描述

    • Queue1：绑定的是china.# ，因此凡是以 china.开头的routing key 都会被匹配到。包括china.news和china.weather
    • Queue2：绑定的是#.news ，因此凡是以 .news结尾的 routing key 都会被匹配。包括china.news和japan.news
    3.8.1演示

    实现思路如下：

    1. 并利用@RabbitListener声明Exchange、Queue、RoutingKey

    2. 在consumer服务中，编写两个消费者方法，分别监听topic.queue1和topic.queue2

    3. 在publisher中编写测试方法，向chen. topic发送消息

    consumer中编写方法：

    /**
     * @param msg
     * @return void
     * @author chenqingxu
     * @description 测试TopicExchange模式的交换机
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "topic.queue1"),
            exchange = @Exchange(name = "chen.topic",type ="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 = "chen.topic",type ="topic"),
            key = {"#.news"}
    ))
    public void listenTopicQueue2(String msg) {
        System.out.println("消费者收到topic.queue2的消息：" + msg);
    }

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23

    Publisher的测试类中编写测试方法

      /**
     * @return void
     * @author chenqingxu
     * @description 将消息发送给topic交换机，指定绑定的rountingkey
     */
    @Test
    public void testSentTopicExchange() {
//        准备交换机名称
        String FanoutExchageName = "chen.topic";
//        需要发送的消息
//         String message = "hello,blue";
//        String message = "hello,yellow";
        String message = "China is awesome";
//        发送消息,routingKey就是topic中和交换机绑定的队列的key
        rabbitTemplate.convertAndSend(FanoutExchageName, "china.new", message);
       // rabbitTemplate.convertAndSend(FanoutExchageName, "china.news", message);
    }

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    3.9 Spirng-AMQP消息转换器

    Spring的消息对象的处理是由org.springframework.amqp.support.converter.MessageConverter来处理额。而默认实现是SimpleMessageConverter，基于JDK的ObjectOutStream完成序列化。如果要修改，需要重新定义一个MessageConverter类型的Bean即可。推荐用JSON方式序列化。

    测试默认转换器：

    消息发送：

    @Test
public void testSendMap() throws InterruptedException {
    // 准备消息
    Map<String,Object> msg = new HashMap<>();
    msg.put("name", "Jack");
    msg.put("age", 21);
    // 发送消息
    rabbitTemplate.convertAndSend("simple.queue","", msg);
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9

    结果：
    在这里插入图片描述总结：
    JDK序列化存在下列问题：

    • 数据体积过大
    • 有安全漏洞
    • 可读性差
    解决方案

    使用JSON序列化，我们可以在父工程导入，也可以在两个子工程中都导入：

            <dependency>
            <groupId>com.fasterxml.jackson.coregroupId>
            <artifactId>jackson-databindartifactId>
        dependency>

    • 1
    • 2
    • 3
    • 4

    1.在消息发送方Publisher编写配置类，也可不写配置类，写在启动类中：

    步骤一：先导入依赖，可以导入父工程中，也可以在提供者的工程中导入，声明在父工程最好，因为消息接收服务也需要导入依赖
    步骤二：在publisher服务中声明MessageConverter，这样就可以覆盖掉默认的消息转换器，这是springBoot的特性

    package cn.itcast.mq;

import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;

@SpringBootApplication
public class PublisherApplication {
    public static void main(String[] args) {
        SpringApplication.run(PublisherApplication.class);
    }
    @Bean
    public MessageConverter jsonMessageConverter(){
              return new Jackson2JsonMessageConverter();
    }
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19

    2.消息接收端Consumer：

    步骤一：导入依赖
    步骤二：声明MessageConverter
    前两步和上面的一样
    步骤三：
    接收消息的类型，发送方发的什么类型的消息，我们就用什么类型来接收

  • 相关阅读:
    分布式事务（Seata）——Seata分布式事务XA模式、AT模式、TCC模式的介绍和对比 & 结合案例分析AT模式和XA模式【源码】
    掘光者网课题库接口
    【Angular1】基础使用及各类方法
    实验十三————网页爬虫
    【PyTorch深度学习项目实战100例】—— 基于ResNet50实现多目标美味蛋糕图像分类 | 第51例
    回溯算法题目
    RK1126平台项目总结
    记录一个Spring自己注入自己的一个坑
    bashplotlib，一个有趣的 Python 数据可视化图形库
    libstdc++.so.6 版本过低导致的编译失败
  • 原文地址：https://blog.csdn.net/m0_45101736/article/details/126529988