幂等生产者和消费者

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.boot.CommandLineRunner;
import org.springframework.context.annotation.Bean;
import java.util.UUID;

@SpringBootApplication
public class IdempotentProducerApp {

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

  @Bean
  public CommandLineRunner runner(
    KafkaTemplate<String, String> kafkaTemplate) {
    return args -> {
      String messageKey = UUID.randomUUID().toString();
      String messageValue = "Hello from Idempotent Producer!";
      System.out.println("Sending message with key: " 
                         + messageKey);
      kafkaTemplate.send("my-idempotent-topic", 
                         messageKey, messageValue)
        .addCallback(
          result -> System.out.println(
            "Message sent successfully!"),
          ex -> System.err.println(
            "Failed to send: " + ex.getMessage())
        );
    };
  }
}
// Add to application.properties:
// spring.kafka.producer.bootstrap-servers=localhost:9092
// spring.kafka.producer.key-serializer=
//   org.apache.kafka.common.serialization.StringSerializer
// spring.kafka.producer.value-serializer=
//   org.apache.kafka.common.serialization.StringSerializer
// spring.kafka.producer.properties.enable.idempotence=true

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import java.util.HashSet;
import java.util.Set;

@SpringBootApplication
public class IdempotentConsumerApp {

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

  @Component
  public static class MyKafkaListener {
    // In a real app, this would be a persistent store (DB, Redis)
    private final Set<String> processedRecordIds = new HashSet<>();

    @KafkaListener(topics = "my-idempotent-topic", 
                   groupId = "idempotent-group")
    public void listen(ConsumerRecord<String, String> record) {
      // Unique ID for the record (topic-partition-offset)
      String recordId = record.topic() + "-" + record.partition()
                        + "-" + record.offset();

      if (processedRecordIds.contains(recordId)) {
        System.out.println("Duplicate record received (ID: " 
                           + recordId + "). Skipping processing.");
        return;
      }

      // Simulate processing the message
      System.out.println("Processing record ID: " + recordId 
                         + ", Key: " + record.key() 
                         + ", Value: " + record.value());
      // Add to processed set AFTER successful processing
      processedRecordIds.add(recordId);

      // In a real scenario, processing might involve DB updates
      // and 'add' would happen as part of a transaction.
    }
  }
}
// Add to application.properties:
// spring.kafka.consumer.bootstrap-servers=localhost:9092
// spring.kafka.consumer.key-deserializer=
//   org.apache.kafka.common.serialization.StringDeserializer
// spring.kafka.consumer.value-deserializer=
//   org.apache.kafka.common.serialization.StringDeserializer
// spring.kafka.consumer.group-id=idempotent-group
// spring.kafka.consumer.auto-offset-reset=earliest