Spring Boot Kafka迁移：如何替代已废弃的@StreamListener？

方案1：Spring Cloud Stream 简化函数式写法（无需Message包装） #

如果继续用Spring Cloud Stream，完全不用套Message<Order>，直接用Consumer<Order>就能实现和原代码一样的简洁度，同时符合官方推荐的函数式模型：

@Bean
public Consumer<Order> consumeOrder() {
    return order -> {
        // 直接处理Order对象，业务逻辑写这里
        // 比如：orderService.process(order);
    };
}

只需要在配置文件里把这个函数绑定到my_topic主题就行（以application.yml为例）：

spring:
  cloud:
    stream:
      bindings:
        consumeOrder-in-0:
          destination: my_topic
          content-type: application/json

这里的consumeOrder-in-0命名规则是函数名 + -in- + 输入索引（因为Consumer是单输入，所以索引为0）。

方案2：用Spring Kafka原生@KafkaListener复刻原写法 #

如果不需要Spring Cloud Stream的额外抽象，直接用Spring Kafka的@KafkaListener能完全复刻你原来@StreamListener的简洁风格，代码结构几乎没变：

@Component
public class OrderListener {

    @KafkaListener(topics = "my_topic")
    public void consumeOrder(Order order) {
        // 业务逻辑和原来一模一样，直接写就行
        // 比如：handleOrder(order);
    }
}

只需要确保Spring Kafka的基础配置正确（application.yml）：

spring:
  kafka:
    consumer:
      bootstrap-servers: localhost:9092 # 替换成你的Kafka地址
      group-id: order-consumer-group # 自定义消费组ID
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.trusted.packages: "com.yourpackage" # 替换成Order类所在的包，或者用*允许所有

方案3：Spring Cloud Stream 声明式通道绑定（适合自定义场景） #

如果需要更明确的通道管理，还可以结合@Input定义通道，再绑定消费逻辑，写法也比较清晰：

首先定义输入通道接口：

public interface OrderInputChannel {
    String ORDER_INPUT = "order-input";

    @Input(ORDER_INPUT)
    SubscribableChannel orderInput();
}

配置文件绑定通道到主题：

spring:
  cloud:
    stream:
      bindings:
        order-input:
          destination: my_topic

最后编写消费逻辑：

@Component
public class OrderListener {

    @Bean
    public Consumer<Order> consumeOrder() {
        return order -> {
            // 业务逻辑
        };
    }

    @Autowired
    public void bindConsumer(OrderInputChannel channel, Consumer<Order> consumer) {
        channel.orderInput().subscribe(msg -> consumer.accept((Order) msg.getPayload()));
    }
}

不过这种写法相对前两种稍显繁琐，更适合需要自定义通道行为的场景。

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