Spring AMQP

大家好我是苏麟今天说一说spring aqmp。

SpringAMQP

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

官方 : Spring AMQP

依赖

        <!--AMQP 包含RabbitMQ-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
            <version>${amqp.version}</version>
        </dependency>
        <dependency>
            <groupId>com.github.luues</groupId>
            <artifactId>spring-boot-starter-rabbitmq</artifactId>
            <version>1.3.0.5.RELEASE</version>
        </dependency>

SpringAMQP提供了三个功能：

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

Basic Queue 简单队列模型

在父工程mq-demo中引入依赖

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

消息发送

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

spring:
  rabbitmq:
    port: 5672
    host: 127.0.0.1 # 主机名
    username: guest # 用户名
    password: guest # 密码
    virtual-host: / # 虚拟主机

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

package com.sl;

import org.junit.jupiter.api.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

/**
 * springAMQP
 */
@RunWith(SpringRunner.class)
@SpringBootTest
class PublisherApplicationTests {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * Simple 简单队列
     */
    @Test
    void testSimple() {
        //队列名
        String queues = "ty";
        //消息
        String message = "hello ty";
        System.out.println("666");
        //发送信息
        rabbitTemplate.convertAndSend(queues, message);
    }

}

消息接收

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

spring:
  rabbitmq:
    port: 5672
    host: 127.0.0.1
    username: guest
    password: guest
    virtual-host: /

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

package com.sl.spring;

import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;



@Component
public class SpringRabbitMQListener {

    @RabbitListener(queues = "ty")
    public void listenerSimpleQueue(String msg) {
        System.out.println("消费者接收到的消息 : " + msg);
    }
}

测试

启动consumer服务，然后在publisher服务中运行测试代码，发送MQ消息 .

我们发送消息之前需要自己创建一个队列

接受消息

WorkQueue 工作队列

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

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

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

消息发送

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

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

package com.sl;

import org.junit.jupiter.api.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

/**
 * springAMQP
 */
@RunWith(SpringRunner.class)
@SpringBootTest
class PublisherApplicationTests {

    @Autowired
    private RabbitTemplate rabbitTemplate;


    /**
     * Work 工作队列
     */
    @Test
    void testWork() {
        String queues = "ty";

        for (int i = 0; i <= 50; i++) {
            String message = "hello rabbitmq__";
            rabbitTemplate.convertAndSend(queues, message + i);
        }
    }

}

消息接收

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

package com.sl.spring;

import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;

/**
 * @className: SpringRabbitMQListence
 * @author: TianYuan ShowTime
 * @date: 2023/7/29-9:30
 **/

@Component
public class SpringRabbitMQListener {

    /**
     * @RabbitListener 声明队列名称
     */

    @RabbitListener(queues = "ty")
    public void listenerWorkQueue1(String msg) throws InterruptedException {
        System.out.println("消费者1......接收到的消息: " + msg + LocalDateTime.now());
        Thread.sleep(20);
    }

    @RabbitListener(queues = "ty")
    public void listenerWorkQueue2(String msg) throws InterruptedException {
        System.out.println("消费者2......接收到的消息" + msg + LocalDateTime.now());
        Thread.sleep(200);
    }

}

测试 :

启动ConsumerApplication后，在执行publisher服务中刚刚编写的发送测试方法testWorkQueue。

这会耗时

能者多劳

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

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

节省了好几秒的时间

总结 :

Work模型的使用：

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

发布/订阅

发布订阅的模型如图：

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

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

Fanout：广播，将消息交给所有绑定到交换机的队列
Direct：定向，把消息交给符合指定routing key 的队列
Topic：通配符，把消息交给符合routing pattern（路由模式）的队列

Consumer：消费者，与以前一样，订阅队列，没有变化 Queue：消息队列也与以前一样，接收消息、缓存消息。

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

Fanout

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

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

1）可以有多个队列
2）每个队列都要绑定到Exchange（交换机）
3）生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无法决定
4）交换机把消息发送给绑定过的所有队列
5）订阅队列的消费者都能拿到消息

我们的计划是这样的：

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

声明队列和交换机

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

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

package com.sl.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 AMQPConfig {

    /**
     * 创建交换机
     *
     * @return
     */
    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange("sl.exchange");
    }

    /**
     * 创建队列
     *
     * @return
     */
    @Bean
    public Queue fanoutQueue1() {
        return new Queue("fanoutQueue1");
    }

    @Bean
    public Queue fanoutQueue2() {
        return new Queue("fanoutQueue2");
    }

    /**
     * 创建连接
     *
     * @return
     */
    @Bean
    public Binding fanoutBinding1(FanoutExchange fanoutExchange, Queue fanoutQueue1) {
        return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
    }

    @Bean
    public Binding fanoutBinding2(FanoutExchange fanoutExchange, Queue fanoutQueue2) {
        return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
    }

}

消息发送

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

package com.sl;

import org.junit.jupiter.api.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

/**
 * springAMQP
 */
@RunWith(SpringRunner.class)
@SpringBootTest
class PublisherApplicationTests {

    @Autowired
    private RabbitTemplate rabbitTemplate;


    /**
     * 发布订阅模型
     * <p>
     * Fanout 广播
     */
    @Test
    void testFanout() {
        String exchangeName = "sl.exchange";

        /**
         *  ctrl + p  查看参数
         */
        rabbitTemplate.convertAndSend(exchangeName, "", "中南海");


    }

}

消息接收

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

package com.sl.spring;

import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;



@Component
public class SpringRabbitMQListener {


    /**
     * Fanout 广播
     * @RabbitListener 声明队列名称
     * @param msg
     */
    @RabbitListener(queues = "fanoutQueue1")
    public void listenerFanoutQueue1(String msg) {
        System.out.println("消费者1111接收到的消息" + msg);
    }

    @RabbitListener(queues = "fanoutQueue2")
    public void listenerFanoutQueue2(String msg) {
        System.out.println("消费者2222接收到的消息" + msg);
    }
}

总结

交换机的作用是什么？

接收publisher发送的消息
将消息按照规则路由到与之绑定的队列
不能缓存消息，路由失败，消息丢失
FanoutExchange的会将消息路由到每个绑定的队列

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

Queue
FanoutExchange
Binding

Direct

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

在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发送消息

消息接收

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

基于@Bean的方式声明队列和交换机比较麻烦，Spring还提供了基于注解方式来声明。在consumer的SpringRabbitListener中添加两个消费者，同时基于注解来声明队列和交换机：

package com.sl.spring;

import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;


@Component
public class SpringRabbitMQListener {


    /**
     * Direct 路由
     *
     * @param msg
     */
    @RabbitListener(bindings = @QueueBinding(value = @Queue("jwd1"),
            exchange = @Exchange("yk.sl"),
            key = {"blue", "red"})
    )
    public void listenerDirectQueue1(String msg) {
        System.out.println("苏麟打死杨科小弟 1号 :" + msg);
    }

    @RabbitListener(bindings = @QueueBinding(value = @Queue("jwd2"),
            exchange = @Exchange("yk.sl"),
            key = {"yellow", "red"})
    )
    public void listenerDirectQueue2(String msg) {
        System.out.println("苏麟打死杨科小弟 2号 :" + msg);
    }
}

消息发送

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

package com.sl;

import org.junit.jupiter.api.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

/**
 * springAMQP
 */
@RunWith(SpringRunner.class)
@SpringBootTest
class PublisherApplicationTests {

    @Autowired
    private RabbitTemplate rabbitTemplate;



    /**
     * 发布订阅模型
     * <p>
     * Direct 路由
     */
    @Test
    void testDirect() {
        String exchangeName = "yk.sl";

        /**
         *  ctrl + p  查看参数
         */
        rabbitTemplate.convertAndSend(exchangeName, "yellow", "奥特曼");

    }
}