Spring Integration Kotlin DSL聚合器配置编译报错求助

问题背景 #

想要聚合消息列表以使用AmqpOutboundEndpoint.multiSend选项，参考Stack Overflow上的XML解决方案后改用Kotlin DSL编写代码，却出现重载解析歧义的编译错误。使用环境为Spring Integration 6.2.1、Kotlin 1.9.22。

代码示例 #

@Configuration
@EnableIntegration
class SampleConfiguration {

  @Bean
  fun sampleFlow(amqpTemplate: AmqpTemplate): StandardIntegrationFlow {
    return IntegrationFlow
      .from("inputChannel")
      .aggregate(Consumer<AggregatorSpec> {
        it.releaseExpression("size() == 100")
          .groupTimeout(1000)
          .sendPartialResultOnExpiry(true)
          .correlationExpression("T(Thread).currentThread().id")
          .poller { p: PollerFactory -> p.fixedRate(1000).maxMessagesPerPoll(100) }
      })
      .handle(Amqp.outboundAdapter(amqpTemplate).exchangeName("sampleExchange").multiSend(true))
      .get()
  }
}

编译错误 #

Overload resolution ambiguity: 
public open fun aggregate(aggregator: Consumer<AggregatorSpec!>?): IntegrationFlowBuilder defined in org.springframework.integration.dsl.IntegrationFlowBuilder
public open fun aggregate(aggregatorProcessor: Any): IntegrationFlowBuilder defined in org.springframework.integration.dsl.IntegrationFlowBuilder

解决方案 #

这个歧义是因为Kotlin的类型推断无法区分接受Consumer<AggregatorSpec>和Any的两个aggregate重载。可以通过以下两种方式消除歧义：

方式一：简化lambda，显式指定参数类型 #

利用Kotlin的SAM转换特性，直接传递带参数类型的lambda，无需显式包裹Consumer<AggregatorSpec>，让编译器明确匹配正确的重载：

@Configuration
@EnableIntegration
class SampleConfiguration {

  @Bean
  fun sampleFlow(amqpTemplate: AmqpTemplate): StandardIntegrationFlow {
    return IntegrationFlow
      .from("inputChannel")
      .aggregate { spec: AggregatorSpec ->
        spec.releaseExpression("size() == 100")
          .groupTimeout(1000)
          .sendPartialResultOnExpiry(true)
          .correlationExpression("T(Thread).currentThread().id")
          .poller { p -> p.fixedRate(1000).maxMessagesPerPoll(100) }
      }
      .handle(Amqp.outboundAdapter(amqpTemplate).exchangeName("sampleExchange").multiSend(true))
      .get()
  }
}

方式二：使用Aggregators.spec()构建配置 #

通过Aggregators.spec()直接创建AggregatorSpec实例，传递给aggregate方法，完全避免类型推断歧义：

@Configuration
@EnableIntegration
class SampleConfiguration {

  @Bean
  fun sampleFlow(amqpTemplate: AmqpTemplate): StandardIntegrationFlow {
    return IntegrationFlow
      .from("inputChannel")
      .aggregate(Aggregators.spec()
        .releaseExpression("size() == 100")
        .groupTimeout(1000)
        .sendPartialResultOnExpiry(true)
        .correlationExpression("T(Thread).currentThread().id")
        .poller { p -> p.fixedRate(1000).maxMessagesPerPoll(100) })
      .handle(Amqp.outboundAdapter(amqpTemplate).exchangeName("sampleExchange").multiSend(true))
      .get()
  }
}

说明 #

修改后的代码会正确配置聚合器，将消息聚合成列表，随后Amqp.outboundAdapter的multiSend(true)会把列表中的消息批量发送到指定AMQP交换机。

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