Spring Cloud Stream：同一Kinesis流多事件类型多消费者配置问询

这个方案是完全可行的，你的思路没问题，但现有配置缺少关键的消息路由/筛选逻辑，导致无法正确分发消息到对应Consumer。以下是具体的调整方案：

1. 修正Consumer Bean的泛型错误 #

你提供的代码中第二个Consumer的泛型写错了，应该对应CustomerCreated类型，否则无法正确接收该类型事件：

@Bean
Consumer<OrderCreated> processOrderCreatedEvent(SomeReceiver receiver) {
    return receiver::receive;
}

@Bean
Consumer<CustomerCreated> processCustomerCreatedEvent(SomeOtherReceiver receiver) {
    return receiver::receive;
}

2. 选择合适的消息分发方式 #

针对同一流的多类型事件，有两种主流配置方式：

方式一：使用函数路由（推荐，更简洁） #

启用Spring Cloud Stream的函数路由功能，通过消息头动态匹配对应的Consumer：

spring:
  cloud:
    function:
      definition: processOrderCreatedEvent;processCustomerCreatedEvent
    stream:
      function:
        routing:
          enabled: true
      bindings:
        functionRouter-in-0:
          destination: events-stream-name
      kinesis:
        bindings:
          functionRouter-in-0:
            consumer:
              header-mode: headers
              # 如果使用自定义type头而非默认的contentType，需配置路由表达式
              # routing-expression: headers['type']

发送消息时，需在消息头中携带spring.cloud.function.definition字段，值为对应的Consumer Bean名称（如processOrderCreatedEvent）；或者通过自定义路由表达式匹配消息中的类型标识（比如你原本的type字段）。

方式二：使用消息筛选器（每个Consumer绑定同一流，仅处理指定类型） #

为每个Consumer绑定添加筛选表达式，只接收匹配类型的消息：

spring:
  cloud:
    function:
      definition: processOrderCreatedEvent;processCustomerCreatedEvent
    stream:
      bindings:
        processOrderCreatedEvent-in-0:
          destination: events-stream-name
          consumer:
            header-mode: headers
            filter-expression: headers['type'] == 'OrderCreated'
        processCustomerCreatedEvent-in-0:
          destination: events-stream-name
          consumer:
            header-mode: headers
            filter-expression: headers['type'] == 'CustomerCreated'
      kinesis:
        bindings:
          processOrderCreatedEvent-in-0:
            consumer:
              # 可添加Kinesis消费者专属配置，如分片数、重试策略等
          processCustomerCreatedEvent-in-0:
            consumer:
              # 对应Kinesis配置

3. 确保消息序列化/反序列化正确 #

必须配置合适的消息转换器，保证能根据消息头的类型信息正确反序列化为对应的事件对象。以Jackson为例：

spring:
  cloud:
    stream:
      default:
        producer:
          use-native-encoding: false
        consumer:
          use-native-decoding: false
      kinesis:
        binder:
          serialization:
            type: json
            jackson:
              # 若需自定义ObjectMapper，可指定全类名
              # objectMapper: com.your.package.CustomEventObjectMapper

只要完成以上配置，就能实现同一Kinesis流中的不同事件类型由独立的Consumer Bean处理的需求。

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