异步出站网关

前面章节讨论的网关是同步的,这意味着发送线程会一直挂起,直到收到回复(或超时)。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 当在 AsyncRabbitTemplatereceiveTimeout 属性设置的时间内未收到回复消息,并且此设置为 true 时,网关会将错误消息发送到入站消息的 errorChannel 标头。当在 AsyncRabbitTemplatereceiveTimeout 属性设置的时间内未收到回复消息,并且此设置为 false 时,网关会将错误消息发送到默认的 errorChannel(如果可用)。默认为 true
10 发送消息时使用的路由键。默认情况下,这是一个空 String。与 'routing-key-expression' 互斥。可选。
11 一个 SpEL 表达式,用于根据消息确定发送消息时使用的路由键,其中消息为根对象(例如,'payload.key')。默认情况下,这是一个空 String。与 'routing-key' 互斥。可选。
12 消息的默认传递模式:PERSISTENTNON_PERSISTENT。如果 header-mapper 设置了传递模式,则会被覆盖。如果 Spring Integration 消息标头 (amqp_deliveryMode) 存在,则 DefaultHeaderMapper 会设置该值。如果未提供此属性且标头映射器未设置它,则默认值取决于 RabbitTemplate 使用的基础 Spring AMQP MessagePropertiesConverter。如果未自定义,则默认为 PERSISTENT。可选。
13 定义相关数据的表达式。提供此表达式时,将配置底层 AMQP 模板以接收发布者确认。需要一个专用的 RabbitTemplate 和一个 CachingConnectionFactory,其 publisherConfirms 属性设置为 true。当收到发布者确认并提供相关数据时,确认将写入 confirm-ack-channelconfirm-nack-channel,具体取决于确认类型。确认的有效负载是此表达式定义的相关数据,并且消息的 'amqp_publishConfirm' 标头设置为 true (ack) 或 false (nack)。对于 nack 实例,会提供一个额外的标头 (amqp_publishConfirmNackCause)。示例:headers['myCorrelationData']payload。如果表达式解析为 Message<?> 实例(例如“#this”),则在 ack/nack 通道上发出的消息将基于该消息,并添加额外的标头。另请参阅 发布者确认和返回的替代机制。可选。
14 发送正向 (ack) 发布者确认的通道。有效负载是 confirm-correlation-expression 定义的相关数据。要求底层的 AsyncRabbitTemplateenableConfirms 属性设置为 true。另请参阅 发布者确认和返回的替代机制。可选(默认为 nullChannel)。
15 从 4.2 版开始。发送负向 (nack) 发布者确认的通道。有效负载是 confirm-correlation-expression 定义的相关数据。要求底层的 AsyncRabbitTemplateenableConfirms 属性设置为 true。另请参阅 发布者确认和返回的替代机制。可选(默认为 nullChannel)。
16 设置后,如果在毫秒数内未收到发布者确认,则网关将合成一个负确认 (nack)。挂起的确认每隔此值的一半进行检查,因此发送 nack 的实际时间将在此值的 1 倍到 1.5 倍之间。另请参阅 发布者确认和返回的替代机制。默认情况下为无(不会生成 nack)。
17 发送返回消息的通道。提供此通道时,将配置底层 AMQP 模板以将无法传递的消息返回到网关。消息由从 AMQP 接收的数据构建,并包含以下附加标头:amqp_returnReplyCodeamqp_returnReplyTextamqp_returnExchangeamqp_returnRoutingKey。要求底层的 AsyncRabbitTemplatemandatory 属性设置为 true。另请参阅 发布者确认和返回的替代机制。可选。
18 设置为 false 时,端点会在应用程序上下文初始化期间尝试连接到代理。这样做可以通过在代理关闭时记录错误消息来实现“快速失败”检测,以检测错误配置。当为 true(默认值)时,当发送第一条消息时,将建立连接(如果由于某些其他组件建立了连接而尚未存在)。

有关更多信息,另请参阅 异步服务激活器

RabbitTemplate

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