Spring AMQP 由 spring-amqp 和spring-rabbit 两部分组成，
spring-amqp 包含org.springframework,amqp.core 包，包含一些不依赖特定消息中间件的类。例：message,exchage,queue,binding 等类
一、连接和资源管理
管理到RabbitM broker 的连接的核心组件是ConnectionFactory接口。ConnectionFactory 接口的职责是提供org.springframework.amqp.rabbit.connection的一个实例，它是com.rabbitmq.client.Connection的包装器
提供3种类型connectionfactory ，分别是：pooledChannelConnectionFactory，ThreadChannelConnectionFactory，CachingConnectionFactory
pooledChannelConnectionFactory
该工厂基于Apache Pool2管理一个连接和两个通道（ channel）池。一个池用于事务性通道，另一个用于非事务性通道。池是具有默认配置的GenericObjectPool。
ThreadChannelConnectionFactory
此连接工厂用于消息顺序有严格要求，且未使用scoped operations 的情况，此连接工厂管理
单个连接和两个threadLocal，一个用于事务c hannel,另一个用于非事务channel，这个连接工厂确保所有的操作是在同一个
线程使用同一个channel中执行。
CachingConnectionFactory ，
此连接工厂中connection 可用被共享使，channel被缓存。根据是否使用事务使用不同的缓存，默认缓存大小是25，使用setChannelCacheSize方
法控制。
丛1.3 版本开始，可以配置cacheMode 属性为CacheMode.CONNECTION 来缓存connection，而不是channel。丛1.5.5 版本开始，connetionLimit属性控制连接数，如果达到此控制连接数，channelCheckoutTimeLimit 和channelcheckoutTimeout 控制等待连接数量和超时时间。
配置底层客户端连接工厂
spring-rabbit的连接工厂依赖Rabbit client ConnectionFactory。在spring-rabbit 工厂上设置的与Rabbit client ConnectionFactory 相等的属性会传递到Rabbit client ConnectionFactory，如果需要设置Rabbit client ConnectionFactory的其它属性，可以创建一个Rabbit client ConnectionFactory 对象 然后spring-rabbit的连接工厂依赖此对象
丛1.4版本开始提供RabbitConnectionFactoryBean 类配置ssl属性，然后需要将此类对象注入到 Rabbit client ConnectionFactory 中。从sring AMQP 1.5开始可以直接在连接工厂设置。
路由连接工厂
AbstractRoutingConnectionFactory 提供一种在几个连接工厂根据lookupKey 选择目标连接工厂的机制，spring 提供一个simpleRoutingConnectFactory 实现类，从simpleResourceHolder 中获取线程上下文信息lookupKey
LocalizedQueueConnectionFactory
当在cluster中使用HA queue 时，需要连接到lead queue ，在连接工厂中配置多个地址时，当一个地址连接失败时会尝试别外的地址。localizedQueueConnectionFactory 使用管理插件的REST API 决定lead queue.
从2.0.2版本开始，您可以将usePublisherConnection属性设置为true，以便在可能的情况下与侦听器容器使用不同的连接
二、发送消息
AmqpTemplate接口定义了发送和接收消息的所有基本操作,发送消息都这是通过此接口实现类执行的，RabbitTemplate 是常用的实现类
常用的方法有
void send(Message message) throws AmqpException; void send(String routingKey, Message message) throws AmqpException; void send(String exchange, String routingKey, Message message) throws AmqpException;
及常用带消息转换的简便方法
void convertAndSend(Object message) throws AmqpException; void convertAndSend(String routingKey, Object message) throws AmqpException; void convertAndSend(String exchange, String routingKey, Object message) throws AmqpException; void convertAndSend(Object message, MessagePostProcessor messagePostProcessor) throws AmqpException; void convertAndSend(String routingKey, Object message,MessagePostProcessor messagePostProcessor) throws AmqpException; void convertAndSend(String exchange, String routingKey, Object message,MessagePostProcessor messagePostProcessor) throws AmqpException;
三、 接收消息
接收消息大体分为两种类型：Polling Consumer 和 注册监听（Asynchronous Consumer）
Polling Consumer
在AmqpTemplate 接口中定义了接收消息相关方法， 这种方式不会自动到Mq服务取消息，需要调用方法去获取消息，且一次只能取一消息。这种方式会每一次接受消息创建一个QueueingConsume，比较费资源，所以很少使用。
AmqpTemplate接口接收消息方法，还有一此方便使用的方法。
Message receive() throws AmqpException; Message receive(String queueName) throws AmqpException; Message receive(long timeoutMillis) throws AmqpException; Message receive(String queueName, long timeoutMillis) throws AmqpException;
使用 Polling Consumer 例
@Component
public class PollReceiveMessage {
private Logger log = LoggerFactory.getLogger(PollReceiveMessage.class);
@Autowired
private AmqpTemplate amqpTemplate;
public String pollReceive() {
Object content = amqpTemplate.receiveAndConvert( "queue.poll");
log.info((String) content);
return (String)content;
}
}
注册监听（Asynchronous Consumer）
这种接收消息的方式，因为注册一个守护 程序 监听消息，也可以叫异步方式。可以通过各种方式注册 守护 程序，常用方式是 RabbitListener注解方式，每次获取多少消息通过预取参数控制，默认目前是250。这里有重要组成部分：
Message Listener
是当接收到消息后进行消费的业务代码，其需要实现MessageListener接口域ChannelAwareMessageListener
public interface MessageListener {
void onMessage(Message message);
}
public interface ChannelAwareMessageListener {
void onMessage(Message message, Channel channel) throws Exception;
监听容器（Container）
是AMQP Queue和MessageListener实例之间的 连接桥梁，目前有两种 Container，分别是SimpleMessageListenerContainer和DirectMessageListenerContainer
@Configuration
@EnableRabbit
public class AppConfig {
@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() {
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
factory.setConnectionFactory(connectionFactory());
factory.setConcurrentConsumers(3);
factory.setMaxConcurrentConsumers(10);
factory.setContainerCustomizer(container -> /* customize the container */); return factory;
}
}
容器工厂提供了添加MessagePostProcessor实例的方法，这些实例在接收消息之后(在调用侦听器之前)和发送回复之前应用。
还可以在 容器工厂上添加RetryTemplate和RecoveryCallback，当retry次数达到时会调用 RecoveryCallback。
factory.setRetryTemplate(retryTemplate);
factory.setReplyRecoveryCallback(ctx -> {
Message failed = SendRetryContextAccessor.getMessage(ctx);
Address replyTo = SendRetryContextAccessor.getAddress(ctx);
Throwable t = ctx.getLastThrowable();
... return null;
});
处理异常
默认情况下，如果带注释的侦听器方法抛出异常，则将其抛出到容器，并根据容器和代理配置对消息进行重新请求和重新交付、丢弃或路由到死信交换。不会向发送方返回任何内容。
从2.0版本开始，@RabbitListener注释有了两个新属性:errorHandler和returnExceptions
批量处理
当消息发送端发送批量消息时接收端也是批量接收消息、处理消息。从2.2版本开始支持批量接收、处理 发送端单条发送消息，通过 监听容器工厂的batchListener 属性审控制。
此外，SimpleMessageListenerContainer中还添加了一个新的属性consumerBatchEnabled。当设置为true时，容器将创建批量消息，最大为batchSize;如果receiveTimeout超时而没有新消息到达，也会处理接收到的部分批量消息。如果接收到生产者创建的批量消息，则将其 批量消息添加到消费者端批处理;因此，实际处理消息数量可能超过batchSize, 当consumerBatchEnabled为true时，deBatchingEnabled必须为true
使用RabbitListener注解注册监听
这是最常用的方法
@Component
public class ReceiveMessage {
private Logger log = LoggerFactory.getLogger(ReceiveMessage.class);
@RabbitListener(queues = "queue.test")
public void processMessage(String content) {
log.info(content);
}
}
注：通过使用RabbitListenerContainerFactory在后台为每个带 RabbitListener注解的方法创建一个消息侦听器容器。
需要在配置类上添加 @EnableRabbit注解来允许 abbitListener注解使用
使用RabbitListenerEndpoint 注册监听端点
@Configuration
@EnableRabbit
public class ProgramReceiveMessage implements RabbitListenerConfigurer {
private Logger log = LoggerFactory.getLogger(ProgramReceiveMessage.class);
@Override
public void configureRabbitListeners(RabbitListenerEndpointRegistrar registrar) {
SimpleRabbitListenerEndpoint endpoint = new SimpleRabbitListenerEndpoint();
endpoint.setId("queue.poll");
endpoint.setQueueNames("queue.poll");
endpoint.setMessageListener(message -> log.info(message.toString()));
registrar.registerEndpoint(endpoint);
}
}
使用 SimpleMessageListenerContainer接口 注册监听
@Configuration
public class ExampleAmqpConfiguration {
private Logger log = LoggerFactory.getLogger(ExampleAmqpConfiguration.class);
@Bean
public SimpleMessageListenerContainer messageListenerContainer() {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(rabbitConnectionFactory());
container.setQueueName("some.queue");
container.setMessageListener(exampleListener());
return container;
}
@Bean
public CachingConnectionFactory rabbitConnectionFactory() {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory("localhost");
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
return connectionFactory;
}
@Bean
public MessageListener exampleListener() {
return new MessageListener() {
@Override
public void onMessage(Message message) {
log.info("received: " + message);
}
};
}
}
四、发布确认和返回(Publisher Confirms and Returns)
发布确认（ Publisher Confirms)
用于确认消息能否发送到到RabbitMq交换机。 发布确认有3种类型，分别是none,correlated,simple 这三种类型，通过 ConnectionFactory 接口属性控制, ConnectionFactory 的实现类中,只有CachingConnectionFacotory支持 correlated类型 发布确认。 发布确认可以是同步确认，也可以是异步回调确认。
simple 类型发布确认
使用PooledChannelConnectionFactory 连接工厂
/**
* 需要 apache commons-pool2,不支持PublisherReturns,支持Simple Confirms
*/
@Bean
public PooledChannelConnectionFactory connectionFactory() {
ConnectionFactory clientnFactory = new ConnectionFactory();
PooledChannelConnectionFactory connectionFactory = new PooledChannelConnectionFactory(clientnFactory);
connectionFactory.setSimplePublisherConfirms(true);
return connectionFactory;
}
同步确认
@Autowired
private RabbitTemplate rabbitTemplate;
public boolean syncConfirmsSend(String message) {
// return rabbitTemplate.invoke(rabbitOperations -> {
// rabbitOperations.convertAndSend("direct.test","routingSimple",message);
// rabbitOperations.waitForConfirmsOrDie(10_00);
// return true;
// },(tag,multiple) -> {
// log.info("Ack: " + tag + ":" + multiple);
// },(tag,multiple) ->{
// log.info("Nack: " + tag + ":" + multiple);
// });
return rabbitTemplate.invoke(rabbitOperations -> {
rabbitOperations.convertAndSend("direct.test","routingSimple",message);
rabbitOperations.waitForConfirmsOrDie(10_00);
return true;
});
}
correlated 类型发布确认
通常这种类型的发布确认采用异步回调的方式进行，通过RabbitTemplate.setConfirmCallback(ConfirmCallback callback)注册ConfirmCallback函数。一个 RabbitTemplate只能有一个 ConfirmCallback回调函数。
从2.1版本开始，CorrelationData对象有一个ListenableFuture，可以使用它来获取结果，而不是在RabbitTemplate上使用ConfirmCallback
使用CachingConnectionFactory连接工厂
@Bean
public CachingConnectionFactory rabbitConnectionFactory() {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory("localhost");
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
connectionFactory.setPublisherReturns(true);
connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
return connectionFactory;
}
RabbitTemplate 配置异步确认
@Bean
public RabbitTemplate customTemplate(ConnectionFactory connectionFactory) {
RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
rabbitTemplate.setReturnsCallback(returnedMessage -> {
log.info("replycode: " + returnedMessage.getReplyCode());
log.info("replyText(): " + returnedMessage.getReplyText());
});
return rabbitTemplate;
}
发送信息
@Autowired
private RabbitTemplate rabbitTemplate;
public boolean send(String message) throws ExecutionException, InterruptedException, TimeoutException {
CorrelationData cd = new CorrelationData();
rabbitTemplate.convertAndSend("direct.test","routingSimple",message,cd);
return true;
}
发布 返回（ Publisher Returns)
用于确认消息能否发送到到队列， 正确发送到Exchange，但没有被路由到Queue才会回调。 需要将RabbitTemplate 的mandatory 属性设置为true，或者mandatory-expression 属性计算为true，还需要将CachingConnectionFactory 的publisherReturns 属性设置为true。返回信息是通过RabbitTemplat.setReturnsCallback(ReturnsCallback callback)注册的ReturnsCallbak函数
@Bean
public CachingConnectionFactory rabbitConnectionFactory() {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory("localhost");
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
connectionFactory.setPublisherReturns(true);
connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
return connectionFactory;
}
RabbitTemplate 配置
@Bean
public RabbitTemplate customTemplate(ConnectionFactory connectionFactory) {
RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
rabbitTemplate.setMandatory(true);
rabbitTemplate.setReturnsCallback(returnedMessage -> {
log.info("replycode: " + returnedMessage.getReplyCode());
log.info("replyText(): " + returnedMessage.getReplyText());
});
rabbitTemplate.setConfirmCallback((correlationData,ack,cause) ->{
log.info("correlationData" + correlationData.getId());
log.info("ack: " + ack);
log.info("cause: " + cause);
});
return rabbitTemplate;
}
public void send(String message) {
CorrelationData cd = new CorrelationData();
rabbitTemplate.convertAndSend("direct.test","routingSimple",message,cd);
}
消费者端 消息确认
消息确认针对两对象，一个是对rabbitMq服务 消耗 确认，另一个是对消息发布者 回复。
对 rabbitMq服务 消息 消耗确认
默认当接收到消息后自动回复确认信息到 rabbitMq服务，可以通过配置监听容器acknowledgeMode 属性手动回复。
监听容器 acknowledgeMode属性可以在@RabbitListener里指定ackMode来修改，"MANUAL" 是修改为手动回复。
手动回复成功（ack）
void basicAck(long deliveryTag, boolean multiple) throws IOException;
deliveryTag： 表示消息投递序号，每次消费或消息重新投递后， deliveryTag 都会增加
multiple：是否批量确认，值为true则会一次性确认所有小于当前消息 deliveryTag的消息
手动回复失败（nack）
void basicNack(long deliveryTag, boolean multiple, boolean requeue) throws IOException;
deliveryTag： 表示消息投递序号，每次消费或消息重新投递后，deliveryTag 都会增加
multiple：是否批量确认，值为true则会一次性确认所有小于当前消息deliveryTag的消息
requeue: 是否重新入队列
@Component
public class ReceiveMessage {
private Logger log = LoggerFactory.getLogger(ReceiveMessage.class);
private static final ConcurrentMap<String,Boolean> concurrentHashMap = new ConcurrentHashMap();
@RabbitListener(queues = "queue.test",ackMode = "MANUAL")
public String processMessage(Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag, @Payload String content) throws IOException {
Boolean ackflage = concurrentHashMap.get(content);
if (Boolean.TRUE.equals(ackflage)) {
log.info(content);
concurrentHashMap.remove(content);
channel.basicAck(tag, false);
return "success";
} else {
log.info(content);
concurrentHashMap.put(content, Boolean.TRUE);
channel.basicNack(tag, false, true);
return "FAIL";
}
}
}
对消息发布者回复
消息发布者回复其实就是再发送回复信息到rabiitMq 服务队列，默认是发送到原始消息 头中指定的ReplyTo 指定队列，也可以使用@SendTo注释来设置发送地址。
发送消息并接收 回复消息
@Autowired
private RabbitTemplate rabbitTemplate;
public Object send(String message) {
CorrelationData cd = new CorrelationData();
Message msg = MessageBuilder.withBody(message.getBytes()).build();
return rabbitTemplate.convertSendAndReceive("direct.test","directRoute",msg ,cd);
}
回复消息默认地址
@Component
public class ReceiveMessage {
private Logger log = LoggerFactory.getLogger(ReceiveMessage.class);
@RabbitListener(queues = "queue.test")
public String processMessage(Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag,
Message content) throws IOException ){
// do some more work
return "success";
}
SendAndReceive 方法的 默认5分秒超时，可以使用 replyTimeout 属性控制，如果没设置maddatory 属性为true，超时返回null,否则超时会抛出AmqpMessageReturnedException异常。
回复消息 @SendTo地址
@Component
public class ReceiveMessage {
private Logger log = LoggerFactory.getLogger(ReceiveMessage.class);
@RabbitListener(queues = "queue.test")
@SendTo("direct.test/routingReply")
public Message processMessage(Message content) {
// do some more work
return MessageBuilder.withBody("success".getBytes()).andProperties(content.getMessageProperties()).build();
}
@RabbitListener(queues = "queue.reply")
public void replyResponse(Message message){
String CorrelationId = message.getMessageProperties().getHeaders().get("spring_returned_message_correlation").toString();
// String CorrelationId = message.getMessageProperties().getCorrelationId();
// do some more work
}
}
@SendTo 的值是exchange/routingKey 格式。
五、消息转换器
消息转换是指将消息对象与Message对象 （ org.springframework.amqp.core.Message ）之间转换。 RabbitTemplate 中为方便提供一些对 消息对象与Message对象的简便方法。
消息转换接口，spring 提供几个其实现类。
public interface MessageConverter {
Message toMessage(Object object, MessageProperties messageProperties)
throws MessageConversionException;
Object fromMessage(Message message) throws MessageConversionException;
}
SimpleMessageConverter
MessageConverter的实现类SimpleMessageConverter是默认的 消息转换器。能处文体字符串、java 序列化对象和byte数组。
Jackson2JsonMessageConverter
使用 Jackson2JsonMessageConverter
@Bean
public Jackson2JsonMessageConverter jsonMessageConverter() {
Jackson2JsonMessageConverter jsonConverter = new Jackson2JsonMessageConverter();
jsonConverter.setClassMapper(classMapper());
return jsonConverter;
}
六、修改消息
spring 提供了许多扩展点。它们允许你在消息发送到RabbitMQ之前或者在消息被接收后立即对消息执行一些处理，
1、使用带MessagePostProcessor 参数的AmqpTemplate 接收和发送方法
2、 AmqpTemplate 设置属性setBeforePublishPostProcessors() and setAfterReceivePostProcessors()
七、异常处理
在 org.springframework.amqp 中所有的异常都是AmqpException 异常的子类,监听（Listenner)抛出的异常是包裹在 ListenerExecutionFailedException异常中， 丛1.3.2版本开始默认异常处理器ConditionalRejectingErrorHandler 对带不可恢复的错误异常不会重新入队列（也叫fatal exception), 从1.6.3版本开始，将用户异常添加到致命列表的简便方法是创建ConditionalRejectingErrorHandler.DefaultExceptionStrategy的子类，将覆盖isUserCauseFatal(Throwable cause)方法，对需要添加到致命异常返回true。对非 fatal exception，默认消息重新入队列，当设置属性defaultRequeueRejected为 false时，消息被废弃，不会重新入队列，
默认重新入队列的行为可能会导致永无止境的循环执行，为解决这问题，可以在监听器上添加StatefulRetryOperationsInterceptor 拦截器，此拦截器带个MessageRecoverer, 此MessageRecoverer 抛出AmqpRejectAndDontRequeueException或 ImmediateRequeueAmqpException 异常，此MessageRecoverer 在retry次数达到时调用。
AmqpRejectAndDontRequeueException。消息不会重新入队列，无论defaultrequeuereejded属性的设置如何。
ImmediateRequeueAmqpException 消息都将被重新入队列，无论defaultrequeuereejject属性的设置如何，从1.3版本开始，提供了一个新的republishmessagercoverer，允许在重试失败后发布失败的消息。
@Bean
RetryOperationsInterceptor interceptor() {
return RetryInterceptorBuilder.stateless()
.maxAttempts(5)
.recoverer(new RepublishMessageRecoverer(amqpTemplate(), "something", "somethingelse"))
.build();
}
对一此同步执行操作，可以捕获并响应异常，也可以使用spring retry ,Spring AMQP提供了两个创建Spring retry 拦截器的bean StatefulRetryOperationsInterceptor 和 StatelessRetryOperationsInterceptor
@Bean
public StatefulRetryOperationsInterceptor interceptor() {
return RetryInterceptorBuilder.stateful()
.maxAttempts(5)
.backOffOptions(1000, 2.0, 10000) // initialInterval, multiplier, maxInterval
.build();
}
八、事务
Spring Rabbit框架支持同步和异步用例中的自动事务管理，在RabbitTemplate和SimpleMessageListenerContainer中，都有一个channelTransacted标志，如果为true，则告诉框架使用事务通道，也可以 提供一个PlatformTransactionManager的实现类作为外部事务管理，作为正在进行的操作的上下文。
@Transactional
public void doSomething() {
String incoming = rabbitTemplate.receiveAndConvert();
// do some more database processing...
String outgoing = processInDatabaseAndExtractReply(incoming);
rabbitTemplate.convertAndSend(outgoing);
}
@Configuration
public class ExampleExternalTransactionAmqpConfiguration {
@Bean
public SimpleMessageListenerContainer messageListenerContainer() {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(rabbitConnectionFactory());
container.setTransactionManager(transactionManager());
container.setChannelTransacted(true);
container.setQueueName("some.queue");
container.setMessageListener(exampleListener());
return container;
}
}
注：1、RabbitTransactionManager 没提供XA transactions功能
2、rabbitMQ事务与其它事务（例如数据库）之间同步是“Best Effort One Phase Commit”，即有可能实际提交时失败，而且时只记录一条日志，不会抛出异常去回滚事务。可以使用提交事务的线程上调用ConnectionUtils.checkAfterCompletion() 来检查。
条件回滚（Conditional Rollback）
@Bean
public AbstractMessageListenerContainer container() {
...
container.setTransactionManager(transactionManager);
RuleBasedTransactionAttribute transactionAttribute = new ListenerFailedRuleBasedTransactionAttribute();
transactionAttribute.setRollbackRules(Collections.singletonList( new NoRollbackRuleAttribute(DontRollBackException.class)));
container.setTransactionAttribute(transactionAttribute);
...
}
来自 “ ITPUB博客 ” ，链接：https://blog.itpub.net/23748560/viewspace-3000673/ ，如需转载，请注明出处，否则将追究法律责任。
