Spring Cloud Stream自定义路由器：如何为消费者指定正确输入？

背景 #

此前仅声明了如下Consumer Bean：

@Bean
public Consumer<Message<MyClass<T>>> myFunction() {
    return ...;
}

对应的配置：

spring:
  cloud:
    stream:
      bindings:
        myFunction-in-0: "input"
        input:
          destination: "my-kinesis"
          binder: kinesis
          group: "my-group"

新增配置后出现错误 #

添加自定义路由器返回FunctionRoutingResult：

return new MessageRoutingCallback.FunctionRoutingResult(
        myCondition ? "myFunction" : "devNull"
);

同时新增devNull Consumer：

@Bean
public Consumer<Message<MyClass<?>>> devNull() {
    return ...;
}

启动时出现以下错误：

Failed to bind properties under 'spring.cloud.stream.bindings.myfunction-in-0' to org.springframework.cloud.stream.config.BindingProperties:

Reason: org.springframework.core.convert.ConverterNotFoundException: No converter found capable of converting from type [java.lang.String] to type [org.springframework.cloud.stream.config.BindingProperties]

Action:

Update your application's configuration

使用版本：spring-cloud-stream 3.2.4，spring-boot-starter-parent 2.6.6，spring-cloud-dependencies 2021.0.3

疑问 #

• 是否是消息泛型导致的问题？
• 能否自动解析为POJO？
• 自定义路由器场景下，消费者的正确输入配置是什么？

解决方案 #

1. 修正配置格式错误 #

原配置中myFunction-in-0: "input"的写法不符合Spring Cloud Stream绑定配置规范，当使用自定义路由器时，这种赋值方式会导致Spring无法将字符串"input"转换为BindingProperties类型。

正确的配置应该移除myFunction-in-0: "input"这一行，路由器会动态将消息路由到对应的Consumer，无需单独为每个Consumer绑定到同一个输入：

spring:
  cloud:
    stream:
      bindings:
        input:
          destination: "my-kinesis"
          binder: kinesis
          group: "my-group"

如果需要为某个Consumer单独配置绑定属性（如内容类型），需采用节点式配置：

spring:
  cloud:
    stream:
      bindings:
        input:
          destination: "my-kinesis"
          binder: kinesis
          group: "my-group"
        myFunction-in-0:
          content-type: application/json
        devNull-in-0:
          content-type: application/json

2. 泛型与POJO自动解析问题 #

泛型本身不是本次配置错误的原因，但会影响消息反序列化。Spring Cloud Stream支持自动将消息体解析为POJO，需满足以下条件：

• 消息的Content-Type头设置正确（如application/json）
• POJO类拥有无参构造函数，字段结构与消息体匹配
• 若使用泛型类型（如MyClass<T>），可通过以下方式明确类型：
  • 在Consumer方法中使用@Payload TypeReference<MyClass<SpecificType>>获取类型
  • 在配置中指定spring.cloud.stream.bindings.input.content-type为application/json，并确保Jackson等序列化库能正确处理泛型

3. 自定义路由器的配置要点 #

• 确保自定义路由器实现MessageRoutingCallback接口，并通过@Bean注册到Spring容器
• 无需为每个Consumer单独配置输入绑定，路由器会根据路由结果动态转发消息
• 若需对不同Consumer做差异化配置，直接针对{functionName}-in-0节点配置属性即可

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