出站网关

以下清单显示了 AMQP 出站网关的可能属性:

  • Java DSL

  • Java

  • XML

@Bean
public IntegrationFlow amqpOutbound(AmqpTemplate amqpTemplate) {
    return f -> f.handle(Amqp.outboundGateway(amqpTemplate)
                    .routingKey("foo")) // default exchange - route to queue 'foo'
            .get();
}

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

    String sendToRabbit(String data);

}
@Bean
@ServiceActivator(inputChannel = "amqpOutboundChannel")
public AmqpOutboundEndpoint amqpOutbound(AmqpTemplate amqpTemplate) {
    AmqpOutboundEndpoint outbound = new AmqpOutboundEndpoint(amqpTemplate);
    outbound.setExpectReply(true);
    outbound.setRoutingKey("foo"); // default exchange - route to queue 'foo'
    return outbound;
}

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

@MessagingGateway(defaultRequestChannel = "amqpOutboundChannel")
public interface MyGateway {

    String sendToRabbit(String data);

}
<int-amqp:outbound-gateway id="outboundGateway"               [id="CO1-1"]1
                           request-channel="myRequestChannel" [id="CO1-2"]2
                           amqp-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
                           error-message-strategy=""          [id="CO1-18"]18
                           lazy-connect="true" />             [id="CO1-19"]19
 <1> 此适配器的唯一 ID。
可选。
 <1> 消息通道,消息被发送到此通道,以便将其转换并发布到 AMQP 交换器。
必填。
 <1> 对已配置的 AMQP 模板的 Bean 引用。
可选(默认为 `amqpTemplate`)。
 <1> 消息应发送到的 AMQP 交换器的名称。
如果未提供,消息将发送到默认的无名交换器。
与 'exchange-name-expression' 互斥。
可选。
 <1> 一个 SpEL 表达式,用于确定消息应发送到的 AMQP 交换器的名称,消息作为根对象。
如果未提供,消息将发送到默认的无名交换器。
与 'exchange-name' 互斥。
可选。
 <1> 注册多个消费者时,此消费者的顺序,从而实现负载均衡和故障转移。
可选(默认为 `Ordered.LOWEST_PRECEDENCE [=Integer.MAX_VALUE]`)。
 <1> 消息通道,从 AMQP 队列接收并转换后,回复应发送到此通道。
可选。
 <1> 网关在将回复消息发送到 `reply-channel` 时等待的时间。
这仅适用于 `reply-channel` 可以阻塞的情况——例如,具有容量限制且当前已满的 `QueueChannel`。
默认为无限。
 <1> 当为 `true` 时,如果在 `AmqpTemplate` 的 `replyTimeout` 属性内没有收到回复消息,网关将抛出异常。
默认为 `true`。
 <1> 发送消息时使用的 `routing-key`。
默认情况下,这是一个空 `String`。
与 'routing-key-expression' 互斥。
可选。
 <1> 一个 SpEL 表达式,用于确定发送消息时使用的 `routing-key`,消息作为根对象(例如,'payload.key')。
默认情况下,这是一个空 `String`。
与 'routing-key' 互斥。
可选。
 <1> 消息的默认投递模式:`PERSISTENT` 或 `NON_PERSISTENT`。
如果 `header-mapper` 设置了投递模式,则此设置将被覆盖。
如果存在 Spring Integration 消息头 `amqp_deliveryMode`,则 `DefaultHeaderMapper` 会设置其值。
如果未提供此属性且头映射器未设置它,则默认值取决于 `RabbitTemplate` 使用的底层 Spring AMQP `MessagePropertiesConverter`。
如果完全没有自定义,默认值为 `PERSISTENT`。
可选。
 <1> 自 4.2 版本起。
定义关联数据的表达式。
提供时,此设置会配置底层 AMQP 模板以接收发布者确认。
需要专用的 `RabbitTemplate` 和 `CachingConnectionFactory`,其中 `publisherConfirms` 属性设置为 `true`。
当收到发布者确认并提供了关联数据时,它会根据确认类型写入 `confirm-ack-channel` 或 `confirm-nack-channel`。
确认的有效负载是关联数据,由该表达式定义。
消息具有一个头 'amqp_publishConfirm',设置为 `true` (`ack`) 或 `false` (`nack`)。
对于 `nack` 确认,Spring Integration 提供了一个额外的头 `amqp_publishConfirmNackCause`。
示例:`headers['myCorrelationData']` 和 `payload`。
如果表达式解析为 `Message<?>` 实例(例如 `#this`),则在 `ack`/`nack` 通道上发出的消息将基于该消息,并添加额外的头。
以前,无论类型如何,都会创建一个新消息,并将关联数据作为其有效负载。
另请参阅 xref:amqp/alternative-confirms-returns.adoc[发布者确认和返回的替代机制]。
可选。
 <1> 接收肯定 (`ack`) 发布者确认的通道。
有效负载是 `confirm-correlation-expression` 定义的关联数据。
如果表达式是 `#root` 或 `#this`,则消息将从原始消息构建,并将 `amqp_publishConfirm` 头设置为 `true`。
另请参阅 xref:amqp/alternative-confirms-returns.adoc[发布者确认和返回的替代机制]。
可选(默认值为 `nullChannel`)。
 <1> 接收否定 (`nack`) 发布者确认的通道。
有效负载是 `confirm-correlation-expression` 定义的关联数据(如果没有配置 `ErrorMessageStrategy`)。
如果表达式是 `#root` 或 `#this`,则消息将从原始消息构建,并将 `amqp_publishConfirm` 头设置为 `false`。
当存在 `ErrorMessageStrategy` 时,消息是一个 `ErrorMessage`,其有效负载为 `NackedAmqpMessageException`。
另请参阅 xref:amqp/alternative-confirms-returns.adoc[发布者确认和返回的替代机制]。
可选(默认值为 `nullChannel`)。
 <1> 设置后,如果在指定时间内(毫秒)未收到发布者确认,网关将合成一个否定确认 (nack)。
待处理的确认每隔该值的 50% 检查一次,因此发送 nack 的实际时间将在该值的 1 倍到 1.5 倍之间。
默认无(不生成 nack)。
 <1> 返回消息发送到的通道。
提供时,底层 AMQP 模板被配置为将无法投递的消息返回给适配器。
当没有配置 `ErrorMessageStrategy` 时,消息将从 AMQP 接收到的数据构建,并包含以下附加头:`amqp_returnReplyCode`、`amqp_returnReplyText`、`amqp_returnExchange` 和 `amqp_returnRoutingKey`。
当存在 `ErrorMessageStrategy` 时,消息是一个 `ErrorMessage`,其有效负载为 `ReturnedAmqpMessageException`。
另请参阅 xref:amqp/alternative-confirms-returns.adoc[发布者确认和返回的替代机制]。
可选。
 <1> 对 `ErrorMessageStrategy` 实现的引用,用于在发送返回或否定确认消息时构建 `ErrorMessage` 实例。
 <1> 当设置为 `false` 时,端点会在应用程序上下文初始化期间尝试连接到 broker。
这允许通过在 broker 关闭时记录错误消息来“快速失败”检测错误配置。
当为 `true`(默认值)时,连接会在发送第一条消息时建立(如果由于其他组件建立而尚未存在)。
Example 1. return-channel

使用 return-channel 需要 RabbitTemplatemandatory 属性设置为 true,以及 CachingConnectionFactorypublisherReturns 属性设置为 true。 当使用多个出站端点进行返回时,每个端点需要一个单独的 RabbitTemplate

底层的 AmqpTemplate 默认 replyTimeout 为五秒。 如果需要更长的超时时间,必须在 template 上进行配置。

请注意,出站适配器和出站网关配置之间唯一的区别是 expectReply 属性的设置。