跳到主要内容

异步 outbound 网关

QWen Plus 中英对照 Asynchronous Outbound Gateway

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

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

@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);

    }

}
java

  • 此适配器的唯一 ID。可选。

  • 消息应发送到的消息通道，以便将它们转换并发布到 AMQP 交换。必需。

  • 配置的 AsyncRabbitTemplate 的 Bean 引用。可选（默认为 asyncRabbitTemplate）。

  • 消息应发送到的 AMQP 交换的名称。如果没有提供，则消息被发送到默认的无名交换。与 'exchange-name-expression' 互斥。可选。

  • 一个 SpEL 表达式，用于确定消息应发送到的 AMQP 交换的名称，以消息作为根对象。如果没有提供，则消息被发送到默认的无名交换。与 'exchange-name' 互斥。可选。

  • 当注册多个消费者时，此消费者的顺序，从而启用负载均衡和故障转移。可选（默认为 Ordered.LOWEST_PRECEDENCE [=Integer.MAX_VALUE]）。

  • 将回复消息在从 AMQP 队列接收并转换后发送到的消息通道。可选。

  • 网关在向 reply-channel 发送回复消息时等待的时间。这仅适用于 reply-channel 可能阻塞的情况——例如当前已满的具有容量限制的 QueueChannel。默认值为无穷大。

  • 当在 AsyncRabbitTemplate 的 `receiveTimeout 属性时间内未收到回复消息且此设置为 true 时，网关会将错误消息发送到入站消息的 errorChannel 头。当在 AsyncRabbitTemplate 的 `receiveTimeout 属性时间内未收到回复消息且此设置为 false 时，网关会将错误消息发送到默认的 errorChannel（如果可用）。默认为 true。

  • 发送消息时使用的路由键。默认情况下，这是一个空的 String。与 'routing-key-expression' 互斥。可选。

  • 一个 SpEL 表达式，用于确定发送消息时使用的路由键，以消息作为根对象（例如，'payload.key'）。默认情况下，这是一个空的 String。与 'routing-key' 互斥。可选。

  • 消息的默认传输模式：PERSISTENT 或 NON_PERSISTENT。如果 header-mapper 设置了传输模式，则覆盖该设置。如果存在 Spring Integration 消息头 (amqp_deliveryMode)，则 DefaultHeaderMapper 设置该值。如果未提供此属性且头映射器未设置它，则默认取决于底层 Spring AMQP MessagePropertiesConverter，由 RabbitTemplate 使用。如果未自定义，则默认为 PERSISTENT。可选。

  • 定义关联数据的表达式。当提供时，这会配置底层 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 通道上发出的消息基于该消息，并添加额外的头。另请参阅 发布确认和返回的替代机制。可选。

  • 正面 (ack) 发布确认发送到的通道。有效载荷是由 confirm-correlation-expression 定义的关联数据。要求底层 AsyncRabbitTemplate 的 enableConfirms 属性设置为 true。另请参阅 发布确认和返回的替代机制。可选（默认为 nullChannel）。

  • 自版本 4.2 起。负面 (nack) 发布确认发送到的通道。有效载荷是由 confirm-correlation-expression 定义的关联数据。要求底层 AsyncRabbitTemplate 的 enableConfirms 属性设置为 true。另请参阅 发布确认和返回的替代机制。可选（默认为 nullChannel）。

  • 当设置时，如果在指定毫秒内未收到发布确认，网关将合成一个负面确认（nack）。每 50% 的这个值检查一次待确认项，因此实际发送 nack 的时间将在 1x 到 1.5x 这个值之间。另请参阅 发布确认和返回的替代机制。默认无（不会生成 nacks）。

  • 返回消息发送到的通道。当提供时，底层 AMQP 模板配置为将不可投递的消息返回给网关。消息由从 AMQP 接收到的数据构造而成，并带有以下附加头：amqp_returnReplyCode、amqp_returnReplyText、amqp_returnExchange 和 amqp_returnRoutingKey。要求底层 AsyncRabbitTemplate 的 mandatory 属性设置为 true。另请参阅 发布确认和返回的替代机制。可选。

  • 当设置为 false 时，端点会在应用程序上下文初始化期间尝试连接到代理。这样做可以通过记录错误消息来快速检测不良配置（如果代理不可用）。当为 true（默认）时，在发送第一条消息时建立连接（如果由于其他组件已经建立了连接而不存在）。

更多信息请参见 异步服务激活器。

important

RabbitTemplate

当你使用确认和返回功能时，我们建议注入到 AsyncRabbitTemplate 中的 RabbitTemplate 应该是专用的。否则，可能会遇到意外的副作用。