首页 SpringCloud

# 初识 RabbitMQ

# 同步和异步通讯

同步通讯：就像打电话，需要实时响应

异步通讯：就像发邮件，不需要马上回复

# 同步通讯

image-20230612210955502

  1. 优点：
    1. 时效性较强，可以立即得到结果
  2. 缺点：
    1. 耦合度高
    2. 性能和吞吐能力下降
    3. 有额外的资源消耗
    4. 有级联失败问题

# 异步通讯

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

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

image-20230612211136598

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

# 技术对比

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

# 2.1 常见 MQ 比较

RabbitMQ ActiveMQ RocketMQ Kafka
公司 / 社区 Rabbit Apache 阿里 Apache
开发语言 Erlang Java Java Scala&Java
协议支持 AMQP，XMPP，SMTP，STOMP OpenWire,STOMP，REST,XMPP,AMQP 自定义协议 自定义协议
可用性 高 一般 高 高
单机吞吐量 一般 差 高 非常高
消息延迟 微秒级 毫秒级 毫秒级 毫秒以内
消息可靠性 高 一般 高 一般

# SpringAMQP

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

image-20210717164024967

image-20210717164038678

SpringAMQP 提供了三个功能：

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

# Basic Queue 简单队列模型

# 引入依赖

<!--AMQP 依赖，包含 RabbitMQ-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

# 消息发送

# application.yml 文件中添加配置

spring:
  rabbitmq:
    host: 192.168.150.101 # 主机名
    port: 5672 # 端口
    virtual-host: / # 虚拟主机
    username: root # 用户名
    password: root # 密码

# 编写测试用例用于消息发送

@SpringBootTest
public class SpringAmqpTest {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    public void testSimpleQueue() {
        // 队列名称
        String queueName = "simple.queue";
        // 消息
        String message = "hello, spring amqp!";
        // 发送消息
        rabbitTemplate.convertAndSend(queueName, message);
    }
}
# 消息接收


# application.yml 文件中添加配置


spring:
  rabbitmq:
    host: 192.168.150.101 # 主机名
    port: 5672 # 端口
    virtual-host: / # 虚拟主机
    username: root # 用户名
    password: root # 密码
# 创建 RabbitListener 类


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


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


image-20210717164238910


当消息处理比较耗时的时候，可能生产消息的速度会远远大于消息的消费速度。长此以往，消息就会堆积越来越多，无法及时处理


此时就可以使用 work 模型，多个消费者共同处理消息处理，速度就能大大提高了


# 消息发送


@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);
    }
}
# 消息接收


@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);
}
# 充分利用效率高的队列


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


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


image-20210717165309625


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


    

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

  • Exchange：交换机，图中的 X。一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于 Exchange 的类型。Exchange 有以下 3 种类型：

      

    • Fanout：广播，将消息交给所有绑定到交换机的队列
      • 

    • Direct：定向，把消息交给符合指定 routing key 的队列
      • 

    • Topic：通配符，把消息交给符合 routing pattern（路由模式） 的队列
      • 

    

    • 

  • Consumer：消费者，与以前一样，订阅队列，没有变化
    • 

  • Queue：消息队列也与以前一样，接收消息、缓存消息。
    • 



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


# Fanout


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


image-20210717165438225


在广播模式下，消息发送流程是这样的：


    

  • 1）  可以有多个队列
    • 

  • 2）  每个队列都要绑定到 Exchange（交换机）
    • 

  • 3）  生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无法决定
    • 

  • 4）  交换机把消息发送给绑定过的所有队列
    • 

  • 5）  订阅队列的消费者都能拿到消息
    • 



我们的计划是这样的：


    

  • 创建一个交换机 itcast.fanout，类型是 Fanout
    • 

  • 创建两个队列 fanout.queue1 和 fanout.queue2，绑定到交换机 itcast.fanout
    • 



image-20210717165509466


# 声明队列和交换机


Spring 提供了一个接口 Exchange，来表示所有不同类型的交换机：


image-20210717165552676


# 创建一个类，声明队列和交换机


@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);
    }
}
# 消息发送


@Test
public void testFanoutExchange() {
    // 队列名称
    String exchangeName = "itcast.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 + "】");
}
# Direct


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


image-20210717170041447


在 Direct 模型下：


    

  • 队列与交换机的绑定，不能是任意绑定了，而是要指定一个 RoutingKey （路由 key）
    • 

  • 消息的发送方在 向 Exchange 发送消息时，也必须指定消息的  RoutingKey 。
    • 

  • Exchange 不再把消息交给每一个绑定的队列，而是根据消息的 Routing Key  进行判断，只有队列的 Routingkey  与消息的  Routing key  完全一致，才会接收到消息
    • 



案例需求如下：


    

  1. 

    利用 @RabbitListener 声明 Exchange、Queue、RoutingKey
    

    2. 

  2. 

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

    3. 

  3. 

    在 publisher 中编写测试方法，向 itcast. direct 发送消息
    

    4. 



image-20210717170223317


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


@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 + "】");
}
# 消息发送


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


# 说明


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


Routingkey  一般都是有一个或多个单词组成，多个单词之间以”.” 分割，例如：  item.insert


通配符规则：


# ：匹配一个或多个词


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


# 图示及案例需求


# 图示


image-20210717170705380


解释：


    

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

  • Queue2：绑定的是 #.news  ，因此凡是以  .news  结尾的  routing key  都会被匹配。包括 china.news 和 japan.news
    • 



# 案例需求


实现思路如下：


    

  1. 

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

    2. 

  2. 

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

    3. 

  3. 

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

    4. 



image-20210717170829229


# 消息发送


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


@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 + "】");
}
# 消息转换器


# 问题产生


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


image-20200525170410401


只不过，默认情况下 Spring 采用的序列化方式是 JDK 序列化。众所周知，JDK 序列化存在下列问题：


    

  • 数据体积过大
    • 

  • 有安全漏洞
    • 

  • 可读性差
    • 



# 配置 JSON 转换器


# 引入依赖


<dependency>
    <groupId>com.fasterxml.jackson.dataformat</groupId>
    <artifactId>jackson-dataformat-xml</artifactId>
    <version>2.9.10</version>
</dependency>
# 配置消息转换器


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


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

      

          
      

  


   


    

  
    
      
    
    更新于
    
  
  
      
        
      
      阅读次数
      
      次
  



      


  
  
请我喝[茶]~(￣▽￣)~*

  

      
      

        Baozi 微信支付
        
微信支付

      

      
      

        Baozi 支付宝
        
支付宝

      

      
      

        Baozi 微信
        
微信