动态和运行时集成流
IntegrationFlow
及其所有依赖组件可以在运行时注册。在 5.0 版本之前,我们使用 BeanFactory.registerSingleton()
钩子。从 Spring Framework 5.0 开始,我们使用 instanceSupplier
钩子进行编程式 BeanDefinition
注册。以下示例展示了如何编程式注册一个 bean:
BeanDefinition beanDefinition =
BeanDefinitionBuilder.genericBeanDefinition((Class<Object>) bean.getClass(), () -> bean)
.getRawBeanDefinition();
((BeanDefinitionRegistry) this.beanFactory).registerBeanDefinition(beanName, beanDefinition);
请注意,在前面的示例中,instanceSupplier
钩子是 genericBeanDefinition
方法的最后一个参数,在这种情况下由一个 lambda 提供。
所有必要的 bean 初始化和生命周期都是自动完成的,就像标准上下文配置中的 bean 定义一样。
为了简化开发体验,Spring Integration 引入了 IntegrationFlowContext
在运行时注册和管理 IntegrationFlow
实例,如下例所示:
@Autowired
private AbstractServerConnectionFactory server1;
@Autowired
private IntegrationFlowContext flowContext;
...
@Test
public void testTcpGateways() {
TestingUtilities.waitListening(this.server1, null);
IntegrationFlow flow = f -> f
.handle(Tcp.outboundGateway(Tcp.netClient("localhost", this.server1.getPort())
.serializer(TcpCodecs.crlf())
.deserializer(TcpCodecs.lengthHeader1())
.id("client1"))
.remoteTimeout(m -> 5000))
.transform(Transformers.objectToString());
IntegrationFlowRegistration theFlow = this.flowContext.registration(flow).register();
assertThat(theFlow.getMessagingTemplate().convertSendAndReceive("foo", String.class), equalTo("FOO"));
}
当我们有多个配置选项并且必须创建多个类似的流程实例时,这非常有用。为此,我们可以迭代我们的选项并在循环中创建和注册 IntegrationFlow
实例。另一种情况是当我们的数据源不是基于 Spring 的,因此我们必须即时创建它。以下示例显示了反应式流事件源的这样一个示例:
Flux<Message<?>> messageFlux =
Flux.just("1,2,3,4")
.map(v -> v.split(","))
.flatMapIterable(Arrays::asList)
.map(Integer::parseInt)
.map(GenericMessage<Integer>::new);
QueueChannel resultChannel = new QueueChannel();
IntegrationFlow integrationFlow =
IntegrationFlow.from(messageFlux)
.<Integer, Integer>transform(p -> p * 2)
.channel(resultChannel)
.get();
this.integrationFlowContext.registration(integrationFlow)
.register();
IntegrationFlowRegistrationBuilder
(作为 IntegrationFlowContext.registration()
的结果)可用于指定要注册的 IntegrationFlow
的 bean 名称,控制其 autoStartup
,以及注册非 Spring Integration 的 bean。通常,这些额外的 bean 是连接工厂(AMQP、JMS、(S)FTP、TCP/UDP 等)、序列化器和反序列化器,或任何其他必需的支持组件。
你可以使用 IntegrationFlowRegistration.destroy()
回调来移除动态注册的 IntegrationFlow
及其所有依赖的bean,当你不再需要它们时。更多信息,请参阅 IntegrationFlowContext Javadoc。
从版本 5.0.6 开始,IntegrationFlow
定义中生成的所有 bean 名称都以前缀形式附加了流 ID。我们建议始终指定一个明确的流 ID。否则,在 IntegrationFlowContext
中会启动一个同步屏障,以生成 IntegrationFlow
的 bean 名称并注册其 beans。我们对这两个操作进行同步,以避免相同的生成 bean 名称可能被用于不同的 IntegrationFlow
实例时出现竞争条件。
此外,从 5.0.6 版开始,注册生成器 API 有一个新方法:useFlowIdAsPrefix()
。如果你希望声明同一流程的多个实例,并避免在流程中的组件具有相同 ID 时发生 bean 名称冲突,这将非常有用,如下例所示:
private void registerFlows() {
IntegrationFlowRegistration flow1 =
this.flowContext.registration(buildFlow(1234))
.id("tcp1")
.useFlowIdAsPrefix()
.register();
IntegrationFlowRegistration flow2 =
this.flowContext.registration(buildFlow(1235))
.id("tcp2")
.useFlowIdAsPrefix()
.register();
}
private IntegrationFlow buildFlow(int port) {
return f -> f
.handle(Tcp.outboundGateway(Tcp.netClient("localhost", port)
.serializer(TcpCodecs.crlf())
.deserializer(TcpCodecs.lengthHeader1())
.id("client"))
.remoteTimeout(m -> 5000))
.transform(Transformers.objectToString());
}
在这种情况下,第一个流程的消息处理器可以通过名为 tcp1.client.handler
的 bean 引用。
使用 useFlowIdAsPrefix()
时需要一个 id
属性。