`AmqpTemplate`

与 Spring 框架和相关项目提供的许多其他高级抽象一样， Spring AMQP 提供了一个起着核心作用的“模板”。定义主要操作的接口称为。这些操作涵盖了发送和接收消息的一般行为。换句话说，它们对于任何实现都不是唯一的 — 因此名称中的“AMQP”。另一方面，该接口的实现与 AMQP 协议的实现相关联。与 JMS 不同，JMS 本身是一个接口级 API，而 AMQP 是一个线级协议。该协议的实现提供自己的 Client 端库，因此 template 接口的每个实现都依赖于特定的 Client 端库。目前，只有一个实现：. 在下面的示例中，我们经常使用 . 但是，当您查看实例化模板或调用 setter 的配置示例或任何代码摘录时，您可以看到实现类型（例如，）。AmqpTemplateRabbitTemplateAmqpTemplateRabbitTemplatespring-doc.cn

如前所述，该接口定义了发送和接收消息的所有基本操作。我们将在发送消息和接收消息中分别探讨消息发送和接收。AmqpTemplatespring-doc.cn

另请参见 Async Rabbit Template。spring-doc.cn

添加重试功能

从版本 1.3 开始，您现在可以将配置为使用来帮助处理代理连接问题。有关完整信息，请参见spring-retry项目。以下只是一个使用指数回退策略和 default 的示例，该策略在将异常引发给调用方之前进行三次尝试。RabbitTemplateRetryTemplateSimpleRetryPolicyspring-doc.cn

以下示例使用 XML 命名空间：spring-doc.cn

<rabbit:template id="template" connection-factory="connectionFactory" retry-template="retryTemplate"/>

<bean id="retryTemplate" class="org.springframework.retry.support.RetryTemplate">
    <property name="backOffPolicy">
        <bean class="org.springframework.retry.backoff.ExponentialBackOffPolicy">
            <property name="initialInterval" value="500" />
            <property name="multiplier" value="10.0" />
            <property name="maxInterval" value="10000" />
        </bean>
    </property>
</bean>

以下示例使用 Java 中的注释：@Configurationspring-doc.cn

@Bean
public RabbitTemplate rabbitTemplate() {
    RabbitTemplate template = new RabbitTemplate(connectionFactory());
    RetryTemplate retryTemplate = new RetryTemplate();
    ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
    backOffPolicy.setInitialInterval(500);
    backOffPolicy.setMultiplier(10.0);
    backOffPolicy.setMaxInterval(10000);
    retryTemplate.setBackOffPolicy(backOffPolicy);
    template.setRetryTemplate(retryTemplate);
    return template;
}

从版本 1.4 开始，除了属性之外，该选项还支持 . 它用作 .retryTemplaterecoveryCallbackRabbitTemplateRetryTemplate.execute(RetryCallback<T, E> retryCallback, RecoveryCallback<T> recoveryCallback)spring-doc.cn

这在一定程度上受到限制，因为重试上下文仅包含字段。对于更复杂的用例，您应该使用 external，以便您可以通过上下文的属性将其他信息传达给。以下示例显示了如何执行此操作：RecoveryCallbacklastThrowableRetryTemplateRecoveryCallback

retryTemplate.execute(
    new RetryCallback<Object, Exception>() {

        @Override
        public Object doWithRetry(RetryContext context) throws Exception {
            context.setAttribute("message", message);
            return rabbitTemplate.convertAndSend(exchange, routingKey, message);
        }

    }, new RecoveryCallback<Object>() {

        @Override
        public Object recover(RetryContext context) throws Exception {
            Object message = context.getAttribute("message");
            Throwable t = context.getLastThrowable();
            // Do something with message
            return null;
        }
    });
}

在这种情况下，您不会将 a 注入 .RetryTemplateRabbitTemplatespring-doc.cn

发布是异步的 — 如何检测成功和失败

发布消息是一种异步机制，默认情况下，RabbitMQ 会丢弃无法路由的消息。要成功发布，您可以接收异步确认，如 Correlated Publisher Confirms and Returns 中所述。请考虑两种故障情况：spring-doc.cn

发布到 Exchange，但没有匹配的目标队列。spring-doc.cn
发布到不存在的 Exchange。spring-doc.cn

第一种情况由发布者返回涵盖，如相关发布者确认和返回中所述。spring-doc.cn

对于第二种情况，消息被丢弃，并且不会生成任何返回。底层通道因异常而关闭。默认情况下，会记录此异常，但您可以向注册以获取此类事件的通知。以下示例演示如何添加：ChannelListenerCachingConnectionFactoryConnectionListenerspring-doc.cn

this.connectionFactory.addConnectionListener(new ConnectionListener() {

    @Override
    public void onCreate(Connection connection) {
    }

    @Override
    public void onShutDown(ShutdownSignalException signal) {
        ...
    }

});

您可以检查信号的属性以确定发生的问题。reasonspring-doc.cn

要检测发送线程上的异常，您可以在上检测到异常。但是，事务会严重影响性能，因此在仅为这一个用例启用事务之前，请仔细考虑这一点。setChannelTransacted(true)RabbitTemplatetxCommit()spring-doc.cn

相关发布者确认并返回

的实现支持 publisher 确认并返回。RabbitTemplateAmqpTemplatespring-doc.cn

对于返回的消息，必须将模板的属性设置为或必须为特定消息的计算结果。此功能需要 a 的，其属性设置为（请参阅发布者确认并返回）。返回值通过调用注册 . 回调必须实现以下方法：mandatorytruemandatory-expressiontrueCachingConnectionFactorypublisherReturnstrueRabbitTemplate.ReturnsCallbacksetReturnsCallback(ReturnsCallback callback)spring-doc.cn

void returnedMessage(ReturnedMessage returned);

具有以下属性：ReturnedMessagespring-doc.cn

message- 返回的消息本身spring-doc.cn
replyCode- 指示退货原因的代码spring-doc.cn
replyText- 退货的文本原因 - 例如NO_ROUTEspring-doc.cn
exchange- 消息发送到的 Exchangespring-doc.cn
routingKey- 使用的路由密钥spring-doc.cn

每个只支持 1 个。另请参阅回复超时。ReturnsCallbackRabbitTemplatespring-doc.cn

对于 publisher confirms（也称为 publisher 确认），模板需要 a 的 property 设置为。确认通过调用注册 a 发送到客户端。回调必须实现此方法：CachingConnectionFactorypublisherConfirmConfirmType.CORRELATEDRabbitTemplate.ConfirmCallbacksetConfirmCallback(ConfirmCallback callback)spring-doc.cn

void confirm(CorrelationData correlationData, boolean ack, String cause);

这是客户端在发送原始消息时提供的对象。对于 an 为 true，对于为 false。例如，原因可能包含的原因，如果在生成时可用。例如，当向不存在的 exchange 发送消息时。在这种情况下，代理将关闭通道。关闭的原因包含在中。这是在 1.4 版本中添加的。CorrelationDataackacknacknacknacknackcausecausespring-doc.cn

只有一个受 .ConfirmCallbackRabbitTemplatespring-doc.cn

当 rabbit template send 操作完成时，通道将关闭。当连接工厂缓存已满时，这排除了接收 confirms 或 returns （当缓存中有空间时，通道未物理关闭，返回和确认正常进行）。当缓存已满时，框架会将关闭时间最多推迟 5 秒，以便有时间接收确认和返回。使用 confirm 时，当收到最后一次确认时，通道将关闭。当仅使用返回时，通道将保持打开状态整整 5 秒。我们通常建议将 connection factory's 设置为足够大的值，以便将发布消息的通道返回到缓存，而不是关闭。您可以使用 RabbitMQ 管理插件监控通道使用情况。如果您看到通道正在快速打开和关闭，则应考虑增加高速缓存大小以减少服务器上的开销。channelCacheSize

在版本 2.1 之前，在收到确认之前，为发布者确认启用的渠道会返回到缓存中。其他一些进程可以签出通道并执行一些导致通道关闭的操作 — 例如将消息发布到不存在的 exchange。这可能会导致确认丢失。版本 2.1 及更高版本在确认未完成时不再将通道返回到缓存。每次操作后，都会在 channel 上执行 logical 操作。通常，这意味着一次只有一个确认在通道上未完成。RabbitTemplateclose()

从版本 2.2 开始，回调是在连接工厂的线程之一上调用的。这是为了避免在从回调中执行 Rabbit 操作时出现潜在的死锁。在以前的版本中，回调直接在连接 I/O 线程上调用;如果您执行某些 RPC 操作（例如打开新通道），这将死锁，因为 I/O 线程阻塞等待结果，但结果需要由 I/O 线程本身处理。对于这些版本，有必要将工作（例如发送消息）移交给回调中的另一个线程。这不再是必需的，因为框架现在将回调调用移交给执行程序。executoramqp-client

只要返回回调在 60 秒或更短的时间内执行，就仍然保证在 ack 之前收到返回的消息。确认计划在返回回传退出后或 60 秒后传递，以先到者为准。

该对象具有 a 可用于获取结果，而不是在模板上使用 a。以下示例演示如何配置实例：CorrelationDataCompletableFutureConfirmCallbackCorrelationDataspring-doc.cn

CorrelationData cd1 = new CorrelationData();
this.templateWithConfirmsEnabled.convertAndSend("exchange", queue.getName(), "foo", cd1);
assertTrue(cd1.getFuture().get(10, TimeUnit.SECONDS).isAck());
ReturnedMessage = cd1.getReturn();
...

由于它是一个，因此您可以在准备就绪时显示结果，也可以将其用于异步回调。该对象是一个具有 2 个属性的简单 bean：和（例如）。未为 broker 生成的实例填充原因。它是为框架生成的实例填充的（例如，在实例未完成时关闭连接）。CompletableFuture<Confirm>get()whenComplete()Confirmackreasonnacknacknackackspring-doc.cn

此外，当同时启用 confirms 和 returns 时，如果无法将返回的消息路由到任何队列，则该属性将填充返回的消息。可以保证在使用 . 返回 A with properties：CorrelationDatareturnackCorrelationData.getReturn()ReturnMessagespring-doc.cn

message （返回的消息）spring-doc.cn
回复代码spring-doc.cn
回复文本spring-doc.cn
交换spring-doc.cn
routingKeyspring-doc.cn

另请参阅 Scoped Operations 以获取等待发布者确认的更简单机制。spring-doc.cn

作用域操作

通常，在使用模板时，会从缓存中签出（或创建） a ，用于操作，并返回到缓存以供重用。在多线程环境中，不能保证下一个操作使用相同的通道。但是，有时您可能希望对通道的使用进行更多控制，并确保在同一通道上执行许多操作。Channelspring-doc.cn

从版本 2.0 开始，提供了一个名为的新方法，其中包含一个 . 在 callback 范围内和提供的参数上执行的任何操作都使用相同的 dedicated ，该操作将在结束时关闭（不会返回到缓存）。如果渠道是，则在收到所有确认后，它会返回到缓存中（请参阅 Correlated Publisher Confirms and Returns）。invokeOperationsCallbackRabbitOperationsChannelPublisherCallbackChannelspring-doc.cn

@FunctionalInterface
public interface OperationsCallback<T> {

    T doInRabbit(RabbitOperations operations);

}

您可能需要这样做的一个例子是，如果您希望在底层 . Spring API 以前没有公开此方法，因为如前所述，通道通常是缓存和共享的。现在提供和，它们委托给在 . 出于显而易见的原因，这些方法不能在该范围之外使用。waitForConfirms()ChannelRabbitTemplatewaitForConfirms(long timeout)waitForConfirmsOrDie(long timeout)OperationsCallbackspring-doc.cn

请注意，其他位置提供了用于将 Confirm 与 requests 相关联的更高级别抽象（请参阅 Correlated Publisher Confirms and Returns）。如果您只想等待 Broker 确认送达，则可以使用以下示例中所示的技术：spring-doc.cn

Collection<?> messages = getMessagesToSend();
Boolean result = this.template.invoke(t -> {
    messages.forEach(m -> t.convertAndSend(ROUTE, m));
    t.waitForConfirmsOrDie(10_000);
    return true;
});

如果您希望在 scope of 的同一通道上调用操作，则必须使用用于操作的相同通道来构造 admin。RabbitAdminOperationsCallbackRabbitTemplateinvokespring-doc.cn

如果模板操作已经在现有事务的范围内执行，例如，在事务处理侦听器容器线程上运行并在事务处理模板上执行操作时，则前面的讨论没有意义。在这种情况下，将在该通道上执行操作，并在线程返回到容器时提交。在这种情况下，没有必要使用。invoke

以这种方式使用 confirms 时，实际上并不需要为将 confirm 与请求相关联而设置的大部分基础设施（除非还启用了 return）。从版本 2.2 开始，连接工厂支持一个名为 . 当此设置为时，将避免使用基础设施，并且确认处理可以更高效。publisherConfirmTypeConfirmType.SIMPLEspring-doc.cn

此外，还在 sent message 中设置属性。如果您希望检查（或记录或以其他方式使用）特定的确认，您可以使用重载的方法来实现，如下例所示：RabbitTemplatepublisherSequenceNumberMessagePropertiesinvokespring-doc.cn

public <T> T invoke(OperationsCallback<T> action, com.rabbitmq.client.ConfirmCallback acks,
        com.rabbitmq.client.ConfirmCallback nacks);

这些对象（for 和 instances）是 Rabbit 客户端回调，而不是模板回调。ConfirmCallbackacknack

以下示例日志和实例：acknackspring-doc.cn

Collection<?> messages = getMessagesToSend();
Boolean result = this.template.invoke(t -> {
    messages.forEach(m -> t.convertAndSend(ROUTE, m));
    t.waitForConfirmsOrDie(10_000);
    return true;
}, (tag, multiple) -> {
        log.info("Ack: " + tag + ":" + multiple);
}, (tag, multiple) -> {
        log.info("Nack: " + tag + ":" + multiple);
}));

作用域操作绑定到线程。有关多线程环境中严格排序的讨论，请参阅多线程环境中的严格消息排序。

多线程环境中的严格消息排序

作用域内操作中的讨论仅在对同一线程执行操作时适用。spring-doc.cn

请考虑以下情况：spring-doc.cn

thread-1将消息发送到队列，并将工作移交给thread-2spring-doc.cn
thread-2将消息发送到同一队列spring-doc.cn

由于 RabbitMQ 的异步性质和缓存通道的使用;不确定是否会使用相同的通道，因此无法保证消息到达队列的顺序。（在大多数情况下，它们会按顺序到达，但无序送达的概率不为零）。要解决此用例，您可以使用大小为 size（与 a 一起使用）的有界通道缓存，以确保消息始终在同一通道上发布，并且保证顺序。为此，如果连接工厂有其他用途（例如使用者），则应为模板使用专用连接工厂，或将模板配置为使用嵌入在主连接工厂中的 publisher 连接工厂（请参阅使用单独的连接）。1channelCheckoutTimeoutspring-doc.cn

这最好用一个简单的 Spring Boot 应用程序来说明：spring-doc.cn

@SpringBootApplication
public class Application {

	private static final Logger log = LoggerFactory.getLogger(Application.class);

	public static void main(String[] args) {
		SpringApplication.run(Application.class, args);
	}

	@Bean
	TaskExecutor exec() {
		ThreadPoolTaskExecutor exec = new ThreadPoolTaskExecutor();
		exec.setCorePoolSize(10);
		return exec;
	}

	@Bean
	CachingConnectionFactory ccf() {
		CachingConnectionFactory ccf = new CachingConnectionFactory("localhost");
		CachingConnectionFactory publisherCF = (CachingConnectionFactory) ccf.getPublisherConnectionFactory();
		publisherCF.setChannelCacheSize(1);
		publisherCF.setChannelCheckoutTimeout(1000L);
		return ccf;
	}

	@RabbitListener(queues = "queue")
	void listen(String in) {
		log.info(in);
	}

	@Bean
	Queue queue() {
		return new Queue("queue");
	}


	@Bean
	public ApplicationRunner runner(Service service, TaskExecutor exec) {
		return args -> {
			exec.execute(() -> service.mainService("test"));
		};
	}

}

@Component
class Service {

	private static final Logger LOG = LoggerFactory.getLogger(Service.class);

	private final RabbitTemplate template;

	private final TaskExecutor exec;

	Service(RabbitTemplate template, TaskExecutor exec) {
		template.setUsePublisherConnection(true);
		this.template = template;
		this.exec = exec;
	}

	void mainService(String toSend) {
		LOG.info("Publishing from main service");
		this.template.convertAndSend("queue", toSend);
		this.exec.execute(() -> secondaryService(toSend.toUpperCase()));
	}

	void secondaryService(String toSend) {
		LOG.info("Publishing from secondary service");
		this.template.convertAndSend("queue", toSend);
	}

}

即使发布是在两个不同的线程上执行的，它们也将使用相同的通道，因为缓存的上限是单个通道。spring-doc.cn

从版本 2.3.7 开始，支持使用 and 方法将线程的通道转移到另一个线程。第一个方法返回一个上下文，该上下文将传递给调用第二个方法的第二个线程。线程可以具有绑定到非事务通道或事务通道（或每个通道中的一个）的 URL;除非使用两个连接工厂，否则无法单独传输它们。示例如下：ThreadChannelConnectionFactoryprepareContextSwitchswitchContextspring-doc.cn

@SpringBootApplication
public class Application {

	private static final Logger log = LoggerFactory.getLogger(Application.class);

	public static void main(String[] args) {
		SpringApplication.run(Application.class, args);
	}

	@Bean
	TaskExecutor exec() {
		ThreadPoolTaskExecutor exec = new ThreadPoolTaskExecutor();
		exec.setCorePoolSize(10);
		return exec;
	}

	@Bean
	ThreadChannelConnectionFactory tccf() {
		ConnectionFactory rabbitConnectionFactory = new ConnectionFactory();
		rabbitConnectionFactory.setHost("localhost");
		return new ThreadChannelConnectionFactory(rabbitConnectionFactory);
	}

	@RabbitListener(queues = "queue")
	void listen(String in) {
		log.info(in);
	}

	@Bean
	Queue queue() {
		return new Queue("queue");
	}


	@Bean
	public ApplicationRunner runner(Service service, TaskExecutor exec) {
		return args -> {
			exec.execute(() -> service.mainService("test"));
		};
	}

}

@Component
class Service {

	private static final Logger LOG = LoggerFactory.getLogger(Service.class);

	private final RabbitTemplate template;

	private final TaskExecutor exec;

	private final ThreadChannelConnectionFactory connFactory;

	Service(RabbitTemplate template, TaskExecutor exec,
			ThreadChannelConnectionFactory tccf) {

		this.template = template;
		this.exec = exec;
		this.connFactory = tccf;
	}

	void mainService(String toSend) {
		LOG.info("Publishing from main service");
		this.template.convertAndSend("queue", toSend);
		Object context = this.connFactory.prepareSwitchContext();
		this.exec.execute(() -> secondaryService(toSend.toUpperCase(), context));
	}

	void secondaryService(String toSend, Object threadContext) {
		LOG.info("Publishing from secondary service");
		this.connFactory.switchContext(threadContext);
		this.template.convertAndSend("queue", toSend);
		this.connFactory.closeThreadChannel();
	}

}

一旦被调用，如果当前线程执行了任何其他操作，它们将在新的通道上执行。当不再需要线程绑定通道时，关闭它很重要。prepareSwitchContext

消息传递集成

从版本 1.4 开始，（构建在）之上，提供了与 Spring Framework 消息传递抽象的集成，即 . 这允许您使用抽象来发送和接收消息。其他 Spring 项目（例如 Spring 集成和 Spring 的 STOMP 支持）也使用这种抽象。涉及两个消息转换器：一个用于在 spring-messaging 和 Spring AMQP 的抽象之间进行转换，另一个用于在 Spring AMQP 的抽象和底层 RabbitMQ 客户端库所需的格式之间进行转换。默认情况下，消息有效负载由提供的实例的消息转换器转换。或者，您可以使用其他一些有效负载转换器注入自定义，如下例所示：RabbitMessagingTemplateRabbitTemplateorg.springframework.messaging.Messagespring-messagingMessage<?>Message<?>MessageMessageRabbitTemplateMessagingMessageConverterspring-doc.cn

MessagingMessageConverter amqpMessageConverter = new MessagingMessageConverter();
amqpMessageConverter.setPayloadConverter(myPayloadConverter);
rabbitMessagingTemplate.setAmqpMessageConverter(amqpMessageConverter);

验证的用户 ID

从版本 1.6 开始，模板现在支持（ when using Java configuration）。如果发送消息，则在评估此表达式后设置 user id 属性（如果尚未设置）。评估的根对象是要发送的消息。user-id-expressionuserIdExpressionspring-doc.cn

以下示例演示如何使用该属性：user-id-expressionspring-doc.cn

<rabbit:template ... user-id-expression="'guest'" />

<rabbit:template ... user-id-expression="@myConnectionFactory.username" />

第一个示例是 Literal 表达式。第二个选项从应用程序上下文中的连接工厂 Bean 获取属性。usernamespring-doc.cn

使用单独的连接

从版本 2.0.2 开始，你可以将该属性设置为尽可能使用与侦听器容器使用的连接不同的连接。这是为了避免当生产者因任何原因被阻止时，使用者被阻止。为此，连接工厂维护第二个内部连接工厂;默认情况下，它与主工厂的类型相同，但如果您希望使用不同的工厂类型进行发布，则可以显式设置此类型。如果 rabbit 模板在侦听器容器启动的事务中运行，则无论此设置如何，都会使用容器的通道。usePublisherConnectiontruespring-doc.cn

通常，您不应将 a 与将此设置为的模板一起使用。使用采用连接工厂的构造函数。如果使用采用模板的其他构造函数，请确保模板的值为。这是因为，通常使用 admin 来声明侦听器容器的队列。使用将属性设置为的模板意味着将在与侦听器容器使用的连接不同的连接上声明独占队列（例如）。在这种情况下，容器不能使用队列。RabbitAdmintrueRabbitAdminfalsetrueAnonymousQueue