一碰就头疼的 Kafka 消息重复问题，立马解决！

Properties props = new Properties();
props.put("enable.idempotence", ture); // 1. 设置幂等
props.put("acks", "all"); // 2. 当 enable.idempotence 为 true，这里默认为 all
props.put("max.in.flight.requests.per.connection", 5); // 3. 注意

Properties props = new Properties();
props.put("enable.idempotence", ture); // 1. 设置幂等
props.put("acks", "all"); // 2. 当 enable.idempotence 为 true，这里默认为 all
props.put("max.in.flight.requests.per.connection", 5); // 3. 最大等待数
props.put("transactional.id", "my-transactional-id"); // 4. 设定事务 id
Producer<String, String> producer = new KafkaProducer<String, String>(props);
// 初始化事务
producer.initTransactions();
try{
    // 开始事务
    producer.beginTransaction();
    // 发送数据
    producer.send(new ProducerRecord<String, String>("Topic", "Key", "Value"));
    // 数据发送及 Offset 发送均成功的情况下，提交事务
    producer.commitTransaction();
} catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
    // 数据发送或者 Offset 发送出现异常时，终止事务
    producer.abortTransaction();
} finally {
    // 关闭 Producer 和 Consumer
    producer.close();
    consumer.close();
}

$ docker run -d --name zookeeper -p 2181:2181 zookeeper
a86dff3689b68f6af7eb3da5a21c2dba06e9623f3c961154a8bbbe3e9991dea4

> 基于 Spring Cloud Alibaba + Gateway + Nacos + RocketMQ + Vue & Element 实现的后台管理系统 + 用户小程序，支持 RBAC 动态权限、多租户、数据权限、工作流、三方登录、支付、短信、商城等功能
>
> * 项目地址：<https://github.com/YunaiV/yudao-cloud>
> * 视频教程：<https://doc.iocoder.cn/video/>
# 举个栗子：topic 需要自己去修改
$ cd ./kafka-2.7.1-src/bin
$ ./kafka-console-producer.sh --broker-list localhost:9092 --topic test_topic

public class KafkaProducerApplication {
    private final Producer<String, String> producer;
    final String outTopic;
    public KafkaProducerApplication(final Producer<String, String> producer,
                                    final String topic) {
        this.producer = producer;
        outTopic = topic;
    }
    public void produce(final String message) {
        final String[] parts = message.split("-");
        final String key, value;
        if (parts.length > 1) {
            key = parts[0];
            value = parts[1];
        } else {
            key = null;
            value = parts[0];
        }
        final ProducerRecord<String, String> producerRecord
            = new ProducerRecord<>(outTopic, key, value);
        producer.send(producerRecord,
                (recordMetadata, e) -> {
                    if(e != null) {
                        e.printStackTrace();
                    } else {
                        System.out.println("key/value " + key + "/" + value + "\twritten to topic[partition] " + recordMetadata.topic() + "[" + recordMetadata.partition() + "] at offset " + recordMetadata.offset());
                    }
                }
        );
    }
    public void shutdown() {
        producer.close();
    }
    public static void main(String[] args) {
        final Properties props = new Properties();
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.CLIENT_ID_CONFIG, "myApp");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        final String topic = "myTopic";
        final Producer<String, String> producer = new KafkaProducer<>(props);
        final KafkaProducerApplication producerApp = new KafkaProducerApplication(producer, topic);
        String filePath = "/home/donald/Documents/Code/Source/kafka-2.7.1-src/examples/src/main/java/kafka/examples/input.txt";
        try {
            List<String> linesToProduce = Files.readAllLines(Paths.get(filePath));
            linesToProduce.stream().filter(l -> !l.trim().isEmpty())
                    .forEach(producerApp::produce);
            System.out.println("Offsets and timestamps committed in batch from " + filePath);
        } catch (IOException e) {
            System.err.printf("Error reading file %s due to %s %n", filePath, e);
        } finally {
            producerApp.shutdown();
        }
    }
}

$ ./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic myTopic

Exception in thread "main" org.apache.kafka.common.config.ConfigException: Must set acks to all in order to use the idempotent producer.
Otherwise we cannot guarantee idempotence.