跳到主要内容
版本：4.1.1

AmqpTemplate

Doubao 中英对照 AmqpTemplate

与 Spring Framework 及相关项目提供的许多其他高级抽象一样，Spring AMQP 提供了一个起核心作用的“模板”。定义主要操作的接口名为 AmqpTemplate。这些操作涵盖发送和接收消息的一般行为。换句话说，它们并非任何实现所独有——因此名称中有“AMQP”。另一方面，该接口也有一些与 AMQP 协议实现绑定的实现。与 JMS 本身就是接口级 API 不同，AMQP 是线级协议。该协议的实现提供自己的客户端库，因此模板接口的每个实现都依赖于特定的客户端库。目前只有一个实现：RabbitTemplate。在下面的示例中，我们经常使用 AmqpTemplate。不过，当你查看配置示例或任何实例化模板或调用 setter 的代码摘录时，就能看到实现类型（例如 RabbitTemplate）。

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

另请参见异步 Rabbit 模板。

添加重试能力​

从 1.3 版本开始，你现在可以配置 RabbitTemplate 使用 RetryTemplate 来帮助处理 broker 连接问题。更多信息参见 Spring Framework 中的核心 Retry 支持。下面只是一个使用指数退避策略和默认 SimpleRetryPolicy 的示例，它在向调用者抛出异常之前尝试三次。

下例在 Java 中使用 @Configuration 注解：

public RabbitTemplate rabbitTemplate() {
    RabbitTemplate template = new RabbitTemplate(connectionFactory());
    RetryPolicy retryPolicy = RetryPolicy.builder()
            .delay(Duration.ofMillis(500))
            .multiplier(2.0)
            .maxDelay(Duration.ofSeconds(10))
            .build();
    template.setRetryTemplate(new RetryTemplate(retryPolicy));
    return template;
}

从 1.4 版本开始，除了 retryTemplate 属性之外，RabbitTemplate 上还支持 recoveryCallback 选项。

备注

RecoveryCallback 有一定局限，即重试上下文只包含 lastThrowable 字段。对于更复杂的用例，你应使用外部的 RetryTemplate。下例展示了如何这样做：

retryTemplate.setRetryListener(new RetryListener() {

    @Override
    public void onRetryPolicyExhaustion(RetryPolicy retryPolicy, Retryable<?> retryable, RetryException ex) {
        Throwable lastException = ex.getLastException();
        // Do something with message
    }

});

retryTemplate.invoke(() -> rabbitTemplate.convertAndSend(exchange, routingKey, message));

在这种情况下，你不会向 RabbitTemplate 注入 RetryTemplate。更多信息参见弹性特性。

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

发布消息是一种异步机制，默认情况下，无法路由的消息会被 RabbitMQ 丢弃。对于成功发布，你可以收到异步确认，如关联的发布确认与退回所述。考虑两种失败场景：

  • 发布到某个交换机，但没有匹配的目标队列。
  • 发布到不存在的交换机。

第一种情况由发布者退回覆盖，如关联的发布确认与退回所述。

对于第二种情况，消息会被丢弃，不会产生退回。底层 channel 会因异常被关闭。默认情况下，该异常会被记录日志，但你可以向 CachingConnectionFactory 注册一个 ChannelListener 来获取此类事件的通知。下例展示了如何添加 ConnectionListener：

this.connectionFactory.addConnectionListener(new ConnectionListener() {

    @Override
    public void onCreate(Connection connection) {
    }

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

});

你可以检查信号的 reason 属性来确定发生的问题。

要在发送线程上检测该异常，你可以在 RabbitTemplate 上 setChannelTransacted(true)，异常会在 txCommit() 时被检测到。不过，事务会显著影响性能，因此请在仅为这一个用例启用事务之前仔细考虑。

关联的发布确认与退回​

AmqpTemplate 的 RabbitTemplate 实现支持发布确认和退回。

对于被退回的消息，模板的 mandatory 属性必须设置为 true，或者 mandatory-expression 必须对某条消息求值为 true。此功能要求 CachingConnectionFactory 的 publisherReturns 属性设置为 true（参见发布确认与退回）。退回通过模板调用 setReturnsCallback(ReturnsCallback callback) 注册 RabbitTemplate.ReturnsCallback 发送给客户端。该回调必须实现以下方法：

void returnedMessage(ReturnedMessage returned);

ReturnedMessage 具有以下属性：

  • message——被退回的消息本身
  • replyCode——指示退回原因的代码
  • replyText——退回原因的文本说明——例如 NO_ROUTE
  • exchange——消息被发送到的交换机
  • routingKey——所使用的路由键

每个 RabbitTemplate 只支持一个 ReturnsCallback。另请参见回复超时。

对于发布确认（也称发布者应答），模板要求 CachingConnectionFactory 的 publisherConfirm 属性设置为 ConfirmType.CORRELATED。确认通过模板调用 setConfirmCallback(ConfirmCallback callback) 注册 RabbitTemplate.ConfirmCallback 发送给客户端。该回调必须实现此方法：

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

CorrelationData 是客户端发送原始消息时提供的对象。ack 为 true 表示 ack，false 表示 nack。对于 nack 实例，cause 可能包含 nack 的原因（如果在生成 nack 时可获得）。一个例子是向不存在的交换机发送消息。在那种情况下，broker 会关闭 channel。关闭原因包含在 cause 中。cause 是 1.4 版本新增的。

一个 RabbitTemplate 只支持一个 ConfirmCallback。

备注

当 rabbit 模板发送操作完成时，channel 会被关闭。当连接工厂缓存已满时，这会阻碍接收确认或退回（当缓存中有空间时，channel 不会被物理关闭，退回和确认正常进行）。当缓存满时，框架会将关闭操作推迟最多五秒，以便有时间接收确认和退回。使用确认时，收到最后一个确认后 channel 关闭。仅使用退回时，channel 会在整整五秒内保持打开。我们通常建议把连接工厂的 channelCacheSize 设置为足够大的值，使发布消息的 channel 被归还到缓存而不是被关闭。你可以使用 RabbitMQ 管理插件监控 channel 使用情况。如果看到 channel 频繁打开和关闭，应考虑增大缓存大小以减少服务器上的开销。

important

在 2.1 版本之前，启用发布确认的 channel 会在收到确认之前归还到缓存。其他进程可能检出该 channel 并执行导致 channel 关闭的操作——例如向不存在的交换机发布消息。这可能导致确认丢失。2.1 及更高版本不再在确认未决时把 channel 归还到缓存。RabbitTemplate 在每次操作后对 channel 执行逻辑上的 close()。通常，这意味着一个 channel 上一次只有一个未决确认。

备注

从 2.2 版本开始，回调在连接工厂的某个 executor 线程上调用。这是为了避免你在回调内部执行 Rabbit 操作时可能发生的死锁。在以前的版本中，回调直接在 amqp-client 连接 I/O 线程上调用；如果你执行某些 RPC 操作（例如打开新 channel），这会导致死锁，因为 I/O 线程阻塞等待结果，而结果又需要由 I/O 线程本身处理。在那些版本中，必须在回调内部把工作（例如发送消息）交接给另一个线程。现在不再需要这样做，因为框架会把回调调用交接给执行器。

important

只要退回回调在 60 秒或更短时间内执行，先于确认收到退回消息的保证仍然保持。确认被安排在退回回调退出后或 60 秒后投递，以先到者为准。

CorrelationData 对象有一个 CompletableFuture，你可以用它获取结果，而不必在模板上使用 ConfirmCallback。下例展示了如何配置 CorrelationData 实例：

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

由于它是一个 CompletableFuture<Confirm>，你可以在结果就绪时 get()，也可以使用 whenComplete() 进行异步回调。Confirm 对象是一个带有 2 个属性的简单 bean：ack 和 reason（用于 nack 实例）。对于 broker 生成的 nack 实例，reason 不会被填充。对于框架生成的 nack 实例（例如在 ack 实例未决时关闭连接），reason 会被填充。

此外，当确认和退回都启用时，如果消息无法路由到任何队列，CorrelationData 的 return 属性会被填充为被退回的消息。可以保证被退回消息属性在 future 被设置为 ack 之前设置。CorrelationData.getReturn() 返回一个 ReturnMessage，其属性为：

  • message（被退回的消息）
  • replyCode
  • replyText
  • exchange
  • routingKey

另请参见作用域操作，了解一种等待发布确认的更简单机制。

作用域操作​

通常，使用模板时，会从缓存中检出一个 Channel（或创建），用于执行操作，然后归还缓存复用。在多线程环境中，无法保证下一个操作使用同一个 channel。不过，有时你希望对 channel 的使用有更多控制，并确保若干操作都在同一个 channel 上执行。

从 2.0 版本开始，提供了一个名为 invoke 的新方法，带有一个 OperationsCallback。在回调作用域内、使用所提供的 RabbitOperations 参数执行的任何操作，都使用同一个专用 Channel，该 channel 在结束时关闭（不归还缓存）。如果该 channel 是 PublisherCallbackChannel，则在收到所有确认之后归还到缓存（参见关联的发布确认与退回）。

@FunctionalInterface
public interface OperationsCallback<T> {

    T doInRabbit(RabbitOperations operations);

}

你可能需要这样做的一个例子是：希望在底层 Channel 上使用 waitForConfirms() 方法。此方法以前未由 Spring API 暴露，因为如前所述，channel 通常是缓存和共享的。RabbitTemplate 现在提供 waitForConfirms(long timeout) 和 waitForConfirmsOrDie(long timeout)，它们委托给在 OperationsCallback 作用域内使用的专用 channel。出于显而易见的原因，这些方法不能在该作用域之外使用。

注意，在其他地方提供了一种让你把确认关联到请求的更高级抽象（参见关联的发布确认与退回）。如果你只想等待 broker 确认投递，可以使用下例所示的技术：

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

如果你希望在 OperationsCallback 作用域内在同一个 channel 上调用 RabbitAdmin 操作，则该 admin 必须是使用与 invoke 操作相同的 RabbitTemplate 构造的。

备注

如果模板操作已经在现有事务的作用域内执行，则前面的讨论不适用——例如在事务性监听器容器线程上运行并在事务性模板上执行操作。在那种情况下，操作在该 channel 上执行，并在线程返回容器时提交。在那种场景下不需要使用 invoke。

以这种方式使用确认时，为把确认关联到请求而搭建的大部分基础设施其实并不需要（除非同时启用了退回）。从 2.2 版本开始，连接工厂支持一个名为 publisherConfirmType 的新属性。当设置为 ConfirmType.SIMPLE 时，可以省去该基础设施，确认处理可以更高效。

此外，RabbitTemplate 会在已发送消息的 MessageProperties 中设置 publisherSequenceNumber 属性。如果你希望检查（或记录或以其他方式使用）特定确认，可以使用重载的 invoke 方法，如下例所示：

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

这些 ConfirmCallback 对象（用于 ack 和 nack 实例）是 Rabbit 客户端回调，而不是模板回调。

下例记录 ack 和 nack 实例：

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

作用域操作绑定到线程。关于多线程环境中的严格顺序，参见多线程环境中的严格消息顺序。

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

作用域操作中的讨论仅适用于操作在同一个线程上执行的情况。

考虑下面的情况：

  • thread-1 向队列发送消息，并把工作交接给 thread-2
  • thread-2 向同一个队列发送消息

由于 RabbitMQ 的异步特性和缓存 channel 的使用，无法保证使用同一个 channel，因此消息到达队列的顺序也无法保证。（在大多数情况下它们会按顺序到达，但乱序投递的概率不为零。）要解决这个用例，你可以使用大小为 1 的有界 channel 缓存（配合 channelCheckoutTimeout），以确保消息始终在同一个 channel 上发布，从而保证顺序。为此，如果你对连接工厂还有其他用途（例如消费者），应为模板使用专用连接工厂，或配置模板使用嵌入在主连接工厂中的发布者连接工厂（参见使用单独的连接）。

下面用一个简单的 Spring Boot 应用程序最能说明这一点：

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

}

尽管发布是在两个不同线程上执行的，但它们都会使用同一个 channel，因为缓存上限为一个 channel。

从 2.3.7 版本开始，ThreadChannelConnectionFactory 支持使用 prepareContextSwitch 和 switchContext 方法把某个线程的 channel 转移到另一个线程。第一个方法返回一个上下文，它被传递给调用第二个方法的第二个线程。一个线程可以绑定一个非事务性 channel 或一个事务性 channel（或两者各一个）；你不能单独转移它们，除非使用两个连接工厂。下面是一个例子：

@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();
	}

}
important

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

消息集成​

从 1.4 版本开始，RabbitMessagingTemplate（构建在 RabbitTemplate 之上）提供了与 Spring Framework 消息抽象的集成——即 org.springframework.messaging.Message。这让你可以使用 spring-messaging 的 Message<?> 抽象发送和接收消息。该抽象被其他 Spring 项目使用，例如 Spring Integration 和 Spring 的 STOMP 支持。其中涉及两个消息转换器：一个在 spring-messaging 的 Message<?> 与 Spring AMQP 的 Message 抽象之间转换，另一个在 Spring AMQP 的 Message 抽象与底层 RabbitMQ 客户端库所需格式之间转换。默认情况下，消息负载由所提供的 RabbitTemplate 实例的消息转换器转换。或者，你可以注入一个带有其他负载转换器的自定义 MessagingMessageConverter，如下例所示：

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

校验用户 ID​

从 1.6 版本开始，模板现在支持 user-id-expression（使用 Java 配置时为 userIdExpression）。发送消息时，会在求值此表达式之后设置用户 id 属性（如果尚未设置）。求值的根对象是要发送的消息。

以下示例展示了如何使用 user-id-expression 属性：

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

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

第一个示例是字面量表达式。第二个从应用程序上下文中的连接工厂 bean 获取 username 属性。

使用单独的连接​

从 2.0.2 版本开始，你可以将 usePublisherConnection 属性设置为 true，以便在可能时使用与监听器容器不同的连接。这是为了避免生产者因任何原因被阻塞时消费者也被阻塞。连接工厂为此维护第二个内部连接工厂；默认情况下它与主工厂类型相同，但如果你希望使用不同的工厂类型进行发布，可以显式设置。如果 rabbit 模板运行在监听器容器启动的事务中，则无论此设置如何，都使用容器的 channel。

important

一般来说，你不应把 RabbitAdmin 与将此属性设置为 true 的模板一起使用。请使用接收连接工厂的 RabbitAdmin 构造函数。如果你使用接收模板的另一个构造函数，请确保模板的该属性为 false。这是因为 admin 通常用于为监听器容器声明队列。使用该属性设置为 true 的模板意味着独占队列（例如 AnonymousQueue）会在与监听器容器不同的连接上声明。在那种情况下，队列无法被容器使用。