作为Kafka消费者异常的Spring boot微服务

@KafkaListener(topics = "myTopic", groupId = "myGroup")
public void listen(String message) {
    // 并行处理消息
    CompletableFuture.runAsync(() -> processMessage(message));
}

props.put("enable.auto.commit", "false");
props.put("auto.offset.reset", "earliest");

@Transactional
public void processMessage(String message) {
    // 幂等性处理逻辑
}

@SpringBootApplication
public class KafkaConsumerApplication {

    public static void main(String[] args) {
        SpringApplication.run(KafkaConsumerApplication.class, args);
    }

    @KafkaListener(topics = "testTopic", groupId = "group-id")
    public void listen(ConsumerRecord<String, String> record) {
        System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
    }
}

spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=my-group
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer