kafka消费者offset记录位置和方式

我不是女神ヾ 2022-04-02 04:20 646阅读 0赞

### 1. 前言 ###

kafka消费者在会保存其消费的进度，也就是offset，存储的位置根据选用的`kafka-api`不同而不同。

### 2. zookeeper存储 ###

首先来说说消费者如果是根据`java-api`来消费，也就是`kafka.javaapi.consumer.ConsumerConnector`，通过配置参数`zookeeper.connect`来消费。这种情况下，消费者的offset会更新到zookeeper的`consumers/{group}/offsets/{topic}/{partition}`目录下，例如：

[zk: 100.5.14.161:2181(CONNECTED) 20] get /consumers/console-consumer-21030/offsets/topic_test_1/0
    37
    cZxid = 0x50000053a
    ctime = Thu Dec 20 17:26:58 CST 2018
    mZxid = 0x50000053a
    mtime = Thu Dec 20 17:26:58 CST 2018
    pZxid = 0x50000053a
    cversion = 0
    dataVersion = 0
    aclVersion = 0
    ephemeralOwner = 0x0
    dataLength = 2
    numChildren = 0

### 3. kafka topic存储 ###

如果是根据kafka默认的api来消费，即`org.apache.kafka.clients.consumer.KafkaConsumer`，我们会配置参数`bootstrap.servers`来消费。而其消费者的offset会更新到一个kafka自带的topic：`__consumer_offsets`下面，查看当前group的消费进度，则要依靠kafka自带的工具`kafka-consumer-offset-checker`，例如：

[2018-12-25 11:48:10,163] WARN WARNING: ConsumerOffsetChecker is deprecated and will be dropped in releases following 0.9.0. Use ConsumerGroupCommand instead. (kafka.tools.ConsumerOffsetChecker$)
    Group           Topic                          Pid Offset          logSize         Lag             Owner
    test_group_1    topic_test_1                   0   489             491             2               none
    test_group_1    topic_test_1                   1   489             490             1               none
    test_group_1    topic_test_1                   2   490             491             1               none

### 4. offset更新方式 ###

offset更新的方式，不区分是用的哪种api，大致分为两类：

*  自动提交，设置enable.auto.commit=true，更新的频率根据参数【auto.commit.interval.ms】来定。这种方式也被称为【at most once】，fetch到消息后就可以更新offset，无论是否消费成功。
 *  手动提交，设置enable.auto.commit=false，这种方式称为【at least once】。fetch到消息后，等消费完成再调用方法【consumer.commitSync()】，手动更新offset；如果消费失败，则offset也不会更新，此条消息会被重复消费一次。

--------------------

##### 参考： https://blog.csdn.net/u013063153/article/details/78122088 #####