出站网关
以下清单显示了 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
需要 RabbitTemplate
的 mandatory
属性设置为 true
,以及 CachingConnectionFactory
的 publisherReturns
属性设置为 true
。
当使用多个出站端点进行返回时,每个端点需要一个单独的 RabbitTemplate
。
底层的 AmqpTemplate
默认 replyTimeout
为五秒。
如果需要更长的超时时间,必须在 template
上进行配置。
请注意,出站适配器和出站网关配置之间唯一的区别是 expectReply
属性的设置。