能否在Spring Integration DSL的IntegrationFlowAdapter中为ServiceActivator加延迟？

问题场景 #

我现有如下Spring Integration适配器代码：

@Component
class MyCustomFlow : IntegrationFlowAdapter() {

    fun singleThreadTaskExecutor(): TaskExecutor {
        val executor = ThreadPoolTaskExecutor()
        executor.maxPoolSize = 1
        executor.initialize()
        return executor
    }

    @Filter
    fun filter(data: SomeData): Boolean = ...

    @Transformer
    fun transform(customer: Data): Message<SomeData> {
      ....
    }

    @ServiceActivator
    fun handle(data: Data): SomeData {
       ....
    }

    @Bean(name = [PollerMetadata.DEFAULT_POLLER])
    fun poller(): PollerSpec? {
        return Pollers.fixedRate(500)
    }

    @Bean
    override fun buildFlow(): IntegrationFlowDefinition<*> {
        return from(MessageChannels.queue("updateCustomersLocation"))
            .channel(MessageChannels.executor(singleThreadTaskExecutor()))
            .split()
            .filter(this)
            .transform(this)
            .handle(this)
            .channel("customerLocationFetched")
    }
}

我知道在适配器外部定义@ServiceActivator时，可以通过poller属性设置消息处理间隔，但目前只能通过新增中间通道，把handle逻辑移到适配器外部来实现延迟。想请教：能否直接在适配器内部给handle对应的ServiceActivator添加类似延迟，以此控制请求发送速率？

解决方案 #

不需要拆分到外部，有三种方式可以在适配器内部实现需求：

1. 在DSL的handle()端点直接配置轮询器

通过handle()方法的配置lambda，为ServiceActivator端点指定轮询器。注意需要确保handle的输入通道是可轮询通道（如QueueChannel），轮询器才能生效：

@Bean
override fun buildFlow(): IntegrationFlowDefinition<*> {
    return from(MessageChannels.queue("updateCustomersLocation"))
        .channel(MessageChannels.executor(singleThreadTaskExecutor()))
        .split()
        .filter(this)
        .transform(this)
        // 新增临时队列通道，为轮询提供可轮询的消息源
        .channel(MessageChannels.queue())
        // 配置handle端点的轮询器
        .handle(this, "handle") { spec ->
            spec.poller(Pollers.fixedDelay(3000)
                .maxMessagesPerPoll(1))
        }
        .channel("customerLocationFetched")
}

这里明确指定方法名"handle"，配合配置lambda实现轮询规则。

2. 直接在适配器内部的handle方法上配置@ServiceActivator的poller属性

直接给适配器内的handle方法添加poller配置，同样需要确保输入通道为可轮询类型：

@ServiceActivator(
    poller = [Poller(fixedDelay = "3000", maxMessagesPerPoll = "1")]
)
fun handle(data: Data): SomeData {
   // 原有处理逻辑
}

此时需要在transform之后新增QueueChannel，保证轮询器能触发间隔处理。

3. 使用delay()组件实现消息延迟

如果只是需要固定延迟，无需依赖轮询器，可以直接在handle前添加delay()处理器：

@Bean
override fun buildFlow(): IntegrationFlowDefinition<*> {
    return from(MessageChannels.queue("updateCustomersLocation"))
        .channel(MessageChannels.executor(singleThreadTaskExecutor()))
        .split()
        .filter(this)
        .transform(this)
        // 添加固定延迟，每个消息延迟3000ms后再处理
        .delay("delayGroup") { d -> d.fixedDelay(3000) }
        .handle(this)
        .channel("customerLocationFetched")
}

这种方式无需修改通道类型，直接为每个消息添加延迟，同样能达到控制发送速率的效果。

补充说明 #

之前需要拆分到外部的原因：轮询器仅对可轮询通道生效，如果handle的输入是订阅型通道（如DirectChannel），直接配置@ServiceActivator的poller属性不会起作用。只要确保输入通道为可轮询类型，就能在适配器内部完成配置。

内容的提问来源于stack exchange，提问作者user19840892