《Kafka笔记》3、Kafka高级API

• 2020 年 10 月 20 日
• 筆記
• kafka

1 Kafka高级API特性

1.1 Offset的自动控制

1.1.1 消费者offset初始策略

一般来说每个消费者消费之后,都会把自己消费到分区的位置（也就是offset提交给Kafka集群），但是对于没有消费过该分区的消费者，他之前并未提交给集群自身偏移量的信息。

Kafka消费者默认对于未订阅的topic的offset的时候，也就是系统并没有存储该消费者的消费分区的记录信息（offset），默认Kafka消费者的默认首次消费策略：latest。

配置项为：auto.offset.reset=latest

可以在官方文档，找到对于各个配置项的解释，例如 //kafka.apache.org/20/documentation.html#brokerconfigs 可以找到auto.offset.reset配置项。

• earliest – 自动将偏移量重置为最早的偏移量

• latest – 自动将偏移量重置为最新的偏移量

• none – 如果未找到消费者组的先前偏移量，则向消费者抛出异常

消费者的配置中增加凸显默认配置，latest可以换成earliest：

// 默认配置，如果系统中没有该消费组的偏移量，该消费者组读取最新的偏移量
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"latest");

// 配置earliest,如果集群没有该消费者组的偏移量，系统会读取该分区最早的偏移量开始消费
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");

1.1.2 消费者offset自动提交策略

Kafka消费者在消费数据的时候默认会定期的提交消费的偏移量，这样就可以保证所有的消息至少可以被消费者消费1次,用户可以通过以下两个参数配置：

enable.auto.commit = true 默认

auto.commit.interval.ms = 5000 默认

如果用户需要自己管理offset的自动提交，可以关闭offset的自动提交，手动管理offset提交的偏移量，注意用户提交的offset偏移量永远都要比本次消费的偏移量+1，因为提交的offset是kafka消费者下一次抓取数据的位置。

// 消费者自动提交开启
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,true);
// 配置offset自动提交时间间隔，10秒自动提交offset
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,10000);

自定义偏移量提交策略，先关闭偏移量自定提交配置后，每次消费完，提交偏移量信息给集群：

public class KafkaConsumerDemo_02 {
    public static void main(String[] args) {
        //1.创建Kafka链接参数
        Properties props=new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"CentOSA:9092,CentOSB:9092,CentOSC:9092");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
        props.put(ConsumerConfig.GROUP_ID_CONFIG,"group01");
        // 关闭offset自动提交
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);

        //2.创建Topic消费者
        KafkaConsumer<String,String> consumer=new KafkaConsumer<String, String>(props);
        //3.订阅topic开头的消息队列
        consumer.subscribe(Pattern.compile("^topic.*$"));

        while (true){
            ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(1));
            Iterator<ConsumerRecord<String, String>> recordIterator = consumerRecords.iterator();
            while (recordIterator.hasNext()){
                ConsumerRecord<String, String> record = recordIterator.next();
                String key = record.key();
                String value = record.value();
                long offset = record.offset();
                int partition = record.partition();
                
                // offset维护的Map
                Map<TopicPartition, OffsetAndMetadata> offsets=new HashMap<TopicPartition, OffsetAndMetadata>();

                // 自己维护offset，每次提交当前信息的offset加1
                offsets.put(new TopicPartition(record.topic(),partition),new OffsetAndMetadata(offset + 1));
                // 异步提交偏移量给集群，且回调打印
                consumer.commitAsync(offsets, new OffsetCommitCallback() {
                    @Override
                    public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
                        System.out.println("完成："+offset+"提交！");
                    }
                });
                System.out.println("key:"+key+",value:"+value+",partition:"+partition+",offset:"+offset);

            }
        }
    }
}

1.2 Acks & Retries(应答和重试)

Kafka生产者在发送完一个的消息之后，要求Leader所在的Broker在规定的时间Ack应答，如果没有在规定时间内应答，Kafka生产者会尝试n次重新发送消息（超时重传）。目的是确保我们的消息，一定要发送的队列中去

acks=1 默认

面试常问，kafka为什么存在数据的写入丢失？其中一种情况为下面的第一点

1、acks=1表示：Leader会将Record写到其本地日志中，但会在不等待所有Follower的完全确认的情况下做出响应。在这种情况下，如果Leader在确认记录后立即失败，但在Follower复制记录之前失败，则记录将丢失，常用在不重要的日志收集时。

2、acks=0表示：生产者根本不会等待服务器的任何确认。该记录将立即添加到网络套接字缓冲区中并视为已发送。在这种情况下，不能保证服务器已收到记录。这种情况是不可靠的，但是性能高

3、acks=all表示：这意味着Leader将等待全套同步副本确认记录。这保证了只要至少一个同步副本仍处于活动状态，记录就不会丢失。这是最有力的保证。这等效于acks = -1设置。用在一些比较重要的系统，不允许丢数据。

如果生产者在规定的时间内，并没有得到Kafka的Leader的Ack应答，Kafka可以开启reties机制。

request.timeout.ms = 30000 默认（30s没有收到leader的ack则重试）

retries = 2147483647 默认（重试次数为Max_Value，默认一直重试）

public class KafkaProducerDemo_01{
    public static void main(String[] args) {
        //1.创建链接参数
        Properties props=new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"CentOSA:9092,CentOSB:9092,CentOSC:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
        props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,UserDefineProducerInterceptor.class.getName());
        props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG,1);
        props.put(ProducerConfig.ACKS_CONFIG,"-1");
        props.put(ProducerConfig.RETRIES_CONFIG,10);

        //2.创建生产者
        KafkaProducer<String,String> producer=new KafkaProducer<String, String>(props);

        //3.封账消息队列
        for(Integer i=0;i< 1;i++){
            ProducerRecord<String, String> record = new ProducerRecord<>("topic01", "key" + i, "value" + i);
            producer.send(record);
        }

        producer.close();
    }
}

可以通过生产者自定义配置重复发送的次数：

// 不包括第一次发送，如果尝试发送三次，失败，则系统放弃发送
props.put(ProducerConfig.RETRIES_CONFIG, 3);

public class KafkaProducerDemo_02 {
    public static void main(String[] args) {
        //1.创建链接参数
        Properties props=new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"CentOSA:9092,CentOSB:9092,CentOSC:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
        props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,UserDefineProducerInterceptor.class.getName());
        props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG,1);
        props.put(ProducerConfig.ACKS_CONFIG,"-1");
        props.put(ProducerConfig.RETRIES_CONFIG,3);
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,true);

        //2.创建生产者
        KafkaProducer<String,String> producer=new KafkaProducer<String, String>(props);

        //3.封账消息队列
        for(Integer i=0;i< 1;i++){
            ProducerRecord<String, String> record = new ProducerRecord<>("topic01", "key" + i, "value" + i);
            producer.send(record);
        }

        producer.close();
    }
}

总结：应答和重试机制，可以尽最大可能保证我们把数据发送到Kafka集群。但也会伴随着一些问题，比如重复数据的产生。在一些订单业务场景中，比如用户下订单的记录，是绝对不能出现重复数据的。怎么保证？Kafka提供了幂等和事务机制，来解决重复数据的问题

1.3 Kafka幂等写机制

1.3.1 Kafka幂等概念

1、HTTP/1.1中对幂等性的定义是：一次和多次请求某一个资源对于资源本身应该具有同样的结果（网络超时等问题除外）。也就是说，其任意多次执行对资源本身所产生的影响均与一次执行的影响相同。

2、Kafka在0.11.0.0版本支持增加了对幂等的支持。幂等是针对生产者角度的特性。幂等可以保证生产者发送的消息，不会丢失（底层retries重试机制支撑），而且不会重复（幂等去重机制保证）。实现幂等的关键点就是服务端可以区分请求是否重复，过滤掉重复的请求。要区分请求是否重复的有两点：

• 唯一标识：要想区分请求是否重复，请求中就得有唯一标识。例如支付请求中，订单号就是唯一标识
• 记录下已处理过的请求标识：光有唯一标识还不够，还需要记录下那些请求是已经处理过的，这样当收到新的请求时，用新请求中的标识和处理记录进行比较，如果处理记录中有相同的标识，说明是重复记录，拒绝掉。

1.3.2 Kafka幂等实现策略

1、幂等又称为exactly once（精准一次）。要停止多次处理消息，必须仅将其持久化到Kafka Topic中仅仅一次。在初始化期间，kafka会给生产者生成一个唯一的ID称为Producer ID或PID。

2、PID和序列号与消息捆绑在一起，然后发送给Broker。由于序列号从零开始并且单调递增，因此，仅当消息的序列号比该PID / TopicPartition对中最后提交的消息正好大1时，Broker才会接受该消息。如果不是这种情况，则Broker认定是生产者重新发送该消息。

3、对应配置项：enable.idempotence= false 默认关闭，开启设置为true

4、注意:在使用幂等性的时候，要求必须开启retries=true和acks=all（保证不丢）

5、max.in.flight.requests.per.connection配置项默认是5，如果我们要保证严格有序，我们可以设置为1。该配置项表达的意思为：在发生阻塞之前，客户端的一个连接上允许出现未确认请求的最大数量。

Tips: 精准一次的概念尝尝出现在流式处理中

代码配置实现：

public class KafkaProducerDemo_02 {
    public static void main(String[] args) {
        //1.创建链接参数
        Properties props=new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"CentOSA:9092,CentOSB:9092,CentOSC:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
        props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,UserDefineProducerInterceptor.class.getName());
        // 将检测超时时间设置为1ms，方便触发看到重试机制
        props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG,1);
        // ACKS要设置为all
        props.put(ProducerConfig.ACKS_CONFIG,"all");
        // 重试3次
        props.put(ProducerConfig.RETRIES_CONFIG,3);
        // 开启幂等
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,true);
        
        // 开启幂等，客户端的一个连接上允许出现未确认请求的最大数量要大于1小于5。设置为1可以保证顺序。
        // 如果有一个发送不成功，就阻塞，一直等待发送成功为止
        props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION,1);

        //2.创建生产者
        KafkaProducer<String,String> producer=new KafkaProducer<String, String>(props);


        ProducerRecord<String, String> record = new ProducerRecord<>("topic01", "idempotence", "test idempotence");
        producer.send(record);
   
        producer.close();
    }
}

总结：开启幂等会保证不会重复传送消息到消息队列，客户端的一个连接上允许出现未确认请求的最大数量设置为1的话，可以保证顺序不会乱。要需要使用幂等功能，kafka的版本需要保证在0.11.0.0版本以上

以上的保证顺序，保证唯一，只是针对一个分区而言，如果kafka有多个分区，那么就需要用Kafka的事务来控制原子性，事务能控制不重复，但无法控制多分区全局有序

1.4 Kafka的事务控制

1、Kafka的幂等性，只能保证一条记录的在分区发送的原子性，但是如果要保证多条记录（多分区不重复，但多分区无法有序，参照第一章）之间的完整性，这个时候就需要开启kafk的事务操作。事务一般是把消费者和生产者绑定，中间业务系统对下游Kafka的生产失败了，中间业务系统消费过上游的Kafka偏移量不提交

2、在Kafka0.11.0.0除了引入的幂等性的概念，同时也引入了事务的概念。通常Kafka的事务分为 生产者事务Only、消费者&生产者事务。一般来说默认消费者消费的消息的级别是read_uncommited数据，这有可能读取到事务失败的数据，所有在开启生产者事务之后，需要用户设置消费者的事务隔离级别。

3、默认配置项为：isolation.level = read_uncommitted

4、该选项有两个值read_committed|read_uncommitted，如果开始事务控制，消费端必须将事务的隔离级别设置为read_committed，能够保证在回滚后清除kafka中存储的该条发送信息

5、开启的生产者事务的时候，只需要指定transactional.id属性即可，一旦开启了事务，默认生产者就已经开启了幂等性。但是要求”transactional.id”的取值必须是唯一的，同一时刻只能有一个”transactional.id”存储在，其他的将会被关闭。

1.4.1 生产者事务only使用场景

1、生产者

public class KafkaProducerDemo02 {
    public static void main(String[] args) {

        //1.生产者&消费者的配置项
        KafkaProducer<String,String> producer=buildKafkaProducer();

        producer.initTransactions();//1、初始化事务

        try{
            while(true){
           
                //2、开启事务控制
                producer.beginTransaction();
                for(i=0; i<10; i++) {
                    if(i == 8) {
                        // 异常
                        int j = 10/0;
                    }
                    //创建Record
                    ProducerRecord<String,String> producerRecord=
                    new ProducerRecord<String,String>("topic01","transation","error......");
                    
                    producer.send(producerRecord);
                    // 事务终止前，把之前数据刷入kafka队列
                    ptoducer.flush();
                }
                //3、提交事务
                producer.sendOffsetsToTransaction(offsets,"group01");
                producer.commitTransaction();
            }
        }catch (Exception e){
            producer.abortTransaction();//4、终止事务
        }finally {
            producer.close();
        }
    }
    
    // 生产者在生产环境的一些常规配置
    public static KafkaProducer<String,String> buildKafkaProducer(){
        Properties props=new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"CentOSA:9092,CentOSB:9092,CentOSC:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
        // 必须配置事务id，且唯一
        props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG,"transaction-id" + UUID.randomUUID().toString());
        // 配置批处理大小，达到1024字节需要提交
        props.put(ProducerConfig.BATCH_SIZE_CONFIG,1024);
        // 当没达到1024字节，但是时间达到了5ms，也需要提交给集群的topic
        props.put(ProducerConfig.LINGER_MS_CONFIG,5);
        // 配置幂等,和重试
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,true);
        // ack
        props.put(ProducerConfig.ACKS_CONFIG,"all");
        // 请求超时重发时间20ms
        props.put(ProducerConfig.REQUERT_TIMEOUT_MS_CONFIG,20000);
        return new KafkaProducer<String, String>(props);
    }
    
    public static KafkaConsumer<String,String> buildKafkaConsumer(String group){
        Properties props=new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"CentOSA:9092,CentOSB:9092,CentOSC:9092");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
        props.put(ConsumerConfig.GROUP_ID_CONFIG,group);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);
        props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG,"read_committed");

        return new KafkaConsumer<String, String>(props);
    }
}

2、消费者

public class KafkaConsumerDemo {
    public static void main(String[] args) {
        //1.创建Kafka链接参数
        Properties props=new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"CentOSA:9092,CentOSB:9092,CentOSC:9092");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
        // 消费者所属的消费组
        props.put(ConsumerConfig.GROUP_ID_CONFIG,"group01");
        // 设置消费者消费事务的隔离级别read_committed，消费者不可能读到未提交的数据
        props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG,"read_committed");

        //2.创建Topic消费者
        KafkaConsumer<String,String> consumer=new KafkaConsumer<String, String>(props);
        //3.订阅topic开头的消息队列
        consumer.subscribe(Pattern.compile("topic01"));

        while (true){
            ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(1));
            Iterator<ConsumerRecord<String, String>> recordIterator = consumerRecords.iterator();
            while (recordIterator.hasNext()){
                ConsumerRecord<String, String> record = recordIterator.next();
                String key = record.key();
                String value = record.value();
                long offset = record.offset();
                int partition = record.partition();
                System.out.println("key:"+key+",value:"+value+",partition:"+partition+",offset:"+offset);
            }
        }
    }
}

1.4.1 生产者消费者事务

public class KafkaProducerDemo02 {
    public static void main(String[] args) {

        //1.生产者&消费者
        KafkaProducer<String,String> producer=buildKafkaProducer();
        KafkaConsumer<String, String> consumer = buildKafkaConsumer("group01");
        
        // 消费者先订阅消费topic数据
        consumer.subscribe(Arrays.asList("topic01"));
        producer.initTransactions();//初始化事务

        try{
            while(true){
                // 消费者1秒钟拉取一次数据
                ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(1));
                // 获取的消息迭代
                Iterator<ConsumerRecord<String, String>> consumerRecordIterator = consumerRecords.iterator();
                //开启事务控制
                producer.beginTransaction();
                // 维护偏移量
                Map<TopicPartition, OffsetAndMetadata> offsets=new HashMap<TopicPartition, OffsetAndMetadata>();
                // 消费者读取到消息后进行业务处理
                while (consumerRecordIterator.hasNext()){
                    ConsumerRecord<String, String> record = consumerRecordIterator.next();
                    //业务处理，这里创建Record，通过消费topic01的消息记录，发送到topic02
                    ProducerRecord<String,String> producerRecord=
                    new ProducerRecord<String,String>("topic02",record.key(),record.value()+"to topic02");
                    producer.send(producerRecord);
                    //记录元数据下次需要提交的便宜量
                    offsets.put(new TopicPartition(record.topic(),record.partition()),
                    new OffsetAndMetadata(record.offset()+1));
                }
                //提交事务，先提交消费者的偏移量，需要指定消费者组
                producer.sendOffsetsToTransaction(offsets,"group01");
                // 提交事务，再提交生产者的偏移量
                producer.commitTransaction();
            }
        }catch (Exception e){
            // 消费者端业务处理逻辑出现错误，要捕获回滚
            producer.abortTransaction();//终止事务
        }finally {
            producer.close();
        }
    }
    public static KafkaProducer<String,String> buildKafkaProducer(){
        Properties props=new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"CentOSA:9092,CentOSB:9092,CentOSC:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
        // 必须配置事务id，且唯一
        props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG,"transaction-id" + UUID.randomUUID().toString());
        // 配置批处理大小，达到1024字节需要提交
        props.put(ProducerConfig.BATCH_SIZE_CONFIG,1024);
        // 当没达到1024字节，但是时间达到了5ms，也需要提交给集群的topic
        props.put(ProducerConfig.LINGER_MS_CONFIG,5);
        // 配置幂等,和重试
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,true);
        // ack
        props.put(ProducerConfig.ACKS_CONFIG,"all");
        // 请求超时重发时间20ms
        props.put(ProducerConfig.REQUERT_TIMEOUT_MS_CONFIG,20000);
        return new KafkaProducer<String, String>(props);
    }
    public static KafkaConsumer<String,String> buildKafkaConsumer(String group){
        Properties props=new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"CentOSA:9092,CentOSB:9092,CentOSC:9092");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
        props.put(ConsumerConfig.GROUP_ID_CONFIG,group);
        // 消费者自动提交offset策略，关闭，必须设置
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);
        // 设置消费者消费事务的隔离级别read_committed，消费者不可能读到未提交的数据
        props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG,"read_committed");

        return new KafkaConsumer<String, String>(props);
    }
}

总结：消费端掌握偏移量控制，生产者端掌握超时重传应答重试和分区幂等。实际的生产开发过程中，要熟练掌握事务控制，包括生产者only和生产者&消费者事务控制。kafka事务在分布式微服务的开发中，有比较强的应用。