RabbitMQ AMQP 1.0 支持

Doubao 中英对照 RabbitMQ AMQP 1.0 Support

4.0 版本引入了 spring-rabbitmq-client 模块，用于在 RabbitMQ 上支持 AMQP 1.0 协议。

该构件基于 com.rabbitmq.client:amqp-client 库，因此只能与 RabbitMQ 及其 AMQP 1.0 协议支持协作。它不能用于任意的 AMQP 1.0 broker。为此，目前推荐使用 JMS bridge 以及相应的 Spring JMS 集成。

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

Maven

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

Gradle

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

spring-rabbit（用于 AMQP 0.9.1 协议）作为传递依赖引入，以便在此新客户端中复用一些通用 API，例如异常、@RabbitListener 支持。目标项目中不必同时使用两种功能，但 RabbitMQ 允许 AMQP 0.9.1 与 1.0 共存。

关于 RabbitMQ AMQP 1.0 Java Client 的更多信息，参见其文档。

RabbitMQ AMQP 1.0 环境

com.rabbitmq.client.amqp.Environment 是必须首先添加到项目中、用于连接管理和其他通用设置的东西。它是连接到某个节点或节点集群的入口。环境允许创建连接。它可以包含在连接之间共享的基础设施相关配置设置，例如线程池、指标和/或观测：

@Bean
Environment environment() {
    return new AmqpEnvironmentBuilder()
            .connectionSettings()
            .port(5672)
            .environmentBuilder()
            .build();
}

同一个 Environment 实例可用于连接不同的 RabbitMQ broker，此时连接设置必须在特定连接上提供。见下文。

AMQP 连接工厂

引入了 org.springframework.amqp.rabbitmq.client.AmqpConnectionFactory 抽象来管理 com.rabbitmq.client.amqp.Connection。不要把它与仅用于 AMQP 0.9.1 协议的 org.springframework.amqp.rabbit.connection.ConnectionFactory 混淆。提供 SingleAmqpConnectionFactory 实现来管理一个连接及其设置。同一个 Connection 可在多个 producer、consumer 和管理操作之间共享。多路复用由 AMQP 客户端库内部用于 AMQP 1.0 协议实现的 link 抽象处理。Connection 具备恢复能力，也处理拓扑。

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

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

有关所有连接相关设置，参见 SingleAmqpConnectionFactory 的 setter。

RabbitMQ 拓扑管理

对于从应用角度进行的拓扑管理（exchange、队列及它们之间的 binding），提供了 RabbitAmqpAdmin，它是现有 AmqpAdmin 接口的一个实现：

@Bean
RabbitAmqpAdmin admin(AmqpConnectionFactory connectionFactory) {
    return new RabbitAmqpAdmin(connectionFactory);
}

必须使用与配置 Broker 中所述相同的 Exchange、Queue、Binding 和 Declarables bean 定义来管理拓扑。来自 spring-rabbit 的 RabbitAdmin 也可以做到这一点，但它是针对 AMQP 0.9.1 连接进行的；而 RabbitAmqpAdmin 基于 AMQP 1.0 连接，因此拓扑恢复会与 producer 和 consumer 的恢复一起平滑地处理。

RabbitAmqpAdmin 在其 start() 生命周期回调中执行相应的 bean 扫描。initialize() 以及所有其他 RabbitMQ 实体管理方法都可以在运行时手动调用。在内部，RabbitAmqpAdmin 使用 com.rabbitmq.client.amqp.Connection.management() API 执行相应的拓扑操作。

RabbitAmqpTemplate

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

通常，只需一个这样的 bean，即可执行所有可能的 template 模式操作：

@Bean
RabbitAmqpTemplate rabbitTemplate(AmqpConnectionFactory connectionFactory) {
    return new RabbitAmqpTemplate(connectionFactory);
}

可以为其配置默认的 exchange 和 routing key，或仅配置队列。RabbitAmqpTemplate 有一个用于接收操作的默认队列，以及另一个用于请求-回复操作的默认队列；如果请求对应的临时队列不存在，则由客户端创建。

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

@Bean
DirectExchange e1() {
    return new DirectExchange("e1");
}

@Bean
Queue q1() {
    return QueueBuilder.durable("q1").deadLetterExchange("dlx1").build();
}

@Bean
Binding b1() {
    return BindingBuilder.bind(q1()).to(e1()).with("k1");
}

...

@Test
void defaultExchangeAndRoutingKey() {
    this.rabbitAmqpTemplate.setExchange("e1");
    this.rabbitAmqpTemplate.setRoutingKey("k1");
	this.rabbitAmqpTemplate.setReceiveQueue("q1");

    assertThat(this.rabbitAmqpTemplate.convertAndSend("test1"))
            .succeedsWithin(Duration.ofSeconds(10));

    assertThat(this.rabbitAmqpTemplate.receiveAndConvert())
            .succeedsWithin(Duration.ofSeconds(10))
            .isEqualTo("test1");
}

这里我们声明了一个 e1 exchange、一个 q1 队列，并用 k1 routing key 将其绑定到该 exchange。然后我们使用 RabbitAmqpTemplate 的默认设置，以相应的 routing key 向上述 exchange 发布消息，并使用 q1 作为接收操作的默认队列。这些方法有重载变体，可发送到特定的 exchange 或队列（用于发送和接收）。带 ParameterizedTypeReference<T> 的 receiveAndConvert() 操作需要向 RabbitAmqpTemplate 注入 SmartMessageConverter。

下一个示例展示用 RabbitAmqpTemplate 实现 RPC（假设与上一个示例使用相同的 RabbitMQ 对象）：

@Test
void verifyRpc() {
    String testRequest = "rpc-request";
    String testReply = "rpc-reply";

    CompletableFuture<Object> rpcClientResult = this.template.convertSendAndReceive("e1", "k1", testRequest);

    AtomicReference<String> receivedRequest = new AtomicReference<>();
    CompletableFuture<Boolean> rpcServerResult =
            this.rabbitAmqpTemplate.<String, String>receiveAndReply("q1",
                     payload -> {
                         receivedRequest.set(payload);
                         return testReply;
                     });

    assertThat(rpcServerResult).succeedsWithin(Duration.ofSeconds(10)).isEqualTo(true);
    assertThat(rpcClientResult).succeedsWithin(Duration.ofSeconds(10)).isEqualTo(testReply);
    assertThat(receivedRequest.get()).isEqualTo(testRequest);
}

关联和 replyTo 队列在内部管理。服务端可以用下面描述的 @RabbitListener POJO 方法实现。

RabbitMQ AMQP 1.0 Consumer

与许多其他 consumer 端消息实现一样，spring-rabbitmq-client 模块自带 RabbitAmqpListenerContainer，它本质上是广为人知的 org.springframework.amqp.core.MessageListenerContainer 的一个实现。它做的事情与 DirectMessageListenerContainer 完全相同，只不过是面向 RabbitMQ AMQP 1.0 支持。它需要一个 AmqpConnectionFactory 和至少一个要消费的队列。此外，还必须提供 MessageListener（或 AMQP 1.0 专用的 RabbitAmqpMessageListener）。可以配置为 autoSettle = false，含义即 AcknowledgeMode.MANUAL。在这种情况下，提供给 MessageListener 的 Message 会在其 MessageProperties 中带有一个 AmqpAcknowledgment 回调，供目标逻辑考虑。

RabbitAmqpMessageListener 有一个针对 com.rabbitmq.client:amqp-client 抽象的契约：

/**
 * Process an AMQP message.
 * @param message the message to process.
 * @param context the consumer context to settle message.
 *                Null if container is configured for {@code autoSettle}.
 */
void onAmqpMessage(Message message, Consumer.Context context);

其中第一个参数是原生接收到的 com.rabbitmq.client.amqp.Message，context 是用于消息确认的原生回调，类似于上面提到的 AmqpAcknowledgment 抽象。

当提供了 batchSize 选项时，RabbitAmqpMessageListener 可以批量处理并确认消息。为此必须实现 MessageListener.onMessageBatch() 契约。batchReceiveDuration 选项用于为未满的批次调度强制释放，以避免内存和consumer credit耗尽。

通常，目标项目中不会直接使用 RabbitAmqpMessageListener 类，而是选择通过 @RabbitListener 的 POJO 方法注解配置来进行声明式 consumer 配置。RabbitAmqpListenerContainerFactory 必须注册在 RabbitListenerAnnotationBeanPostProcessor.DEFAULT_RABBIT_LISTENER_CONTAINER_FACTORY_BEAN_NAME 之下，@RabbitListener 注解处理过程会将 RabbitAmqpMessageListener 实例注册到 RabbitListenerEndpointRegistry。目标 POJO 方法的调用由特定的 RabbitAmqpMessageListenerAdapter 实现处理，它扩展了 MessagingMessageListenerAdapter 并复用其大量功能，包括请求-回复场景（无论是否异步）。因此，注解驱动的 Listener 端点中描述的所有概念也同样适用于此 RabbitAmqpMessageListener。

除了传统的消息 payload 和 headers 之外，@RabbitListener POJO 方法契约还可以带有以下参数：

• com.rabbitmq.client.amqp.Message——未经任何转换的原生 AMQP 1.0 消息；
• org.springframework.amqp.core.Message——由原生 AMQP 1.0 消息转换而来的 Spring AMQP 消息抽象；
• org.springframework.messaging.Message——由 Spring AMQP 消息转换而来的 Spring Messaging 抽象；
• Consumer.Context——RabbitMQ AMQP 客户端 consumer 确认 API；
• org.springframework.amqp.core.AmqpAcknowledgment——Spring AMQP 确认抽象：委托给 Consumer.Context。

以下示例展示一个用于 RabbitMQ AMQP 1.0 交互、带手动确认的简单 @RabbitListener：

@Bean(RabbitListenerAnnotationBeanPostProcessor.DEFAULT_RABBIT_LISTENER_CONTAINER_FACTORY_BEAN_NAME)
RabbitAmqpListenerContainerFactory rabbitAmqpListenerContainerFactory(AmqpConnectionFactory connectionFactory) {
    return new RabbitAmqpListenerContainerFactory(connectionFactory);
}

final List<String> received = Collections.synchronizedList(new ArrayList<>());

CountDownLatch consumeIsDone = new CountDownLatch(11);

@RabbitListener(queues = {"q1", "q2"},
        ackMode = "#{T(org.springframework.amqp.core.AcknowledgeMode).MANUAL}",
        concurrency = "2",
        id = "testAmqpListener")
void processQ1AndQ2Data(String data, AmqpAcknowledgment acknowledgment, Consumer.Context context) {
    try {
        if ("discard".equals(data)) {
            if (!this.received.contains(data)) {
                context.discard();
            }
            else {
                throw new MessageConversionException("Test message is rejected");
            }
        }
        else if ("requeue".equals(data) && !this.received.contains(data)) {
            acknowledgment.acknowledge(AmqpAcknowledgment.Status.REQUEUE);
        }
        else {
            acknowledgment.acknowledge();
        }
        this.received.add(data);
    }
    finally {
        this.consumeIsDone.countDown();
    }
}