kafka位移提交

缺乏、安全感 2023-07-24 14:55 28阅读 0赞

位移，反应到kakfa的源码中，就是offset。offset，有人叫偏移量，有人叫位移。<深入理解kafka>的作者做了一下区分，如果讲的是消息在分区中的位置，就用偏移量，如果讲的是消费者端，那就用位移表示。  
kafka消费完消息，需要提交位移。kafka默认是自动提交位移的，定时提交，中间的时间间隔为：5秒。这个值是由auto.commit.interval.ms配置。自动提交位移会有重复消费的风险。  
原因是：当消费者拉取了一批消息，开始消费。过5秒钟，将该批次的最大位移进行提交。然后开始拉取下一批次的消息。拉取成功后，消费者就崩溃了，没有将本批次的消费位移提交。待消费者恢复后，会检查服务端的\_consumer\_offset\_主题，我上一次提交的位移是多少，查到的是上一批次的最大消息位移。那本次拉取的部分消息就又消费了一遍。消息重复消费。  
提交的位移=本批次消息的最大位移+1。  
如图，如果当前消费的位置是x，则提交的唯一就是x+1。下一次拉取也是从x+1开始  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5ODM5MDc1_size_16_color_FFFFFF_t_70]  
自动提交位移是kafka操作，应用程序不需要插手。如果我们想自己操作位移提交，那就需要将enable.auto.commit设定为false。然后使用消费者的API提交位移。提交位移的API一共有2大类。  
//同步提交  
commitSync()  
//异步提交  
commitAsync()  
**同步提交**，消费线程会阻塞，等待提交操作成功。阻塞也是分情况的，如果提交过程中发生了不可恢复的异常，比如：CommitFailedException、WakeupException、InterruptException、AuthenticationException、AuthorizationException等，就不会阻塞，而会直接抛出异常。我们要捕获这些异常，进行相应的处理  
**异步提交**，消费线程不阻塞。可能位移提交还没有返回，新一批次的拉取操作已经开始了。  
从安全性角度来考虑，同步提交，肯定要更靠谱一点，但是kafka的吞吐量会急速下滑。  
生产中，如果采用同步提交的方式。不要采用消费一条，提交一条的方式。更多的是采用拉取一个批次的消息，然后开始处理，最后提交本批次消息的最大位移。  
异步提交的相关API有：

void commitAsync();
    
        void commitAsync(OffsetCommitCallback var1);
    
        void commitAsync(Map<TopicPartition, OffsetAndMetadata> var1, OffsetCommitCallback var2);

第一个，就是普通的异步提交  
第二个API，注册了一个位移提交的回调函数，我们可以从该回调函数中判断，异步提交位移是否成功了。  
第三个API，第一个入参是一个MAP对象，代表提交哪个分区的哪个位移，粒度更细。同时，也注册了一个回调函数。

kafkaConsumer.commitAsync(offsetAndMetadataMaps, new OffsetCommitCallback() { 
                    @Override
                    public void onComplete(Map<TopicPartition, OffsetAndMetadata> map, Exception e) { 
                    //提交失败，e有值。提交成功，e为空
                        if (e != null) { 
                            e.printStackTrace();
                        } else { 
                            System.out.println(map);
                        } 
                    }
                });

异步提交，同样会有失败的情况发生。<深入理解kafka>的作者给了一个思路。我们可以维护一个递增的序号，每次位移提交成功后，就递增该序号。当提交位移失败，进行重试时，先比较准备提交的位移和序号。如果后者大，说明已经提交了更大的唯一，不需要再次提交。如果两者相等，此时可以提交位移。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5ODM5MDc1_size_16_color_FFFFFF_t_70]: /images/20230528/c67c83f0f8814b298ed24c1c417c2f11.png