spring:
  kafka:
    bootstrap-servers: localhost:9092
    
    # 生产者配置
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      acks: all
      retries: 3
    
    # 消费者配置
    consumer:
      group-id: my-group
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      auto-offset-reset: earliest
      properties:
        spring.json.trusted.packages: '*'
    
    # 监听器配置
    listener:
      # Spring Boot 4 - 使用虚拟线程
      concurrency: 10
      type: batch  # 批量消费

@Service
public class OrderEventProducer {
    private final KafkaTemplate<String, OrderEvent> kafkaTemplate;
    
    public OrderEventProducer(KafkaTemplate<String, OrderEvent> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }
    
    @Observed(name = "kafka.send")
    public CompletableFuture<SendResult<String, OrderEvent>> sendOrderEvent(OrderEvent event) {
        return kafkaTemplate.send("orders", event.orderId(), event);
    }
}

record OrderEvent(String orderId, String userId, BigDecimal amount, Instant timestamp) {}

@Service
public class OrderEventConsumer {
    private static final Logger log = LoggerFactory.getLogger(OrderEventConsumer.class);
    
    /**
     * Spring Boot 4 - 自动使用虚拟线程
     */
    @KafkaListener(topics = "orders", groupId = "order-processor")
    @Observed(name = "kafka.consume")
    public void consume(OrderEvent event) {
        log.info("Consumed order event: {}, thread: {}", 
            event.orderId(), Thread.currentThread());
        
        // 处理订单事件
        processOrder(event);
    }
    
    /**
     * 批量消费
     */
    @KafkaListener(topics = "orders-batch", groupId = "batch-processor")
    public void consumeBatch(List<OrderEvent> events) {
        log.info("Consumed {} events in batch", events.size());
        events.parallelStream().forEach(this::processOrder);
    }
    
    private void processOrder(OrderEvent event) {
        // 业务逻辑
    }
}