通用 AMQP 1.0 支持

Doubao 中英对照 Generic AMQP 1.0 Support

4.1 版本引入了 spring-amqp-client 模块，用于支持 AMQP 1.0 协议。

该构件基于 Qpid ProtonJ2 Client Library，可与任何支持 AMQP 1.0 协议的对等端协作，包括 RabbitMQ broker。

必须将此依赖添加到项目中，才能与 AMQP 1.0 支持交互：

Maven

<dependency>
  <groupId>org.springframework.amqp</groupId>
  <artifactId>spring-amqp-client</artifactId>
  <version>4.1.1</version>
</dependency>

Gradle

implementation 'org.springframework.amqp:spring-amqp-client:4.1.1'

AMQP 1.0 环境

org.apache.qpid:protonj2-client 非常灵活，并提供便捷的 API，即使没有任何专门的 Spring 实现，也可以在 Spring 应用中使用它。例如，org.apache.qpid.protonj2.client.Message 类就是一个 builder 模式实现。无论如何，spring-amqp-client 都为连接生命周期管理以及发送/接收消息操作的 AMQP 1.0 协议交互提供了高层的 Spring 模式实现。AMQP 1.0 环境从一个 org.apache.qpid.protonj2.client.Client 实例开始：

@Bean
Client protonClient() {
    return Client.create();
}

同一个 Client 可用于连接不同的 broker，连接设置必须在特定连接上提供。不过，此 bean 不是必需的，SingleAmqpConnectionFactory 会在内部创建实例。另见下文的AMQP 1.0 连接工厂。

AMQP 1.0 连接工厂

引入了 org.springframework.amqp.client.AmqpConnectionFactory 抽象来管理 org.apache.qpid.protonj2.client.Connection。提供 SingleAmqpConnectionFactory 实现来管理一个连接及其设置。同一个 Connection 可在多个 producer 和 consumer 之间共享。多路复用由 AMQP 客户端库内部用于 AMQP 1.0 协议实现的 link 抽象处理。Connection 通过为 ConnectionOptions 提供的 ReconnectOptions 具备恢复能力。

在大多数情况下，向项目中添加这个 bean 即可：

@Bean
AmqpConnectionFactory connectionFactory() {
    return new SingleAmqpConnectionFactory();
}

有关所有连接相关选项（包括 host、port、user），参见 SingleAmqpConnectionFactory 的 setter。更高级的配置可通过 setConnectionOptions(ConnectionOptions) 完成，它是 ProtonJ 库的 builder API。前面提到的使用全部默认值的 SingleAmqpConnectionFactory 实例将连接到 localhost:5672，并进行无限次重连。

备注

对于 SingleAmqpConnectionFactory，可以省略 Client 注入，相应实例由工厂在内部创建。

AmqpClient

AmqpClient 是一个 fluent API，用于针对 AMQP 1.0 协议对消息执行发送和接收操作。它需要一个 AmqpConnectionFactory，并可通过 AmqpClient.Builder API 配置一些默认值。DefaultAmqpClient 是 AmqpClient 契约的内部实现。即使 org.apache.qpid:protonj2-client 库自带 org.apache.qpid.protonj2.client.Message 实现，AmqpClient 仍暴露基于广为人知的 org.springframework.amqp.core.Message 的 API，并附带所有支持类（如 MessageProperties 和 MessageConverter 抽象）。与 org.apache.qpid.protonj2.client.Message 之间的转换由 AmqpClient 实现内部通过 ProtonUtils 支持类完成。所有发送和接收方法都返回 CompletableFuture，以便最终获取操作结果。与普通对象交互需要消息体转换，默认使用 SimpleMessageConverter。关于转换的更多信息，参见 AmqpClient.Builder.messageConverter(MessageConverter) 配置和消息转换器。

通常，只需一个这样的 bean，即可执行所有可能的发送和接收操作：

@Bean
AmqpClient amqpClient(AmqpConnectionFactory connectionFactory) {
    return AmqpClient.builder(connectionFactory)
            .defaultToAddress("/queues/some_queue_as_default")
            .messageConverter(new JacksonJsonMessageConverter())
            .build();
}

备注

在上面的示例中，用于 defaultToAddress 的 /queues/ 前缀是 RabbitMQ 中目标地址约定的一个示例。

以下是 AmqpClient 操作的一些示例：

CompletableFuture<Boolean> sendFuture = this.amqpClient.send(Message.create("test_data"));

CompletableFuture<Boolean> sendFuture =
    this.amqpClient
            .to("/queues/test_queue")
            .message(new org.springframework.amqp.core.Message("test_data2".getBytes()))
            .send();

CompletableFuture<Boolean> sendFuture =
        this.amqpClient
                .to("/queues/test_queue")
                .body("convert")
                .priority(7)
                .header("test_header", "test_value")
                .messageId("some_id")
                .userId("guest")
                .send();

AmqpClient 还提供一个 fluent API，用于按需从地址接收消息。为此，应调用 AmqpClient.from(String fromAddress) 方法，它返回一个 ReceiveSpec，用于各种接收操作行为。它们都会产生一个 CompletableFuture，其中要么是原生 ProtonJ 消息、Spring AMQP 消息，要么只是从消息体转换而来的负载。带非 Object.class 泛型参数的 receiveAndConvert() 要求 AmqpClient 中的 messageConverter 是 SmartMessageConverter。例如，当为 AmqpClient 提供了 JacksonJsonMessageConverter 时，以下示例就能产生正确的数据模型转换：

record TestData(String data) { }

 this.amqpClient
        .to("/queues/test_queue")
        .body(new TestData("convert"))
        .send();

CompletableFuture<TestData> receiveFuture =
        this.amqpClient.from("/queues/test_queue")
                    .receiveAndConvert();

关于 AmqpClient 所暴露的 fluent API，更多信息参见这些方法的 Javadoc。

AMQP 1.0 Consumer

AMQP 1.0 协议的事件驱动 consumer 实现为 AmqpMessageListenerContainer，与框架中所有其他listener 容器实现类似。AmqpMessageListenerContainer 需要一个 AmqpConnectionFactory，并在内部使用 org.apache.qpid.protonj2.client.Receiver 实例，从所提供的 queueNames（本质上即 AMQP 1.o 地址）消费消息。receiver link 默认使用 100 个初始 credit。每完成一次投递，就会为即将到来的新消息补充一定数量的 credit。消费的消息由所提供的 MessageListener 处理；如果 autoAccept 设为 false，则会在相应的 Spring AMQP 消息属性中填入 AmqpAcknowledgment 实现，以便在目标消息监听器逻辑中手动进行投递确认。

consumersPerQueue 选项（默认 1）为每个要消费的地址实现并发行为。

Duration receiveTimeout 选项（默认 1 second）控制阻塞式的 Receiver.receive() 操作，该操作循环调用直到 consumer 停止。

每个 consumer（内部的 AmqpMessageListenerContainer.AmqpConsumer 实例）通过 Executor 调度运行。AmqpMessageListenerContainer.taskExecutor 属性默认设为 SimpleAsyncTaskExecutor。

消息处理错误可通过 ErrorHandler 配置处理。

当 Receiver.receive() 操作失败时，receiver link 会断开，即使连接本身已重新建立，ProtonJ 客户端也不会恢复它。consumer 会向 ErrorHandler 报告此失败（如果未提供 ErrorHandler，则将其作为错误记录），然后根据 recoveryBackOff 选项重新打开其 receiver link；该选项默认是一个间隔 5 seconds、无限次尝试的 FixedBackOff。setRecoveryInterval(Duration) 是一个便捷选项，仅用于更改恢复尝试之间的间隔。如果配置的 BackOff 尝试次数耗尽，则 consumer 停止并从容器中移除，因此当没有剩余 consumer 时，AmqpMessageListenerContainer.isRunning() 返回 false。

如果提供了某些 AOP 拦截器（例如 TransactionInterceptor），MessageListener 可以在 AmqpMessageListenerContainer 中被代理。

AmqpMessageListenerContainer 还提供 pause() 和 resume() API，分别将 AMQP link credit 设为 0 和补充 credit。此功能保持 Receiver 处于活动状态，但不会有来自 AMQP 对等端的投递。

以下示例展示 AmqpMessageListenerContainer 的简单配置：

BlockingQueue<Message> receivedMessages = new LinkedBlockingQueue<>();

@Bean
AmqpMessageListenerContainer amqpMessageListenerContainer(AmqpConnectionFactory connectionFactory) {
    var amqpMessageListenerContainer = new AmqpMessageListenerContainer(connectionFactory);
    amqpMessageListenerContainer.setQueueNames("address1", "address2");
    amqpMessageListenerContainer.setConsumersPerQueue(3);
    amqpMessageListenerContainer.setAutoAccept(false);
    amqpMessageListenerContainer.setReceiveTimeout(Duration.ofMillis(100));
    amqpMessageListenerContainer.setupMessageListener(this.receivedMessages::add);
    return amqpMessageListenerContainer;
}

ProtonJ Delivery 消费

为方便起见，提供了 ProtonDeliveryListener 契约来处理原生 ProtonJ Delivery 对象，而非 Spring AMQP 消息。在需要完全控制 Delivery 实例的场景下，它会很有用。例如，根据某些目标应用逻辑添加一些拒绝条件、将接收到的数据作为 InputStream 处理、动态补充 link credit。原生 Delivery 处理的另一种场景是：producer 和 consumer 都处理原生 ProtonJ Message 契约，包括 ProtonJ 消息体编码/解码的内部机制（与只接受 byte[] 消息体的 Spring AMQP 消息不同）。

备注

如果 AmqpMessageListenerContainer 配置为 autoAccept = true（默认），则仍可以跳过手动投递确认。AmqpMessageListenerContainer 会自动执行 credit 补充。

ProtonDeliveryListener 实现应作为常规 MessageListener 注入到 AmqpMessageListenerContainer：

BlockingQueue<Delivery> receivedDeliveries = new LinkedBlockingQueue<>();

@Bean
AmqpMessageListenerContainer protonDeliveryListenerContainer(AmqpConnectionFactory connectionFactory) {
    var amqpMessageListenerContainer = new AmqpMessageListenerContainer(connectionFactory);
    amqpMessageListenerContainer.setQueueNames(TEST_QUEUE_FOR_NATIVE_PROTON);
    amqpMessageListenerContainer.setAutoAccept(false);
    amqpMessageListenerContainer.setupMessageListener((ProtonDeliveryListener) this.receivedDeliveries::add);
    return amqpMessageListenerContainer;
}

AMQP 1.0 注解配置

spring-amqp-client 提供了一种便捷方式，通过注解配置 AMQP 1.0 基础设施。可以在目标项目的 @Configuration 类上设置 @EnableAmqp 注解，以触发该基础设施的注册。由 @EnableAmqp 注解导入的 AmqpDefaultConfiguration 提供支持 @AmqpListener 的基础设施 bean（见下文）：

• AmqpListenerEndpointRegistry——以编程方式注册 AMQP listener 容器的全局组件。
• AmqpListenerAnnotationBeanPostProcessor（位于 AmqpDefaultConfiguration.AMQP_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME 之下）——一个 BeanPostProcessor，用于解析带 @AmqpListener 的 POJO 方法，并通过 AmqpListenerEndpointRegistry 注册 listener 容器。

@AmqpListener

为了进行高层且便捷的配置，框架提供了 @AmqpListener 注解，用于将 POJO 方法标记为 listener 调用器，供框架所创建的 AmqpMessageListenerContainer 实例使用。该注解与 @RabbitListener、@KafkaListener、@JmsListener 有诸多相似之处。它指定要消费的 AMQP 地址，也可以用许多其他 listener 容器属性来配置。所有 @AmqpListener 属性都可以配置为属性占位符（${}）和/或 SpEL 表达式（#{}）。也支持请求-回复方法签名，如果请求消息中未提供 replyTo，@SendTo 可用作兜底回复地址。这些方法可以具有异步返回类型：Project Reactor 的 Mono、CompletableFuture 或 Kotlin suspend 函数。

• 基于 Mono 的请求-回复示例（带可选属性）：

@Bean
AmqpMessageListenerContainerFactory customContainerFactory(AmqpConnectionFactory connectionFactory) {
	var containerFactory = new MethodAmqpMessageListenerContainerFactory(connectionFactory);
	containerFactory.setAutoStartup(false);
	containerFactory.setAutoAccept(true);
	containerFactory.setConcurrency(5);
	containerFactory.setReceiveTimeout(Duration.ofSeconds(5));
	containerFactory.setGracefulShutdownPeriod(Duration.ofSeconds(36));
	containerFactory.setInitialCredits(55);
	containerFactory.setTaskExecutor(this.taskExecutor);
	containerFactory.setAdviceChain(new DebugInterceptor());
	containerFactory.setListenerErrorHandler(mock());
	containerFactory.setDefaultRequeueRejected(false);
	containerFactory.setHeaderMapper(this.headerMapper);
	containerFactory.setMessageConverter(this.messageConverter);
	return containerFactory;
}

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

@Bean
AmqpHeaderMapper headerMapper() {
	return new SimpleAmqpHeaderMapper();
}

@Bean
DebugInterceptor debugInterceptor() {
	return new DebugInterceptor();
}

@AmqpListener(
        id = "monoListener",
        addresses = TEST_QUEUE2 + "," + TEST_QUEUE3,
        concurrency = "2",
        autoStartup = "#{true}",
        autoAccept = "true",
        containerFactory = "customContainerFactory",
        receiveTimeout = "${receive.timeout:22000}",
        gracefulShutdownPeriod = "pt2m",
        initialCredits = "101",
        defaultRequeueRejected = "true",
        executor = "#{taskExecutor}",
        messageConverter = "jsonMessageConverter",
        headerMapper = "headerMapper",
        adviceChain = "debugInterceptor")
Mono<DataOut> requestReplyWithJsonAndOtherAnnotationAttributes(DataIn dataIn) {
       DataOut dataOut = new DataOut(dataIn.data + "_out");
       return Mono.just(dataOut);
}

id 属性不是必需的，目标 listener 容器 bean 名称将基于类名、方法名再加上 .listenerContainer 后缀。addresses——可以是名称数组，或每一项都可以是逗号分隔的名称集合。当通过外部配置属性提供 AMQP 地址列表时，这会很有用。containerFactory（见下文）、executor、messageConverter 和 headerMapper 属性是可选的，可以配置为 bean 名称或 bean 引用（作为 SpEL 表达式，如上面示例中的 executor）。adviceChain 属性是 Advice 实现的 bean 名称列表。

• 带显式 @SendTo 的 Kotlin suspend 函数示例：

@Bean
fun jsonMessageConverter() = JacksonJsonMessageConverter()

@Bean
fun amqpConnectionFactory(protonClient: Client) =
    SingleAmqpConnectionFactory(protonClient)
        .setPort(amqpPort())

@Bean
fun amqpClient(connectionFactory: AmqpConnectionFactory, jsonMessageConverter: MessageConverter) =
    AmqpClient.builder(connectionFactory)
        .messageConverter(jsonMessageConverter)
        .build()

@Bean(AmqpDefaultConfiguration.DEFAULT_AMQP_LISTENER_CONTAINER_FACTORY_BEAN_NAME)
fun containerFactory(connectionFactory: AmqpConnectionFactory, jsonMessageConverter: MessageConverter) =
    MethodAmqpMessageListenerContainerFactory(connectionFactory)
        .also { it.setMessageConverter(jsonMessageConverter) }

@AmqpListener(addresses = ["TEST_QUEUE"])
@SendTo("TEST_REPLY_TO")
suspend fun requestReplyAsyncListener(payload: DataIn) = DataOut(payload.data + "_out")

data class DataIn(val data: String)

data class DataOut(val data: String)

@AmqpListener 是 @Repeatable 的，因此会为调用同一服务方法的每个注解注册多个 listener 容器。

AmqpMessageListenerContainerFactory

AmqpMessageListenerContainerFactory 抽象可用于根据所提供的 AmqpListenerEndpoint 创建 AmqpMessageListenerContainer。如果 listener 容器属性未被 AmqpListenerEndpoint 实例覆盖，工厂会提供默认值。框架提供了基于 MessageListener 注入的 SimpleAmqpListenerEndpoint 实现，以及基于特定 POJO 方法调用的 MethodAmqpListenerEndpoint。两种 endpoint 实现都指定要消费的 AMQP 地址。AmqpMessageListenerContainerFactory 的 MethodAmqpMessageListenerContainerFactory 扩展专门用于基于 MethodAmqpListenerEndpoint 的 AmqpMessageListenerContainer 实例（内部使用 AmqpMessagingListenerAdapter 作为消息监听器）。当通过 AmqpListenerAnnotationBeanPostProcessor 解析 @AmqpListener 注解时，框架依赖此工厂的 bean。如果 POJO 方法上未提供 @AmqpListener(containerFactory) 属性，框架期望有一个全局的 AmqpMessageListenerContainerFactory bean，名称为 AmqpDefaultConfiguration.DEFAULT_AMQP_LISTENER_CONTAINER_FACTORY_BEAN_NAME。

与 @RabbitListener 和 RabbitListenerEndpointRegistry 基础设施不同，AmqpListenerEndpointRegistry 使用 AmqpMessageListenerContainerFactory 和 AmqpListenerEndpoint 来创建 AmqpMessageListenerContainer 实例，并将其作为单例 bean 直接注册到应用上下文。此 AmqpMessageListenerContainer 实例的 bean 名称取自 AmqpListenerEndpoint.getId()，或基于 MessageListener 类名与 .listenerContainer 后缀生成；对于 MethodAmqpListenerEndpoint，则是服务类名加方法名及前述后缀。

更多信息参见 AmqpListenerEndpoint、AmqpMessageListenerContainerFactory 和 AmqpListenerEndpointRegistry 的 Javadoc。