08. Java Spring Kafka 实战

spring:
  kafka:
    bootstrap-servers: kafka1:9092,kafka2:9092
    producer:
      acks: all
      retries: 3
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    consumer:
      group-id: order-service
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

@Service
public class OrderEventPublisher {
    @Autowired private KafkaTemplate<String, String> kafkaTemplate;

    public void publishOrderCreated(Order order) {
        kafkaTemplate.send("orders.created", order.getUserId(),
            JsonUtils.toJson(order));
    }
}

@Component
public class OrderEventConsumer {

    @KafkaListener(topics = "orders.created",
                   groupId = "order-service",
                   concurrency = "3")  // 3 个并发消费者
    public void onOrderCreated(ConsumerRecord<String, String> record,
                               Acknowledgment ack) {
        Order order = JsonUtils.fromJson(record.value(), Order.class);
        processOrder(order);
        ack.acknowledge();  // 手动提交
    }

    @KafkaListener(topics = "orders.cancelled")
    public void onOrderCancelled(String message) {
        // 处理取消
    }
}

@Configuration
public class KafkaConfig {
    @Bean
    public DefaultErrorHandler errorHandler() {
        BackOff fixedBackOff = new FixedBackOff(2000, 3);  // 重试3次，间隔2s
        DefaultErrorHandler handler = new DefaultErrorHandler(fixedBackOff);
        handler.addNotRetryableExceptions(IllegalArgumentException.class);
        return handler;
    }
}