跳到主要内容
版本：3.5.10

AMQP

QWen Max 中英对照 AMQP

高级消息队列协议（AMQP）是一种平台无关的、面向消息中间件的线级协议。Spring AMQP 项目将核心 Spring 概念应用于基于 AMQP 的消息解决方案的开发。Spring Boot 通过 RabbitMQ 提供了多种使用 AMQP 的便利功能，包括 spring-boot-starter-amqp starter。

RabbitMQ 支持​

RabbitMQ 是一个基于 AMQP 协议的轻量级、可靠、可扩展且可移植的消息代理。Spring 通过 AMQP 协议使用 RabbitMQ 进行通信。

RabbitMQ 的配置由 spring.rabbitmq.* 中的外部配置属性控制。例如，你可以在 application.properties 中声明以下部分：

spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=secret

或者，你也可以使用 addresses 属性来配置相同的连接：

spring.rabbitmq.addresses=amqp://admin:secret@localhost
备注

以这种方式指定地址时，host 和 port 属性将被忽略。如果地址使用 amqps 协议，则会自动启用 SSL 支持。

有关更多支持的基于属性的配置选项，请参见 RabbitProperties。若要配置 Spring AMQP 所使用的 RabbitMQ ConnectionFactory 的底层细节，请定义一个 ConnectionFactoryCustomizer Bean。

如果上下文中存在一个 ConnectionNameStrategy Bean，它将被自动用于为自动配置的 CachingConnectionFactory 所创建的连接命名。

要对 RabbitTemplate 进行应用范围的、可叠加的自定义，可以使用一个 RabbitTemplateCustomizer Bean。

发送消息​

Spring 的 AmqpTemplate 和 AmqpAdmin 会自动配置，你可以直接将它们注入到你自己的 Bean 中，如下例所示：

import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.stereotype.Component;

@Component
public class MyBean {

	private final AmqpAdmin amqpAdmin;

	private final AmqpTemplate amqpTemplate;

	public MyBean(AmqpAdmin amqpAdmin, AmqpTemplate amqpTemplate) {
		this.amqpAdmin = amqpAdmin;
		this.amqpTemplate = amqpTemplate;
	}

	// ...

	public void someMethod() {
		this.amqpAdmin.getQueueInfo("someQueue");
	}

	public void someOtherMethod() {
		this.amqpTemplate.convertAndSend("hello");
	}

}
备注

RabbitMessagingTemplate 可以以类似的方式注入。如果定义了一个 MessageConverter Bean，它会自动关联到自动配置的 AmqpTemplate。

如有必要，任何定义为 bean 的 Queue 都会自动用于在 RabbitMQ 实例上声明相应的队列。

要重试操作，你可以在 AmqpTemplate 上启用重试（例如，在代理连接丢失的情况下）：

spring.rabbitmq.template.retry.enabled=true
spring.rabbitmq.template.retry.initial-interval=2s

重试默认是禁用的。你也可以通过声明一个 RabbitRetryTemplateCustomizer Bean，以编程方式自定义 RetryTemplate。

如果你需要创建更多的 RabbitTemplate 实例，或者想要覆盖默认配置，Spring Boot 提供了一个 RabbitTemplateConfigurer Bean，你可以使用它来初始化一个 RabbitTemplate，其配置与自动配置所使用的工厂相同。

向 Stream 发送消息​

要向特定的流发送消息，请指定该流的名称，如下例所示：

spring.rabbitmq.stream.name=my-stream

如果定义了 MessageConverter、StreamMessageConverter 或 ProducerCustomizer bean，它会自动关联到自动配置的 RabbitStreamTemplate。

如果你需要创建更多的 RabbitStreamTemplate 实例，或者想要覆盖默认配置，Spring Boot 提供了一个 RabbitStreamTemplateConfigurer Bean，你可以使用它来初始化一个 RabbitStreamTemplate，其配置与自动配置所使用的工厂相同。

接收消息​

当 Rabbit 基础设施存在时，任何 Bean 都可以使用 @RabbitListener 注解来创建一个监听器端点。如果没有定义 RabbitListenerContainerFactory，则会自动配置一个默认的 SimpleRabbitListenerContainerFactory，并且你可以通过 spring.rabbitmq.listener.type 属性切换到 direct 容器。如果定义了 MessageConverter 或 MessageRecoverer Bean，它们会自动与默认的工厂关联。

以下示例组件在 someQueue 队列上创建一个监听器端点：

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class MyBean {

	@RabbitListener(queues = "someQueue")
	public void processMessage(String content) {
		// ...
	}

}
提示

更多详情请参见 @EnableRabbit。

如果你需要创建更多的 RabbitListenerContainerFactory 实例，或者想要覆盖默认配置，Spring Boot 提供了 SimpleRabbitListenerContainerFactoryConfigurer 和 DirectRabbitListenerContainerFactoryConfigurer，可用于初始化 SimpleRabbitListenerContainerFactory 和 DirectRabbitListenerContainerFactory，并使其具有与自动配置所使用的工厂相同的设置。

提示

你选择哪种容器类型并不重要。这两个 bean 由自动配置暴露出来。

例如，以下配置类暴露了另一个使用特定 MessageConverter 的工厂：

import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration(proxyBeanMethods = false)
public class MyRabbitConfiguration {

	@Bean
	public SimpleRabbitListenerContainerFactory myFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer) {
		SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
		ConnectionFactory connectionFactory = getCustomConnectionFactory();
		configurer.configure(factory, connectionFactory);
		factory.setMessageConverter(new MyMessageConverter());
		return factory;
	}

	private ConnectionFactory getCustomConnectionFactory() {
		return ...
	}

}

然后你可以在任何使用 @RabbitListener 注解的方法中使用该工厂，如下所示：

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class MyBean {

	@RabbitListener(queues = "someQueue", containerFactory = "myFactory")
	public void processMessage(String content) {
		// ...
	}

}

你可以启用重试机制来处理监听器抛出异常的情况。默认情况下，使用的是 RejectAndDontRequeueRecoverer，但你也可以自定义一个 MessageRecoverer。当重试次数耗尽时，消息会被拒绝，并根据 Broker 的配置选择丢弃或路由到死信交换机（dead-letter exchange）。默认情况下，重试是禁用的。你还可以通过声明一个 RabbitRetryTemplateCustomizer Bean 来以编程方式自定义 RetryTemplate。

important

默认情况下，如果重试被禁用且监听器抛出异常，消息将被无限期地重新投递。你可以通过以下两种方式修改此行为：将 defaultRequeueRejected 属性设置为 false，以避免进行任何重新投递；或者抛出 AmqpRejectAndDontRequeueException，以表明该消息应被拒绝。当启用重试并且达到最大投递尝试次数时，就会使用后一种机制。