Spring Integration

Appearance

Spring Integration Pollers 轮询器详解 ​

1. 什么是 Pollers？ ​

在 Spring Integration 中，Pollers（轮询器） 是用于控制消息端点（如 MessageSource）如何主动检查新消息的核心机制。就像⏰ 定时检查邮箱一样，轮询器决定了系统以什么频率检查消息源是否有新消息到达。

1.1 轮询器的工作原理 ​

2. Kotlin 轮询器配置实战 ​

2.1 基础轮询配置 ​

kotlin
import org.springframework.context.annotation.Bean
import org.springframework.integration.dsl.Pollers
import org.springframework.integration.scheduling.PollerMetadata

@Configuration
class IntegrationConfig {

    //  // 默认轮询器配置
    @Bean(name = [PollerMetadata.DEFAULT_POLLER])
    fun defaultPoller(): PollerMetadata = Pollers
        .fixedRate(500)                  // 每500ms轮询一次
        .maxMessagesPerPoll(10)          // 每次最多处理10条消息
        .errorChannel("errorChannel")    // 错误处理通道
        .get()                           // 获取PollerMetadata实例
}

2.2 配置项详解 ​

配置方法说明默认值
fixedRate(millis)固定频率轮询必需
fixedDelay(millis)固定延迟轮询-
cron(expression)Cron表达式调度-
maxMessagesPerPoll(n)每次轮询最大消息数-1(无限制)
errorChannel(name)错误处理通道无
transactional()启用事务禁用

2.3 带事务的轮询配置 ​

kotlin
@Bean
fun transactionalPoller(): PollerMetadata = Pollers
    .fixedDelay(1000)
    .maxMessagesPerPoll(5)
    .transactional { //  // 事务配置
        it.isolation(TransactionDefinition.ISOLATION_READ_COMMITTED)
        it.propagation(TransactionDefinition.PROPAGATION_REQUIRED)
    }
    .get()

3. 关键注意事项 ​

WARNING

FactoryBean 特殊处理
当使用 DSL 配置 PollerSpec 时，切勿在 Bean 定义中调用 getObject() 方法：

kotlin
// ❌ 错误做法
@Bean
fun badPoller() = Pollers.fixedRate(500).getObject()

// ✅ 正确做法
@Bean
fun goodPoller() = Pollers.fixedRate(500)

PollerSpec 本身是 FactoryBean，Spring 会自动将其转换为 PollerMetadata

IMPORTANT

轮询策略选择指南

  • 使用 fixedRate：当需要严格定时执行，不考虑上次执行时长
  • 使用 fixedDelay：当需要保证执行间隔，避免重叠执行
  • 使用 cron：复杂调度需求（如工作日特定时间）

4. 错误处理最佳实践 ​

4.1 错误通道配置 ​

kotlin
@Bean
fun errorFlow() = integrationFlow("errorChannel") {
    handle { message: Message<*> ->
        val exception = message.payload as Throwable
        logger.error("消息处理失败: ${exception.message}")
        // 添加自定义恢复/报警逻辑
    }
}

@Bean
fun resilientPoller() = Pollers
    .fixedRate(1000)
    .errorChannel("errorChannel") // [!code highlight] // 绑定错误通道
    .get()

4.2 错误处理模式对比 ​

kotlin
// 轮询器内直接处理错误
Pollers.fixedRate(500)
   .errorHandler { ex ->
        logger.error("轮询异常", ex)
        // 简单场景适用
    }
kotlin
// 通过错误通道解耦处理
Pollers.fixedRate(500)
   .errorChannel("errorChannel")
   // 复杂错误处理场景推荐

5. 性能优化技巧 ​

5.1 动态调整轮询频率 ​

kotlin
@Bean
fun adaptivePoller(): PollerMetadata {
    val dynamicRate = AtomicLong(1000)

    return Pollers.fixedRate { dynamicRate.get() }
        .advice(object : RequestHandlerRetryAdvice() {
            override fun doInvoke(callback: RequestHandlerRetryAdvice.Callback): Any? {
                try {
                    return callback.execute()
                } catch (e: Exception) {
                    dynamicRate.updateAndGet { rate ->
                        min(rate * 2, 10000) // 错误时降低频率
                    }
                    throw e
                }
            }
        })
        .get()
}

5.2 批处理优化 ​

kotlin
@Bean
fun batchPoller() = Pollers.fixedRate(5000)  // 5秒批次
    .maxMessagesPerPoll(100)                // 每批100条
    .taskExecutor(ThreadPoolTaskExecutor().apply {
        corePoolSize = 4
        maxPoolSize = 10
        setQueueCapacity(100)
    })
    .get()

6. 常见问题解决 ​

6.1 消息积压问题 ​

CAUTION

症状：消息处理速度跟不上生产速度
解决方案：

  1. 增加 maxMessagesPerPoll
  2. 使用 TaskExecutor 添加线程池
  3. 优化下游处理器性能
kotlin
Pollers.fixedRate(500)
   .maxMessagesPerPoll(50) // [!code ++] // 增加每次处理量
   .taskExecutor(SimpleAsyncTaskExecutor()) // [!code ++] // 添加异步处理

6.2 资源竞争问题 ​

NOTE

当多个轮询器访问共享资源（如文件系统）时：
✅ 使用 FileSystemPersistentAcceptOnceFileListFilter
✅ 配置分布式锁（如通过 Redis）

kotlin
@Bean
fun fileReadingPoller() = Pollers.fixedDelay(2000)
    .advice(lockInterceptor()) // 添加锁拦截器
    .get()

private fun lockInterceptor(): Advice = RedisLockRegistry(redisTemplate)
    .obtain("fileLock")
    .lockInterruptibly(30, TimeUnit.SECONDS)

7. 进阶：响应式轮询器 ​

Spring Integration 5.5+ 支持响应式轮询：

kotlin
@Bean
fun reactivePoller() = Pollers
    .from(Flux.interval(Duration.ofMillis(200)))
    .maxMessagesPerPoll(10)
    .get()

响应式 vs 传统轮询 ​

特性传统轮询响应式轮询
线程模型阻塞线程非阻塞
背压支持无自动支持
资源消耗较高较低
适用场景简单同步任务高并发IO密集型

总结 ​

Spring Integration Pollers 是消息驱动架构的核心调度机制，关键要点：

  1. 配置优先：使用 Kotlin DSL 配置更简洁安全
  2. 错误隔离：通过专用通道处理错误
  3. 动态调整：根据系统负载智能调整轮询参数
  4. 资源保护：使用事务/锁机制保护共享资源
  5. 响应式演进：高并发场景考虑响应式实现

"轮询器如同系统的心跳，合理配置才能保证消息血液的健康流动" 💓