入站通道适配器

通常，消息流从入站通道适配器（例如 <int-jdbc:inbound-channel-adapter>）开始。适配器使用 <poller> 配置，并定期请求 MessageSource<?> 生成消息。Java DSL 也允许从 MessageSource<?> 启动 IntegrationFlow。为此，IntegrationFlow 流式 API 提供了重载的 IntegrationFlow.from(MessageSource<?> messageSource) 方法。您可以将 MessageSource<?> 配置为 bean 并将其作为该方法的参数提供。IntegrationFlow.from() 的第二个参数是 Consumer<SourcePollingChannelAdapterSpec> lambda，它允许您为 SourcePollingChannelAdapter 提供选项（例如 PollerMetadata 或 SmartLifecycle）。以下示例展示了如何使用流式 API 和 lambda 创建 IntegrationFlow

@Bean
public MessageSource<Object> jdbcMessageSource() {
    return new JdbcPollingChannelAdapter(this.dataSource, "SELECT * FROM something");
}

@Bean
public IntegrationFlow pollingFlow() {
    return IntegrationFlow.from(jdbcMessageSource(),
                c -> c.poller(Pollers.fixedRate(100).maxMessagesPerPoll(1)))
            .transform(Transformers.toJson())
            .channel("furtherProcessChannel")
            .get();
}

对于那些不需要直接构建Message对象的用例，您可以使用基于java.util.function.Supplier的IntegrationFlow.fromSupplier()变体。Supplier.get()的结果会自动包装在Message中（如果它本身不是Message）。