Spring Cloud Stream Kafka消费者多反序列化器配置指南

spring:
  cloud:
    stream:
      bindings:
        listenCloudEvent-in-0:
          destination: com.test.cloudevent
          group: test-app-group
          consumer:
            configuration:
              value:
                deserializer: io.cloudevents.kafka.CloudEventDeserializer # 错误示例

org.springframework.messaging.converter.MessageConversionException: Could not read JSON: Cannot construct instance of `io.cloudevents.CloudEvent` (no Creators, like default constructor, exist): abstract types either need to be mapped to concrete types, have custom deserializer, or contain additional type information

spring:
  application:
    name: test-app
  cloud:
    stream:
      kafka:
        binder:
          consumerProperties: # 全局Kafka消费者属性，作为默认值
            value:
              deserializer: org.apache.kafka.common.serialization.StringDeserializer
          brokers: localhost:9092
          autoCreateTopics: true
          replicationFactor: 1
        bindings: # Kafka绑定特定属性，会覆盖binder级别或通用stream级别配置
          listenCloudEvent-in-0:
            consumer:
              configuration: # 注意这里是kafka.bindings.<channelName>.consumer.configuration
                value:
                  deserializer: io.cloudevents.kafka.CloudEventDeserializer
          listenString-in-0:
            consumer:
              configuration:
                value:
                  deserializer: org.apache.kafka.common.serialization.StringDeserializer # 明确指定，或依赖binder级别默认值
      bindings: # 通用Stream绑定属性
        listenCloudEvent-in-0:
          destination: com.test.cloudevent
          group: test-app-group
        listenString-in-0:
          destination: com.test.string
          group:  test-app-group
    function:
      definition: listenCloudEvent;listenString

import io.cloudevents.CloudEvent;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.Message;
import reactor.core.publisher.Flux;
import java.util.function.Consumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@Configuration
public class KafkaListeners {

    private static final Logger log = LoggerFactory.getLogger(KafkaListeners.class);

    @Bean
    public Consumer<Flux<Message<CloudEvent>>> listenCloudEvent() {
        return inboundMessage -> inboundMessage
                .onErrorStop() // 遇到错误时停止处理当前流
                .doOnNext(message -> log.info("[{}] CloudEvent message received. ID: {}",
                                               Thread.currentThread().getName(),
                                               message.getPayload().getId()))
                .subscribe(); // 订阅Flux以开始消费
    }

    // 假设还有另一个监听器处理字符串消息
    @Bean
    public Consumer<Flux<Message<String>>> listenString() {
        return inboundMessage -> inboundMessage
                .onErrorStop()
                .doOnNext(message -> log.info("[{}] String message received. Payload: {}",
                                               Thread.currentThread().getName(),
                                               message.getPayload()))
                .subscribe();
    }
}