一、Kafka 生产者发送原理

在消息发送的过程中，涉及到了两个线程 一一 main 线程和 Sender 线程。在 main 线程中创建了一个双端队列 RecordAccumulator。main 线程将消息发送给 RecordAccumulator，Sender 线程不断从RecordAccumulator 中拉取消息发送到 Kafka Broker。

1.1、发送流程

1.2、生产者重要参数列表

二、异步发送 API

2.1、异步普通发送

2.1.1、需求

创建 Kafka 生产者，采用异步方式发送到 Kafka Broker

2.1.2、创建 kafka 工程，引入依赖

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.2.3</version>
</dependency>

2.1.3、编写不带回调函数的 API 代码

public class CustomProducer {
    public static void main(String[] args) {
        // 创建 Kafka 生产者配置对象
        Properties properties = new Properties();

        // 给 Kafka 配置对象添加配置信息
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        // 指定对应 key，value 序列化，必 须是 key.serializer,value.serializer
        // properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 创建 Kafka 生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

        // 调用 send 方法，发送消息
        for (int i = 0; i < 5; i++) {
            kafkaProducer.send(new ProducerRecord<>("first", "test" + i));
        }

        // 关闭资源
        kafkaProducer.close();
    }
}

控制台中查看是否接收到消息

$ kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic first
test0
test1
test2
test3
test4

2.2、带回调函数的异步发送

回调函数会在 producer 收到 ack 时调用，为异步调用，该方法有两个参数，分别是元数据信息(RecordMetadata) 和异常信息 (Exception)， 如果 Exception 为 null，说明消息发送成功，如果Exception 不为 null，说明消息发送失败。

注意：发送消息失败会自动重试，不需要我们在回调函数中手动重试，这里的回调是在消息发送到队列中，队列返回的回调信息

public class CustomProducerCallback {
    public static void main(String[] args) throws InterruptedException {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
        for (int i = 0; i < 5; i++) {
            kafkaProducer.send(new ProducerRecord<>("first", "test" + i), new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    // 该方法在 Producer 收到 ack 时调用，为异步调用
                    if (e == null) {
                        // 没有异常，输出信息到控制台
                        System.out.println("主题:" + recordMetadata.topic() + " -> 分区:" + recordMetadata.partition());
                    } else {
                        // 出现异常打印
                        e.printStackTrace();
                    }
                }
            });

            // 延迟一会看到数据发往不同分区
            Thread.sleep(2);
        }

        kafkaProducer.close();
    }
}

代码运行结果

主题:first -> 分区:2
主题:first -> 分区:2
主题:first -> 分区:2
主题:first -> 分区:0
主题:first -> 分区:0

终端接收到的消息

$ kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic first
test3
test4
test0
test1
test2

三、同步发送 API

所谓同步发送，就是在外部数据发送到队列中后，队列中的数据必须要全部发布发送完毕处理完后，再继续对外部数据进行处理。
只需要在异步发送的基础上，再调用 get() 方法即可。

public class CustomerProducerSync {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 创建 Kafka 生产者配置对象
        Properties properties = new Properties();

        // 给 Kafka 配置对象添加配置信息
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        // 指定对应 key，value 序列化(必须) key.serializer,value.serializer
        // properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 创建 Kafka 生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

        // 调用 send 方法，发送消息
        for (int i = 0; i < 5; i++) {
            // 异步发送 默认
            // kafkaProducer.send(new ProducerRecord<>("first", "test" + i));
            // 同步发送
            kafkaProducer.send(new ProducerRecord<>("first", "test" + i)).get();
        }

        // 关闭资源
        kafkaProducer.close();
    }
}

查看终端消费者结果

$ kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic first
test0
test1
test2
test3
test4

现实结果与异步没有太大的差别，只是底层原理有区别。

四、生产者分区

4.1、分区好处

1. 便于合理使用存储资源，每个Partition在一个 Broker 上存储，可以把海量的数据按照分区切割成一块一块数据存储在多台 Broker上。合理控制分区的任务，可以实现负载均衡的效果。
2. 提高并行度，生产者可以以分区为单位发送数据;消费者可以以分区为单位进行消费数据。

4.2、生产者发送消息的分区策略

4.2.1、默认的分区器 DefaultPartitioner

/**
 * The default partitioning strategy:
 * <ul>
 * <li>If a partition is specified in the record, use it
 * <li>If no partition is specified but a key is present choose a partition based on a hash of the key
 * <li>If no partition or key is present choose the sticky partition that changes when the batch is full.
 * 
 * See KIP-480 for details about sticky partitioning.
 */
 public class DefaultPartitioner implements Partitioner {}

在 PruducerRecord 类中，可以看到如下构造方法，大致分为三类

4.2.1.1、指明 partition

直接指明的值作为 partition 值，例如 partition=0，所有数据写入分区 0

public ProducerRecord(String topic, Integer partition, Long timestamp, K key, V value, Iterable<Header> headers) {}

public ProducerRecord(String topic, Integer partition, Long timestamp, K key, V value) {}

public ProducerRecord(String topic, Integer partition, K key, V value, Iterable<Header> headers) {}

public ProducerRecord(String topic, Integer partition, K key, V value) {}

4.2.1.2、没指明 partition 值但有 key

将 key 的 hash 值与 topic 的 partition 数进行取余得到 partition 值

例如：key1 的 hash 值 = 5，key2 的 hash 值 = 6，topic 的 partition 数 = 2，那么 key1 对应的 value1 写入 1 号分区，key2 对应的 vlaue2 写入 0 号分区

public ProducerRecord(String topic, K key, V value) {}

4.2.1.3、既没有 partition 值又没有 key 值

kafka 采用 Sticky Partition （黏性分区器），会随机选择一个分区，并尽可能一直使用该分区，待该分区的 batch 已满或者已完成，kafka 再随机一个分区进行使用（和上一次的分区不同）。

例如：第一次随机选择 0 号分区，等 0 号分区当前批次满了（默认为 16k）或者 linger.ms 设置的时间到了，Kafka 再随机选择一个分区进行使用（如果还是 0 会继续选择）。

public ProducerRecord(String topic, V value) {}

4.2.2、案例一

将数据发往指定 partition 情况下，例如，将所有数据发往分区 1 中。

public class CustomProducerCallbackPartitions {
    public static void main(String[] args) throws InterruptedException {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
        for (int i = 0; i < 5; i++) {
            kafkaProducer.send(new ProducerRecord<>("first",1,"", "test" + i), new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    // 该方法在 Producer 收到 ack 时调用，为异步调用
                    if (e == null) {
                        // 没有异常，输出信息到控制台
                        System.out.println("主题:" + recordMetadata.topic() + " -> 分区:" + recordMetadata.partition());
                    } else {
                        // 出现异常打印
                        e.printStackTrace();
                    }
                }
            });
        }
        kafkaProducer.close();
    }
}

开启 Kafka 消费者进行查看

$ kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic first
test0
test1
test2
test3
test4

控制台输出如下

主题:first -> 分区:1
主题:first -> 分区:1
主题:first -> 分区:1
主题:first -> 分区:1
主题:first -> 分区:1

4.2.3、案例二

没有指明 partition 值但是有 key 的情况下，将 key 的 hash 值与 topic 的 partition 数进行取余达到 parition 值

public class CustomProducerCallbackPartitions {
    public static void main(String[] args) throws InterruptedException {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
        for (int i = 0; i < 5; i++) {
            kafkaProducer.send(new ProducerRecord<>("first","a", "test" + i), new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    // 该方法在 Producer 收到 ack 时调用，为异步调用
                    if (e == null) {
                        // 没有异常，输出信息到控制台
                        System.out.println("主题:" + recordMetadata.topic() + " -> 分区:" + recordMetadata.partition());
                    } else {
                        // 出现异常打印
                        e.printStackTrace();
                    }
                }
            });
        }

        kafkaProducer.close();
    }
}

开启 Kafka 消费者进行查看

$ kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic first
test0
test1
test2
test3
test4

当 key=”a” 控制台输出如下

主题:first -> 分区:1
主题:first -> 分区:1
主题:first -> 分区:1
主题:first -> 分区:1
主题:first -> 分区:1

当 key=”b” 控制台输出如下

主题:first -> 分区:2
主题:first -> 分区:2
主题:first -> 分区:2
主题:first -> 分区:2
主题:first -> 分区:2

当 key=”f” 控制台输出如下

主题:first -> 分区:0
主题:first -> 分区:0
主题:first -> 分区:0
主题:first -> 分区:0
主题:first -> 分区:0

4.2.4、案例三

当既没有指定 partition 值也没有 key，采用 Sticky Partition，下面案例通过延时放大效果

public class CustomProducerCallback {
    public static void main(String[] args) throws InterruptedException {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
        for (int i = 0; i < 10; i++) {
            kafkaProducer.send(new ProducerRecord<>("first", "test" + i), new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    // 该方法在 Producer 收到 ack 时调用，为异步调用
                    if (e == null) {
                        // 没有异常，输出信息到控制台
                        System.out.println("主题:" + recordMetadata.topic() + " -> 分区:" + recordMetadata.partition());
                    } else {
                        // 出现异常打印
                        e.printStackTrace();
                    }
                }
            });

            // 延迟一会看到数据发往不同分区
            Thread.sleep(2);
        }

        kafkaProducer.close();
    }
}

开启 Kafka 消费者进行查看

$ kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic first
test0
test1
test2
test3
test4
test5
test6
test7
test8
test9

控制台输出如下

主题:first -> 分区:0
主题:first -> 分区:0
主题:first -> 分区:0
主题:first -> 分区:0
主题:first -> 分区:0
主题:first -> 分区:2
主题:first -> 分区:2
主题:first -> 分区:2
主题:first -> 分区:2
主题:first -> 分区:0

4.3、自定义分区器

可根据具体需求，自己重新实现分区器

4.3.1、需求

例如我们实现一个分区器实现，发送过来的数据中如果包含 hudu，就发往 0 号分区，不包含 hudu，就发往 1 号分区。

4.3.2、实现步骤

4.3.2.1、实现 Partitioner 接口

4.3.2.2、重写 partition() 方法

public class MyPartitioner implements Partitioner {
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        // 获取消息
        String msgValue = value.toString();

        // 创建 partition
        int partition;

        // 判断消息是否包含 hudu
        if (msgValue.contains("hudu")) {
            partition = 0;
        } else {
            partition = 1;
        }

        // 返回分区号
        return partition;
    }

    // 关闭资源
    @Override
    public void close() {

    }

    // 配置方法
    @Override
    public void configure(Map<String, ?> configs) {

    }
}

4.3.2.3、使用分区器的方法

在生产者的配置中添加分区器的参数

public class CustomProducerCallbackPartitioner {
    public static void main(String[] args) throws InterruptedException {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // 关联自定义分区器
        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "com.hudu.producer.MyPartitioner");

        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
        for (int i = 0; i < 10; i++) {
            String msg = i / 5 == 0 ? "test" + i : "hudu" + i;
            kafkaProducer.send(new ProducerRecord<>("first", msg), new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    // 该方法在 Producer 收到 ack 时调用，为异步调用
                    if (e == null) {
                        // 没有异常，输出信息到控制台
                        System.out.println("主题:" + recordMetadata.topic() + " -> 分区:" + recordMetadata.partition());
                    } else {
                        // 出现异常打印
                        e.printStackTrace();
                    }
                }
            });
        }

        kafkaProducer.close();
    }
}

2.3.2.4、测试

开启 kafka 消费者

$ kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic first
hudu5
hudu6
hudu7
hudu8
hudu9
test0
test1
test2
test3
test4

控制台输出

主题:first -> 分区:1
主题:first -> 分区:1
主题:first -> 分区:1
主题:first -> 分区:1
主题:first -> 分区:1
主题:first -> 分区:0
主题:first -> 分区:0
主题:first -> 分区:0
主题:first -> 分区:0
主题:first -> 分区:0

五、生产经验 – 生产者提高吞吐量

• batch.size:批次大小，默认 16k
• linger.ms:等待时间，修改为 5～100ms
• comperssion.type:压缩 snappy
• RecordAccumulator:缓冲区大小，修改为 64m

public class CustomProducerParameters {
    public static void main(String[] args) {
        // 创建 Kafka 生产者配置对象
        Properties properties = new Properties();

        // 给 Kafka 配置对象添加配置信息:bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        // 指定对应 key，value 序列化(必须) key.serializer,value.serializer
        // properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // batch.size：批次大小，默认 16k
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);

        // linger.ms：等待时间，默认为 0
        properties.put(ProducerConfig.LINGER_MS_CONFIG, 1);

        // RecordAccumulator：缓冲区大小，默认为 23m，buffer.memory
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);

        // compression.type：压缩，默认 none，可配置 gzip、snappy、lz4 和 zstd
        properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");

        // 创建 Kafka 生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

        // 调用 send 方法，发送消息
        for (int i = 0; i < 5; i++) {
            kafkaProducer.send(new ProducerRecord<>("first", "test" + i));
        }
        // 关闭资源
        kafkaProducer.close();
    }
}

开启 kafka 消费者

$ kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic first
test0
test1
test2
test3
test4

六、生产经验 – 数据可靠性

6.1、Ack 应答机制

6.1.1、0

生产者发送过来的数据，不需要等待数据落盘应答

数据可靠性分析：如果 Leader 宕机了，会导致丢数

6.1.2、1

生产者发送过来的数据，Leader 收到数据后应答

数据可靠性分析：如果 Leader 应答完成后，还没开始同步副本，Leader 挂了。重新选举 Leader 后，信的 Leader 不会后到之前未同步的信息，因为生产者认为已经发送成功了。会导致丢数

6.1.3、-1(all)

生产者发送过来的数据，Leader 和 ISR 队列里面的所有节点收集齐数据后应答。

问题：如果Leader 收到数据，所有 Follower 都开始同步数据，但是有一个 Follower 因为某种故障，迟迟无法与 leader 进行同步，那么这个问题怎么解决。

Leader 维护了一个动态的 in-sync replica set （ISR），意为和 Leader 保持同步的 Follower + Leader 集合 (leader: 0, isr: 0,1,2)。

如果 Folower 长时间未向 Leader 发送通信请求或同步数据，则该 Follower 将被踢出 ISR。该时间阈值由 replica.lag.time.max.ms 参数设定，默认 30s。例如 2 超时，(leader: 0, isr: 0,1)。

这样就不用等长期联系不上或者已经故障的节点。

数据可靠性分析：如果副本数设置为 1 个，或者 ISR 里main应答的最小副本数量 (min.insync.replicas 默认为 1) 设置为 1，和 ack=1 效果是一样的。仍然有丢数的风险 (leader: 0, isr:0)。

完全可靠条件 = ACK 级别设置为 -1 + 分区副本大于等于 2 + ISR 里应答的最小副本数量大于等于 2。

6.2、可靠性总结

acks=0，生产者发送过来数据就不管了，可靠性差，效率高;

acks=1，生产者发送过来数据Leader应答，可靠性中等，效率中等;

acks= 1，生产者发送过来数据Leader和ISR队列里面所有Follwer应答，可靠性高，效率低;

在生产环境中，acks=0很少使用; acks=1，一般用于传输普通日志，允许丢个别数据; acks=-1, 一般用于传输和钱相关的数据，对可靠性要求比较高的场景。

6.3、代码配置

分区数以及副本数在创建 topic 时指定

$ kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --topic first --create --partitions 3 --replication-factor 3

public class CustomProducerAcks {
    public static void main(String[] args) {
        // 创建 Kafka 生产者配置对象
        Properties properties = new Properties();

        // 给 Kafka 配置对象添加配置信息
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        // 指定对应 key，value 序列化，必 须是 key.serializer,value.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 设置 acks
        properties.put(ProducerConfig.ACKS_CONFIG,"all");

        // 重试次数 retries，默认是 int 最大值，2147483647
        properties.put(ProducerConfig.RETRIES_CONFIG,3);

        // 创建 Kafka 生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

        // 调用 send 方法，发送消息
        for (int i = 0; i < 5; i++) {
            kafkaProducer.send(new ProducerRecord<>("first", "test" + i));
        }

        // 关闭资源
        kafkaProducer.close();
    }
}

七、生产经验 – 去重

当 acks 为 -1 时，生产者发送过来的数据，Leader 和 ISR 队列里面的所有节点收齐数据后应答。

如果 Producer 发送数据，在 Leader 以及 ISR 队列里面数据还未全部落盘前，Leader 宕机了，导致没有返回 ack，此时重新选举 Leader，并不会返回 ack，Producer 会再次发送数据，这就导致了数据的重复。

7.1、数据传递语义

至少一次 (At Least Once) = ACK级别设置为-1 + 分区副本大于等于2 + ISR里应答的最小副本数量大于等于2

最多一次 (At Most Once) = ACK级别设置为 0

总结:

At Least Once可以保证数据不丢失，但是不能保证数据不重复；

At Most Once可以保证数据不重复，但是不能保证数据不丢失。

精确一次 (Exactly Once)：对于一些非常重要的信息，比如和钱相关的数据，要求数据既不能重复也不丢失。Kafka 0.11版本以后，引入了一项重大特性：幂等性和事务。

7.2、幂等性

7.2.1、幂等性原理

幂等性就是指Producer不论向Broker发送多少次重复数据，Broker端都只 会持久化一条，保证了不重复。

精确一次(Exactly Once) = 幂等性 + 至少一次 ( ack = -1 + 分区副本数 >= 2 + ISR最小副本数量 >= 2)。

重复数据的判断标准:具有<PID, Partition, SeqNumber>相同主键的消息提交时，Broker只 会持久化一条。其中 PID 是 Kafka 每次重启都会分配-一个新的; Partition 表示分区号; Sequence Number是单调自增的。

所以幂等性只能保证的是在单分区单会话内不重复。

7.2.2、使用幂等性

开启参数 enable.idempotnece 默认为 true，false 关闭

7.3、生产者事务

7.3.1、事务原理

Producer 在使用事务功能前，必须先自定，义一个唯一的transactional.id。 有了transactional.id， 即使客户端挂掉了，它重启后也能继续处理未完成的事务。注意开启事务，必须开始幂等性

7.3.2、Kafka 事务一共有如下 5 个 API

// 初始化事务
public void initTransactions();

// 开启事务
public void beginTransaction() throws ProducerFencedException;

// 在事务内提交已经消费的偏移量(主要用于消费者)
public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,String consumerGroupId) throws ProducerFencedException;

// 提交事务
public void commitTransaction() throws ProducerFencedException;

// 放弃事务(类似于事务回滚的操作)
public void abortTransaction() throws ProducerFencedException;

7.3.3、单个 Producer，使用事务保证消息的仅一次发送

public class CustomProducerTransactions {
    public static void main(String[] args) {
        // 创建 Kafka 生产者配置对象
        Properties properties = new Properties();

        // 给 Kafka 配置对象添加配置信息
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        // 指定对应 key，value 序列化(必须) key.serializer,value.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 指定事务id
        properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transactional_id_01");

        // 创建 Kafka 生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

        kafkaProducer.initTransactions();

        kafkaProducer.beginTransaction();

        try {
            // 调用 send 方法，发送消息
            for (int i = 0; i < 5; i++) {
                kafkaProducer.send(new ProducerRecord<>("first", "test" + i));
            }
            // 提交事务
            kafkaProducer.commitTransaction();
        } catch (Exception e) {
            // 终止事务
            kafkaProducer.abortTransaction();
        } finally {
            // 关闭资源
            kafkaProducer.close();
        }
    }
}

八、生产经验 – 数据乱序

1. kafka在1.x版本之前保证数据单分区有序，条件如下：max.in.flight.requests.per.connection=1 (不需要考虑是否开启幂等性)。
2. kafka在1.x及以后版本保证数据单分区有序，条件如下：
   1. 未开启幂等性：max.in.flight.requests.per.connection需要设置为1。
   2. 开启幂等性：max.in.flight.requests.per.connection需要设置小于等于5。原因说明:因为在kafka 1.x 以后，启用幂等后，kafka服务端会缓存 producer 发来的最近 5 个 request 的元数据，故无论如何，都可以保证最近 5 个 request 的数据都是有序的。

九、生产经验 – 数据有序

单分区内，有序(有条件)，多分区，分区与分区无序

具体之后章节说明

本作品采用《CC 协议》，转载必须注明作者和本文链接