SpringAMQP详解_spring amqp

一，Spring AMQP介绍

Java API 方式编程，有什么问题？

Spring 封装 RabbitMQ 的时候，它做了什么事情？

1、管理对象（队列、交换机、绑定）

2、封装方法（发送消息、接收消息）

Spring AMQP 是对 Spring 基于 AMQP 的消息收发解决方案，它是一个抽象层，不依赖于特定的 AMQP Broker 实现和客户端的抽象，所以可以很方便地替换。比如我们可以使用 spring-rabbit 来实现。

二，Spring AMQP核心组件

1，ConnectionFactory

Spring AMQP 的连接工厂接口，用于创建连接。CachingConnectionFactory 是ConnectionFactory 的一个实现类。

2，RabbitAdmin

RabbitAdmin 是 AmqpAdmin 的实现，封装了对 RabbitMQ 的基础管理操作，比如对交换机、队列、绑定的声明和删除等。

/**
 * @author yhd
 * @createtime 2021/1/28 19:24
 * @description Spring AMQP configuration class
 */
@SpringBootConfiguration
public class AMQPConfig {

    private static final String EXCHANGE_NAME = "amqp.yhd.exchange";

    private static final String QUEUE_NAME = "amqp.yhd.queue";

    private static final String ROUTING_KEY = "amqp.admin";

    @Bean
    public ConnectionFactory factory() {
        CachingConnectionFactory factory = new CachingConnectionFactory();
        factory.setAddresses("121.199.31.160");
        factory.setPort(5672);
        factory.setUsername("root");
        factory.setPassword("root");
        return factory;
    }

    @Bean
    public AmqpAdmin amqpAdmin( ConnectionFactory factory) {
        RabbitAdmin admin = new RabbitAdmin(factory);
        //声明一个交换机 交换机名  是否持久化  是否自动删除
        admin.declareExchange(new DirectExchange(EXCHANGE_NAME, true, false));
        //队列名 持久化 是否批处理  自动删除
        admin.declareQueue(new Queue(QUEUE_NAME, true, false, false));
        //声明一个绑定 队列名 ，绑定类型，交换机名，路由键 参数
        admin.declareBinding(new Binding(QUEUE_NAME, Binding.DestinationType.QUEUE, EXCHANGE_NAME, ROUTING_KEY, null));
        return admin;
    }
}
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

为什么我们在配置文件（Spring）或者配置类（SpringBoot）里面定义了交换机、队列、绑定关系，并没有直接调用 Channel 的 declare 的方法，Spring 在启动的时候就可以帮我们创建这些元数据？这些事情就是由 RabbitAdmin 完成的。

RabbitAdmin 实 现 了 InitializingBean 接 口 ， 里 面 有 唯 一 的 一 个 方 法afterPropertiesSet()，这个方法会在 RabbitAdmin 的属性值设置完的时候被调用。

在 afterPropertiesSet ()方法中，调用了一个 initialize()方法。这里面创建了三个Collection，用来盛放交换机、队列、绑定关系。

最后依次声明返回类型为 Exchange、Queue 和 Binding 这些 Bean，底层还是调用了 Channel 的 declare 的方法。

declareExchanges(channel, exchanges.toArray(new Exchange[exchanges.size()]));
declareQueues(channel, queues.toArray(new Queue[queues.size()]));
declareBindings(channel, bindings.toArray(new Binding[bindings.size()]));
1
2
3

3，Message

Message 是 Spring AMQP 对消息的封装。两个重要的属性：body：消息内容。 messageProperties：消息属性。

4，RabbitTemplate 消息模板

RabbitTemplate 是 AmqpTemplate 的一个实现（目前为止也是唯一的实现），用来简化消息的收发，支持消息的确认（Confirm）与返回（Return）。跟 JDBCTemplate一 样 ， 它 封 装 了 创 建 连 接 、 创 建 消 息 信 道 、 收 发 消 息 、 消 息 格 式 转 换（ConvertAndSend→Message）、关闭信道、关闭连接等等操作。

针对于多个服务器连接，可以定义多个 Template。可以注入到任何需要收发消息的地方使用。

    /**
     * return callback   &&  confirm callable
     *
     * @param factory
     * @return
     */
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory factory) {
        RabbitTemplate template = new RabbitTemplate(factory);
        template.setMandatory(true);
        template.setReturnCallback((Message message,
                                    int replyCode,
                                    String replyText,
                                    String exchange,
                                    String routingKey) -> {

        });

        template.setConfirmCallback((CorrelationData correlationData, boolean ack, String cause) -> {
            if (ack) {
                log.info("消息确认成功！");
            } else {
                log.info("消息确认失败！");
            }
        });
        return template;
    }
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

5，Messager Listener 消息监听

MessageListener

MessageListener 是 Spring AMQP 异步消息投递的监听器接口，它只有一个方法onMessage，用于处理消息队列推送来的消息，作用类似于 Java API 中的 Consumer。

MessageListenerContainer

MessageListenerContainer可以理解为MessageListener的容器，一个Container只有一个 Listener，但是可以生成多个线程使用相同的 MessageListener 同时消费消息。

Container 可以管理 Listener 的生命周期，可以用于对于消费者进行配置。

例如：动态添加移除队列、对消费者进行设置，例如 ConsumerTag、Arguments、并发、消费者数量、消息确认模式等等。

    /**
     * 消息监听器容器
     * @param connectionFactory
     * @return
     */
    @Bean
    public SimpleMessageListenerContainer messageContainer(ConnectionFactory connectionFactory) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        //监听的队列
        container.setQueues(new Queue(QUEUE_NAME, true, false, false));
        // 最小消费者数
        container.setConcurrentConsumers(1);
        // 最大的消费者数量
        container.setMaxConcurrentConsumers(5);
        //是否重回队列
        container.setDefaultRequeueRejected(false);
        //签收模式
        container.setAcknowledgeMode(AcknowledgeMode.AUTO);
        container.setExposeListenerChannel(true);
        //消费端的标签策略
        container.setConsumerTagStrategy(queue -> queue + "_" + UUID.randomUUID().toString());
        return container;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

在 SpringBoot2.0 中新增了一个 DirectMessageListenerContainer。

MessageListenerContainerFactory

Spring 去整合 IBM MQ、JMS、Kafka 也是这么做的。

    /**
     * 
     * @param connectionFactory
     * @return
     */
    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        factory.setAcknowledgeMode(AcknowledgeMode.NONE);
        factory.setAutoStartup(true);
        return factory;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14

可以在消费者上指定，当我们需要监听多个 RabbitMQ 的服务器的时候，指定不同的 MessageListenerContainerFactory。

@Slf4j
@Component
@PropertySource("classpath:application.properties")
@RabbitListener(queues = "${amqp.yhd.queue}", containerFactory = "rabbitListenerContainerFactory")
public class FirstConsumer {

    @RabbitHandler
    public void process(@Payload String message) {
        log.info("First Queue received msg : {}", message);
    }
}
1
2
3
4
5
6
7
8
9
10
11

继承关系

6，转换器 MessageConvertor

MessageConvertor 的 作用？

RabbitMQ 的消息在网络传输中需要转换成 byte[]（字节数组）进行发送，消费者需要对字节数组进行解析。

在 Spring AMQP 中，消息会被封装为 org.springframework.amqp.core.Message对象。消息的序列化和反序列化，就是处理 Message 的消息体 body 对象。

如果消息已经是 byte[]格式，就不需要转换。

如果是 String，会转换成 byte[]。

如果是 Java 对象，会使用 JDK 序列化将对象转换为 byte[]（体积大，效率差）。

在 调 用 RabbitTemplate 的 convertAndSend() 方 法 发 送 消 息 时 ， 会 使 用MessageConvertor 进行消息的序列化，默认使用 SimpleMessageConverter。

在某些情况下，我们需要选择其他的高效的序列化工具。如果我们不想在每次发送消息时自己处理消息，就可以直接定义一个 MessageConvertor。

@Bean
public RabbitTemplate rabbitTemplate(final ConnectionFactory connectionFactory) {
    final RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
    rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
    return rabbitTemplate;
}
1
2
3
4
5
6

MessageConvertor 如何 工作？

调 用 了 RabbitTemplate 的 convertAndSend() 方 法 时 会 使 用 对 应 的MessageConvertor 进行消息的序列化和反序列化。

序列化：Object —— Json —— Message(body) —— byte[]

反序列化：byte[] ——Message —— Json —— Object

有 哪些 MessageConvertor ？

在 Spring 中提供了一个默认的转换器：SimpleMessageConverter。

Jackson2JsonMessageConverter（RbbitMQ 自带）：将对象转换为 json，然后再转换成字节数组进行传递。

如何 自定义 MessageConverter ？

例如：我们要使用 Gson 格式化消息：

创建一个类，实现 MessageConverter 接口，重写 toMessage()和 fromMessage()方法。

toMessage(): Java 对象转换为 Message
fromMessage(): Message 对象转换为 Java 对象
1
2

三，SpringBoot集成RabbitMQ

为什么没有定义 Spring AMQP 的任何一个对象，也能实现消息的收发？Spring Boot 做了什么？

老套路

源码：RabbitAutoConfiguration