Kafka 生产者

一、Kafka 生产者发送原理

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

1.1、发送流程

Kafka 生产者

1.2、生产者重要参数列表

参数名称 描述
bootstrap.servers 生产者连接集群所需的 broker 地址清单。例如 localhost:9092,可以配置一个或多个,多个地址用逗号隔开
key.serializer 和 value.serializer 指定发送消息的 key 和 value 的序列化类型,一定要写全类名
buffer.memory RecordAccumulator 缓冲区总大小,默认 32m
batch.size 缓冲区一批数据最大值,默认 16k。适当增加可以提高吞吐量,太大会导致数据传输延迟增加
linger.ms 如果数据迟迟未达到 batch.size,sneder 等待 linger.time 之后就会发送数据。单位 ms,默认 0ms,没有延迟,建议设置 5~100ms之间
acks 0:生产者发送过来的数据,不需等待数据落盘应答。1:生产者发送过来的数据,Leader 收到数据后应答。-1(all):生产者发送过来的数据,Leader 和 isr 队列里所有节点收齐数据后应答。默认 -1,-1 和 all 等价
max.in.flight.requests.per.connection 允许最多没有返回 ack 的次数,默认为 5,开启幂等性要保证该值是 1~5 数字
retries 当消息发送出现错误时,系统会重发消息。retries 表示重试次数。默认为 int 最大值,2147483647。如果设置了重试还想保证消息有序性,需设置 MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION=1 否则在重试此消息失败时,其它消息可能发送成功
retry.backoff.ms 两次重试之间的间隔,默认是 100ms
enable.idempotence 是否开启幂等性,默认为 true,开启
compression.type 生产者发送的所有数据的压缩方式。默认是 none,即不压缩。支持压缩类型:none、gzip、snappy、lz4 和 zstd

二、异步发送 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是单调自增的。

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

Kafka 生产者

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 的数据都是有序的。

Kafka 生产者

九、生产经验 – 数据有序

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

具体之后章节说明

本作品采用《CC 协议》,转载必须注明作者和本文链接
讨论数量: 0
(= ̄ω ̄=)··· 暂无内容!

讨论应以学习和精进为目的。请勿发布不友善或者负能量的内容,与人为善,比聪明更重要!
未填写
文章
247
粉丝
12
喜欢
211
收藏
57
排名:750
访问:8610
私信
所有博文
社区赞助商