SpringAMQP详解

一，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";@Beanpublic ConnectionFactory factory() {CachingConnectionFactory factory = new CachingConnectionFactory();factory.setAddresses("121.199.31.160");factory.setPort(5672);factory.setUsername("root");factory.setPassword("root");return factory;}@Beanpublic 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;}
}

为什么我们在配置文件（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()]));

3，Message

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

4，RabbitTemplate 消息模板

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

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

    /*** return callback   &&  confirm callable** @param factory* @return*/@Beanpublic 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;}

5，Messager Listener 消息监听

MessageListener

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

MessageListenerContainer

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

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

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

    /*** 消息监听器容器* @param connectionFactory* @return*/@Beanpublic 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;}

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

MessageListenerContainerFactory

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

    /*** * @param connectionFactory* @return*/@Beanpublic SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();factory.setConnectionFactory(connectionFactory);factory.setMessageConverter(new Jackson2JsonMessageConverter());factory.setAcknowledgeMode(AcknowledgeMode.NONE);factory.setAutoStartup(true);return factory;}

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

@Slf4j
@Component
@PropertySource("classpath:application.properties")
@RabbitListener(queues = "${amqp.yhd.queue}", containerFactory = "rabbitListenerContainerFactory")
public class FirstConsumer {@RabbitHandlerpublic void process(@Payload String message) {log.info("First Queue received msg : {}", message);}
}

继承关系

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;
}

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 对象

三，SpringBoot集成RabbitMQ

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

老套路

源码：RabbitAutoConfiguration