Spring Cloud Stream Kafka 消费者特定绑定配置详解

spring:
  application:
    name: test-app
  cloud:
    stream:
      kafka:
        binder:
          # 全局默认消费者属性，如果未在绑定级别覆盖，则使用此值
          consumerProperties:
            value:
              deserializer: org.apache.kafka.common.serialization.StringDeserializer # 默认设置为 StringDeserializer
          brokers: localhost:9092
          autoCreateTopics: true
          replicationFactor: 1
        # 注意：这是 Kafka Binder 特定的绑定配置
        bindings:
          listenCloudEvent-in-0:
            consumer:
              configuration:
                value:
                  deserializer: io.cloudevents.kafka.CloudEventDeserializer # 针对 com.test.cloudevent 主题的 CloudEvent 反序列化器
      # 这是 Spring Cloud Stream 通用绑定配置
      bindings:
        listenCloudEvent-in-0:
          destination: com.test.cloudevent
          group: test-app-group
          # 注意：此处不应放置 Kafka 特有的 consumer.configuration 配置
        listenString-in-0:
          destination: com.test.string
          group:  test-app-group
          # 此绑定将使用全局默认的 StringDeserializer，因为没有 Kafka 特定的覆盖
    function:
      definition: listenCloudEvent;listenString

import io.cloudevents.CloudEvent;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.Message;
import reactor.core.publisher.Flux;
import java.util.function.Consumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component; // 假设这是一个组件类

@Component
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 Listener] Message with ID '{}' received.", message.getPayload().getId()))
                .subscribe();
    }

    // 假设还有另一个用于处理字符串消息的函数
    @Bean
    public Consumer<Flux<Message<String>>> listenString() {
        return inboundMessage -> inboundMessage
                .onErrorStop()
                .doOnNext(message -> log.info("[String Listener] Message received: {}", message.getPayload()))
                .subscribe();
    }
}