异步出站网关

上一节中讨论的网关是同步的，因为发送线程会暂停，直到收到回复（或超时）。Spring Integration 版本 4.3 添加了一个异步网关，它使用来自 Spring AMQP 的 AsyncRabbitTemplate。当发送消息时，线程在发送操作完成后立即返回，并且当收到消息时，回复将在模板的监听器容器线程上发送。当网关在轮询线程上调用时，这很有用。线程被释放，可用于框架中的其他任务。

以下列表显示了 AMQP 异步出站网关的可能配置选项

Java DSL
Java
XML

@Configuration
public class AmqpAsyncApplication {

    @Bean
    public IntegrationFlow asyncAmqpOutbound(AsyncRabbitTemplate asyncRabbitTemplate) {
        return f -> f
                .handle(Amqp.asyncOutboundGateway(asyncRabbitTemplate)
                        .routingKey("queue1")); // default exchange - route to queue 'queue1'
    }

    @MessagingGateway(defaultRequestChannel = "asyncAmqpOutbound.input")
    public interface MyGateway {

        String sendToRabbit(String data);

    }

}

@Configuration
public class AmqpAsyncConfig {

    @Bean
    @ServiceActivator(inputChannel = "amqpOutboundChannel")
    public AsyncAmqpOutboundGateway amqpOutbound(AsyncRabbitTemplate asyncTemplate) {
        AsyncAmqpOutboundGateway outbound = new AsyncAmqpOutboundGateway(asyncTemplate);
        outbound.setRoutingKey("foo"); // default exchange - route to queue 'foo'
        return outbound;
    }

    @Bean
    public AsyncRabbitTemplate asyncTemplate(RabbitTemplate rabbitTemplate,
                     SimpleMessageListenerContainer replyContainer) {

        return new AsyncRabbitTemplate(rabbitTemplate, replyContainer);
    }

    @Bean
    public SimpleMessageListenerContainer replyContainer() {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(ccf);
        container.setQueueNames("asyncRQ1");
        return container;
    }

    @Bean
    public MessageChannel amqpOutboundChannel() {
        return new DirectChannel();
    }

}

<int-amqp:outbound-async-gateway id="asyncOutboundGateway"    (1)
                           request-channel="myRequestChannel" (2)
                           async-template=""                  (3)
                           exchange-name=""                   (4)
                           exchange-name-expression=""        (5)
                           order="1"                          (6)
                           reply-channel=""                   (7)
                           reply-timeout=""                   (8)
                           requires-reply=""                  (9)
                           routing-key=""                     (10)
                           routing-key-expression=""          (11)
                           default-delivery-mode""            (12)
                           confirm-correlation-expression=""  (13)
                           confirm-ack-channel=""             (14)
                           confirm-nack-channel=""            (15)
                           confirm-timeout=""                 (16)
                           return-channel=""                  (17)
                           lazy-connect="true" />             (18)

1	此适配器的唯一 ID。可选。
2	消息通道，应将消息发送到该通道以将其转换为 AMQP 交换机并发布到该交换机。必需。
3	对已配置的 `AsyncRabbitTemplate` 的 Bean 引用。可选（默认为 `asyncRabbitTemplate`）。
4	应将消息发送到的 AMQP 交换机的名称。如果未提供，则消息将发送到默认的无名交换机。与 'exchange-name-expression' 互斥。可选。
5	一个 SpEL 表达式，用于评估以确定应将消息发送到的 AMQP 交换机的名称，其中消息是根对象。如果未提供，则消息将发送到默认的无名交换机。与 'exchange-name' 互斥。可选。
6	当注册多个消费者时，此消费者的顺序，从而启用负载均衡和故障转移。可选（默认为 `Ordered.LOWEST_PRECEDENCE [=Integer.MAX_VALUE]`）。
7	消息通道，应在从 AMQP 队列接收并转换后将回复发送到该通道。可选。
8	网关在将回复消息发送到`reply-channel`时等待的时间。这仅适用于`reply-channel`可以阻塞的情况，例如，当前已满的具有容量限制的`QueueChannel`。默认值为无穷大。
9	如果在`AsyncRabbitTemplate`的`receiveTimeout`属性设置的时间内未收到回复消息，并且此设置设置为`true`，则网关会将错误消息发送到入站消息的`errorChannel`标头。如果在`AsyncRabbitTemplate`的`receiveTimeout`属性设置的时间内未收到回复消息，并且此设置设置为`false`，则网关会将错误消息发送到默认的`errorChannel`（如果可用）。默认值为`true`。
10	发送消息时使用的路由键。默认情况下，这是一个空`String`。与'routing-key-expression'互斥。可选。
11	用于确定发送消息时使用的路由键的SpEL表达式，其中消息作为根对象（例如，'payload.key'）。默认情况下，这是一个空`String`。与'routing-key'互斥。可选。
12	消息的默认传递模式：`PERSISTENT`或`NON_PERSISTENT`。如果`header-mapper`设置了传递模式，则会被覆盖。如果存在Spring Integration消息标头（`amqp_deliveryMode`），则`DefaultHeaderMapper`会设置该值。如果未提供此属性并且标头映射器未设置它，则默认值取决于`RabbitTemplate`使用的底层Spring AMQP `MessagePropertiesConverter`。如果未自定义，则默认值为`PERSISTENT`。可选。
13	定义相关数据的表达式。提供此表达式时，将配置底层AMQP模板以接收发布者确认。需要一个专用的`RabbitTemplate`和一个`CachingConnectionFactory`，其`publisherConfirms`属性设置为`true`。当收到发布者确认并且提供了相关数据时，确认将写入`confirm-ack-channel`或`confirm-nack-channel`，具体取决于确认类型。确认的有效负载是此表达式定义的相关数据，并且消息的'amqp_publishConfirm'标头设置为`true`（`ack`）或`false`（`nack`）。对于`nack`实例，将提供一个额外的标头（`amqp_publishConfirmNackCause`）。示例：`headers['myCorrelationData']`、`payload`。如果表达式解析为`Message<?>`实例（例如，“#this”），则在`ack`/`nack`通道上发出的消息将基于该消息，并添加了额外的标头。另请参阅发布者确认和返回的替代机制。可选。
14	发送正向（`ack`）发布者确认的通道。有效负载是`confirm-correlation-expression`定义的相关数据。需要底层`AsyncRabbitTemplate`将其`enableConfirms`属性设置为`true`。另请参阅发布者确认和返回的替代机制。可选（默认值为`nullChannel`）。
15	从版本4.2开始。发送负向（`nack`）发布者确认的通道。有效负载是`confirm-correlation-expression`定义的相关数据。需要底层`AsyncRabbitTemplate`将其`enableConfirms`属性设置为`true`。另请参阅发布者确认和返回的替代机制。可选（默认值为`nullChannel`）。
16	如果在指定的时间（毫秒）内未收到发布者确认，则网关将合成一个否定确认（nack）。挂起的确认每隔此值的 50% 检查一次，因此实际发送 nack 的时间将在此值的 1 倍到 1.5 倍之间。另请参见发布者确认和返回的替代机制。默认情况下为无（不会生成 nack）。
17	返回消息发送到的通道。提供此通道时，底层 AMQP 模板配置为将无法传递的消息返回到网关。消息由从 AMQP 收到的数据构建，并包含以下附加标头：`amqp_returnReplyCode`、`amqp_returnReplyText`、`amqp_returnExchange` 和 `amqp_returnRoutingKey`。要求底层 `AsyncRabbitTemplate` 的 `mandatory` 属性设置为 `true`。另请参见发布者确认和返回的替代机制。可选。
18	当设置为 `false` 时，端点尝试在应用程序上下文初始化期间连接到代理。这样做可以通过在代理关闭时记录错误消息来实现对错误配置的“快速失败”检测。当设置为 `true`（默认值）时，在发送第一条消息时建立连接（如果由于其他组件已建立连接而尚未存在）。

另请参见异步服务激活器以获取更多信息。

RabbitTemplate

当您使用确认和返回时，我们建议将连接到 AsyncRabbitTemplate 的 RabbitTemplate 专用。否则，可能会遇到意外的副作用。