kafka的offset笔记

心已赠人 2023-01-10 13:14 127阅读 0赞

# 版本 #

這個看起來有點多此一舉，

我一開始也是這麼想的。

後來經過測試發現，新版本的kafka已經不再兼容老版本的kafka中的命令了，所以本篇記錄是爲了針對新版本的kafka的相關操作的。

<table style="width:500px;"> 
 <tbody> 
  <tr> 
   <td><strong>組件</strong></td> 
   <td><strong>版本</strong></td> 
  </tr> 
  <tr> 
   <td>Zookeeper</td> 
   <td>3.6.0</td> 
  </tr> 
  <tr> 
   <td>Kafka</td> 
   <td>2.5.0</td> 
  </tr> 
 </tbody> 
</table>

# 概念 #

**①生产者Offset**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FwcGxleXVjaGk_size_16_color_FFFFFF_t_70][]

生产者写入topic的各个partition时，有多少个partition就有多少个offset

**②消费者Offset**

![file][]

这是某一个分区的offset情况，我们已经知道生产者写入的offset是最新最大的值也就是12，

而当Consumer A进行消费时，他从0开始消费，一直消费到了9，他的offset就记录在了9，

Consumer B就纪录在了11。

等下一次他们再来消费时，他们可以选择接着上一次的位置消费，当然也可以选择从头消费，或者跳到最近的记录并从“现在”开始消费。

此时，每个ｐａｒｔｉｔｉｏｎ有多少消费组，那就有多少个offset

**消费者组**

![file][file 1]

费者组的概念其实并不影响对offset的理解，上面的情况Consumer A，Consumer B如果是同组就不能同时消费一个分区的消息，不同组的消费者可以同时消费一个分区的消息。

还有一种offset的说法，就是consumer消费未提交时，本地是有另外一个offset的，这个offset不一定与集群中记录的offset一致。

所以，kafka每一个topic分区和生产者，消费者不同，是有多个offset的。

概念总结如下：

offset是指某一个**分区**的偏移量。

topic partition offset 这三个唯一确定一条消息。

生产者的offset其实就是最新的offset。

消费者的offset是他自己维护的，他可以选择分区最开始，最新，也可以记住他消费到哪了。

消费者组是为了不同组的消费者可以同时消费一个分区的消息。

# 關於Group這個概念 #

**首先注意，producer沒有group的說法，**

**kafka提到group一定是consumer這邊。**

# **Group-id位置** #

<table style="width:500px;"> 
 <tbody> 
  <tr> 
   <td style="width:319px;"><strong>具體文件</strong></td> 
   <td style="width:179px;"><strong>變量</strong></td> 
  </tr> 
  <tr> 
   <td style="width:319px;">$KAFKA/config/consumer.properties</td> 
   <td style="width:179px;">group.id</td> 
  </tr> 
  <tr> 
   <td style="width:319px;">$KAFKA/config/connect-distributed.properties</td> 
   <td style="width:179px;">group.id</td> 
  </tr> 
 </tbody> 
</table>

如果在新建topic的時候，不特別指定，那麼默認使用的是consumer.properties裏面的group.id

### \_\_consumer\_offsets的哪個partition保存了consumer group的位移信息 ###

查看$KAFKA/config/consumer.properties

得到group.id是test-consumer-group，

填入下面的代碼並運行

public class kafka_hash {
    
            public static void main(String args[]){
                System.out.println(Math.abs("test-consumer-group".hashCode()) % 50);
    
            }
        }

實驗結果爲31

也就是說**\_\_consumer\_offsets的partition 31保存了consumer group的位移信息**

**記住31這個數字，後面會用到**

# 操作 #

<table> 
 <tbody> 
  <tr> 
   <td style="width:297px;"> <h3><strong>操作</strong></h3> </td> 
   <td> <h3><strong>具體命令</strong></h3> </td> 
  </tr> 
  <tr> 
   <td style="width:297px;">创建topic</td> 
   <td>$KAFKA/bin/kafka-topics.sh --zookeeper Desktop:2181 --create --topic mytopic --replication-factor 3 --partitions 3</td> 
  </tr> 
  <tr> 
   <td style="width:297px;">查看topic列表</td> 
   <td>$KAFKA/bin/kafka-topics.sh --list --zookeeper Desktop:2181</td> 
  </tr> 
  <tr> 
   <td style="width:297px;">生產數據</td> 
   <td>$KAFKA/bin/kafka-console-producer.sh --broker-list Desktop:9091 --topic mytopic</td> 
  </tr> 
  <tr> 
   <td style="width:297px;">存放在各個partition的offset終點</td> 
   <td>$KAFKA/bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list Desktop:9091,Laptop:9092,Laptop:9093 --topic mytopic --time -1</td> 
  </tr> 
  <tr> 
   <td style="width:297px;">存放在各個partition的offset起點</td> 
   <td> <p>$KAFKA/bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list Desktop:9091,Laptop:9092,Laptop:9093 --topic mytopic --time -2</p> </td> 
  </tr> 
  <tr> 
   <td style="width:297px;"><strong>查询__consumer_offsets topic所有内容</strong></td> 
   <td>$KAFKA/bin/kafka-console-consumer.sh --consumer.config $KAFKA/config/consumer.properties \<br> --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" \<br> --bootstrap-server Desktop:9091 --topic __consumer_offsets --from-beginning</td> 
  </tr> 
  <tr> 
   <td style="width:297px;">查詢<strong>__consumer_offsets的partition 31包含的關於mytopic的offset信息</strong></td> 
   <td> <pre><code>$KAFKA/bin/kafka-console-consumer.sh  \
--formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" \
--bootstrap-server Desktop:9091 --topic __consumer_offsets  --partition 31 --from-beginning</code></pre> <p>&nbsp;</p> </td> 
  </tr> 
 </tbody> 
</table>

上面的最後一句命令會得到：

\[test-consumer-group,mytopic,1\]::OffsetAndMetadata(offset=3, leaderEpoch=Optional\[0\], metadata=, commitTimestamp=1611494686173, expireTimestamp=None)  
\[test-consumer-group,mytopic,0\]::OffsetAndMetadata(offset=4, leaderEpoch=Optional\[0\], metadata=, commitTimestamp=1611494686173, expireTimestamp=None)  
\[test-consumer-group,mytopic,2\]::OffsetAndMetadata(offset=4, leaderEpoch=Optional\[0\], metadata=, commitTimestamp=1611494691174, expireTimestamp=None)  
\[test-consumer-group,mytopic,1\]::OffsetAndMetadata(offset=3, leaderEpoch=Optional\[0\], metadata=, commitTimestamp=1611494691174, expireTimestamp=None)

上述表格中produce端需要的数据如下:

{"user_id": "543462", "item_id":"1715", "category_id": "1464116", "behavior": "pv", "ts": "2017-11-26T01:00:00Z"}
    {"user_id": "662867", "item_id":"2244074", "category_id": "1575622", "behavior": "pv", "ts": "2017-11-26T01:00:00Z"}
    {"user_id": "561558", "item_id":"3611281", "category_id": "965809", "behavior": "pv", "ts": "2017-11-26T01:00:00Z"}
    {"user_id": "894923", "item_id":"3076029", "category_id": "1879194", "behavior": "pv", "ts": "2017-11-26T01:00:00Z"}
    {"user_id": "834377", "item_id":"4541270", "category_id": "3738615", "behavior": "pv", "ts": "2017-11-26T01:00:00Z"}
    {"user_id": "315321", "item_id":"942195", "category_id": "4339722", "behavior": "pv", "ts": "2017-11-26T01:00:00Z"}
    {"user_id": "625915", "item_id":"1162383", "category_id": "570735", "behavior": "pv", "ts": "2017-11-26T01:00:00Z"}
    {"user_id": "578814", "item_id":"176722", "category_id": "982926", "behavior": "pv", "ts": "2017-11-26T01:00:00Z"}
    {"user_id": "873335", "item_id":"1256540", "category_id": "1451783", "behavior": "pv", "ts": "2017-11-26T01:00:00Z"}
    {"user_id": "429984", "item_id":"4625350", "category_id": "2355072", "behavior": "pv", "ts": "2017-11-26T01:00:00Z"}
    {"user_id": "866796", "item_id":"534083", "category_id": "4203730", "behavior": "pv", "ts": "2017-11-26T01:00:00Z"}
    {"user_id": "937166", "item_id":"321683", "category_id": "2355072", "behavior": "pv", "ts": "2017-11-26T01:00:00Z"}

# 其他 #

另外，根據\[5\],在同一個consumer group中，對於同一個topic而言，只能有一個consumer消費到其中特定的一條數據。

\[6\]中提到了Coordinator

**Reference:**

\[1\][Kafka到底有几个Offset？——Kafka核心之偏移量机制][Kafka_Offset_Kafka]  
\[2\][自己维护kafka\_offset中的坑][kafka_offset]

\[3\][kafka查询最新producer offset的命令][kafka_producer offset]

\[4\][Kafka 如何读取offset topic内容 (\_\_consumer\_offsets)][Kafka _offset topic_ _consumer_offsets]

\[5\][多个consumer使用同一个group.id消费同一个topic][consumer_group.id_topic]

\[6\][Kafka消费组(consumer group)][Kafka_consumer group]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FwcGxleXVjaGk_size_16_color_FFFFFF_t_70]: /images/20221119/becdc3cba5f44a699ccd0f4948126c35.png
[file]: /images/20221119/2455471dd5be4c5fa75777088c4b4f83.png
[file 1]: https://img-blog.csdnimg.cn/img_convert/09e807209b530b8fa6ac4cf14dde1cf4.png
[Kafka_Offset_Kafka]: https://www.cnblogs.com/tree1123/p/11419526.html
[kafka_offset]: https://www.jianshu.com/p/316e50a570dd?utm_campaign
[kafka_producer offset]: https://blog.csdn.net/wyqwilliam/article/details/84428565
[Kafka _offset topic_ _consumer_offsets]: https://www.cnblogs.com/huxi2b/p/6061110.html
[consumer_group.id_topic]: https://blog.csdn.net/u012307002/article/details/51308693/
[Kafka_consumer group]: https://www.cnblogs.com/huxi2b/p/6223228.html