消息通道
除了带有 EIP 方法的 IntegrationFlowBuilder 之外,Java DSL 还提供了一个流畅的 API 来配置 MessageChannel 实例。为此,提供了 MessageChannels 构建器工厂。以下示例展示了如何使用它:
@Bean
public PriorityChannelSpec priorityChannel() {
return MessageChannels.priority(this.mongoDbChannelMessageStore, "priorityGroup")
.interceptor(wireTap());
}
相同的 MessageChannels 构建器工厂可以在 IntegrationFlowBuilder 的 channel() EIP 方法中使用,以连接端点,类似于在 XML 配置中连接 input-channel/output-channel 对。默认情况下,端点与 DirectChannel 实例连接,其中 bean 名称基于以下模式:[IntegrationFlow.beanName].channel#[channelNameIndex]。此规则也适用于通过内联 MessageChannels 构建器工厂使用而产生的未命名通道。但是,所有 MessageChannels 方法都有一个知道 channelId 的变体,您可以使用它来设置 MessageChannel 实例的 bean 名称。MessageChannel 引用和 beanName 可以用作 bean 方法调用。以下示例显示了使用 channel() EIP 方法的可能方式:
@Bean
public QueueChannelSpec queueChannel() {
return MessageChannels.queue();
}
@Bean
public PublishSubscribeChannelSpec<?> publishSubscribe() {
return MessageChannels.publishSubscribe();
}
@Bean
public IntegrationFlow channelFlow() {
return IntegrationFlow.from("input")
.fixedSubscriberChannel()
.channel("queueChannel")
.channel(publishSubscribe())
.channel(MessageChannels.executor("executorChannel", this.taskExecutor))
.channel("output")
.get();
}
-
from("input")意味着“'查找并使用 ID 为 "input" 的MessageChannel,或者创建一个'”。 -
fixedSubscriberChannel()生成一个FixedSubscriberChannel实例并将其注册为channelFlow.channel#0。 -
channel("queueChannel")的工作方式相同,但使用现有的queueChannelbean。 -
channel(publishSubscribe())是 bean 方法引用。 -
channel(MessageChannels.executor("executorChannel", this.taskExecutor))是IntegrationFlowBuilder,它将IntegrationComponentSpec暴露给ExecutorChannel并将其注册为executorChannel。 -
channel("output")将DirectChannelbean 注册为output,只要不存在同名的 bean。
Note: 前面的 IntegrationFlow 定义是有效的,并且其所有通道都应用于带有 BridgeHandler 实例的端点。
请注意不要在不同的 IntegrationFlow 实例中通过 MessageChannels 工厂使用相同的内联通道定义。即使 DSL 解析器将不存在的对象注册为 bean,它也无法从不同的 IntegrationFlow 容器中确定相同的对象 (MessageChannel)。以下示例是错误的:
@Bean
public IntegrationFlow startFlow() {
return IntegrationFlow.from("input")
.transform(...)
.channel(MessageChannels.queue("queueChannel"))
.get();
}
@Bean
public IntegrationFlow endFlow() {
return IntegrationFlow.from(MessageChannels.queue("queueChannel"))
.handle(...)
.get();
}
该错误示例的结果是以下异常:
Caused by: java.lang.IllegalStateException:
Could not register object [queueChannel] under bean name 'queueChannel':
there is already object [queueChannel] bound
at o.s.b.f.s.DefaultSingletonBeanRegistry.registerSingleton(DefaultSingletonBeanRegistry.java:129)
为了使其正常工作,您需要为该通道声明 @Bean,并从不同的 IntegrationFlow 实例中使用其 bean 方法。