异步出站网关

前面讨论的网关是同步的，这意味着发送线程会被挂起，直到接收到回复（或发生超时）。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 会设置其值。如果未提供此属性且 header mapper 未设置，则默认值取决于 RabbitTemplate 使用的底层 Spring AMQP MessagePropertiesConverter。如果未自定义该转换器，则默认是 PERSISTENT。可选。
13 一个表达式，用于定义关联数据。提供此表达式后，会配置底层 AMQP 模板接收发布者确认。需要一个专用的 RabbitTemplate 和一个设置了 publisherConfirms 属性为 true 的 CachingConnectionFactory。当接收到发布者确认并提供了关联数据时，确认会写入到 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 倍之间。另请参见 发布者确认和返回的替代机制。默认无（不会生成 nacks）。
17 退回的消息将发送到此通道。提供此项后，底层 AMQP 模板会被配置为将无法投递的消息返回到网关。消息将根据从 AMQP 接收的数据构建，并包含以下附加头：amqp_returnReplyCode, amqp_returnReplyText, amqp_returnExchange 和 amqp_returnRoutingKey。要求底层的 AsyncRabbitTemplate 将其 mandatory 属性设置为 true。另请参见 发布者确认和返回的替代机制。可选。
18 当设置为 false 时，端点会在应用上下文初始化期间尝试连接到 broker。这样做可以通过在 broker 关闭时记录错误消息来快速检测错误的配置。“快速失败”。当设置为 true（默认）时，连接会在发送第一条消息时建立（如果它尚未因其他组件而存在）。

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

RabbitTemplate

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