对于最新的稳定版本,请使用 Spring Integration 6.4.0! |
异步出站网关
上一节中讨论的网关是同步的,因为发送线程将暂停,直到
收到回复(或发生超时)。
Spring 集成版本 4.3 添加了一个异步网关,它使用AsyncRabbitTemplate
来自 Spring AMQP。
发送消息时,线程会在发送作完成后立即返回,收到消息时,将在模板的侦听器容器线程上发送回复。
当在 poller 线程上调用网关时,这可能很有用。
线程已释放,可用于框架中的其他任务。
以下清单显示了 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 | Bean 引用配置的AsyncRabbitTemplate .
可选(默认为asyncRabbitTemplate ). |
4 | 应将消息发送到的 AMQP 交换的名称。 如果未提供,则消息将发送到默认的 no-name 交换。 与 'exchange-name-expression' 互斥。 自选。 |
5 | 一个 SPEL 表达式,用于确定消息发送到的 AMQP 交换的名称,并将消息作为根对象。 如果未提供,则消息将发送到默认的 no-name 交换。 与 'exchange-name' 互斥。 自选。 |
6 | 注册多个使用者时此使用者的顺序,从而启用负载均衡和故障转移。
可选(默认为Ordered.LOWEST_PRECEDENCE [=Integer.MAX_VALUE] ). |
7 | 从 AMQP 队列接收回复并进行转换后,应将回复发送到的消息通道。 自选。 |
8 | 网关在将回复消息发送到reply-channel .
这仅适用于reply-channel can 阻止 — 例如QueueChannel 容量限制当前已满。
默认值为 infinity。 |
9 | 当AsyncRabbitTemplate’s `receiveTimeout 属性,并且该设置为true ,网关会向入站消息的errorChannel 页眉。
当AsyncRabbitTemplate’s `receiveTimeout 属性,并且该设置为false ,网关会向默认的errorChannel (如果可用)。
它默认为true . |
10 | 发送 Messages 时使用的 routing-key。
默认情况下,这是一个空的String .
与 'routing-key-expression' 互斥。
自选。 |
11 | 一个 SPEL 表达式,用于确定发送消息时要使用的路由密钥。
将消息作为根对象(例如,'payload.key')。
默认情况下,这是一个空的String .
与 'routing-key' 互斥。
自选。 |
12 | 消息的默认传递模式:PERSISTENT 或NON_PERSISTENT .
如果header-mapper 设置传递模式。
如果 Spring 集成消息头 (amqp_deliveryMode ) 存在,则DefaultHeaderMapper 设置值。
如果未提供此属性并且 Headers 映射器未设置它,则默认值取决于底层的 Spring AMQPMessagePropertiesConverter 由RabbitTemplate .
如果未自定义,则默认值为PERSISTENT .
自选。 |
13 | 定义相关数据的表达式。
如果提供,这会将底层 AMQP 模板配置为接收发布者确认。
需要专用的RabbitTemplate 以及CachingConnectionFactory 及其publisherConfirms 属性设置为true .
收到发布者确认并提供关联数据后,确认将写入confirm-ack-channel 或confirm-nack-channel ,具体取决于确认类型。
确认的有效负载是此表达式定义的关联数据,并且消息的 'amqp_publishConfirm' 标头设置为true (ack ) 或false (nack ).
为nack 实例、附加标头 (amqp_publishConfirmNackCause ) 提供。
例子:headers['myCorrelationData'] ,payload .
如果表达式解析为Message<?> 实例(例如 “#this”)上,则在ack /nack channel 基于该消息,并添加了其他标头。
另请参阅发布者确认和返回的替代机制。
自选。 |
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 倍之间。 另请参阅发布者确认和返回的替代机制。 默认无(不会生成 nack)。 |
17 | 返回的消息发送到的通道。
如果提供,则底层 AMQP 模板配置为将无法送达的消息返回到网关。
该消息是根据从 AMQP 接收的数据构建的,并带有以下附加标头:amqp_returnReplyCode ,amqp_returnReplyText ,amqp_returnExchange 和amqp_returnRoutingKey .
需要底层AsyncRabbitTemplate 使其mandatory 属性设置为true .
另请参阅发布者确认和返回的替代机制。
自选。 |
18 | 当设置为false 时,终端节点会尝试在应用程序上下文初始化期间连接到代理。
这样做可以在代理宕机时记录错误消息,从而“快速失败”检测错误配置。
什么时候true (默认值),则连接已建立(如果由于已建立其他组件而尚不存在
it) 的 API 请求。 |
有关更多信息,另请参阅 Asynchronous Service Activator。
RabbitTemplate (兔模板)
当您使用 Confirmation 和 Returns 时,我们建议 |