异步出站网关

前一节讨论的网关是同步的，即发送线程会一直暂停，直到收到回复（或发生超时）。 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"    [id="CO1-1"]1
                           request-channel="myRequestChannel" [id="CO1-2"]2
                           async-template=""                  [id="CO1-3"]3
                           exchange-name=""                   [id="CO1-4"]4
                           exchange-name-expression=""        [id="CO1-5"]5
                           order="1"                          [id="CO1-6"]6
                           reply-channel=""                   [id="CO1-7"]7
                           reply-timeout=""                   [id="CO1-8"]8
                           requires-reply=""                  [id="CO1-9"]9
                           routing-key=""                     [id="CO1-10"]10
                           routing-key-expression=""          [id="CO1-11"]11
                           default-delivery-mode""            [id="CO1-12"]12
                           confirm-correlation-expression=""  [id="CO1-13"]13
                           confirm-ack-channel=""             [id="CO1-14"]14
                           confirm-nack-channel=""            [id="CO1-15"]15
                           confirm-timeout=""                 [id="CO1-16"]16
                           return-channel=""                  [id="CO1-17"]17
                           lazy-connect="true" />             [id="CO1-18"]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`（默认值）时，连接在发送第一条消息时建立（如果由于其他组件已建立而尚未存在）。

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

Example 1. RabbitTemplate

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