7.【kafka运维】 kafka-consumer-groups.sh消费者组管理

清疚 2022-09-04 10:58 225阅读 0赞

### 文章目录 ###

*   *  消费者组管理 kafka-consumer-groups.sh
     *   *  1. 查看消费者列表\`--list\`
         *  2. 查看消费者组详情\`--describe\`
         *  3. 删除消费者组\`--delete\`
         *  4. 重置消费组的偏移量 \`--reset-offsets\`
         *  5. 删除偏移量\`delete-offsets\`
     *  More

> **日常运维** 、**问题排查** 怎么能够少了滴滴开源的  
> [滴滴开源LogiKM一站式Kafka监控与管控平台][LogiKM_Kafka]

## 消费者组管理 kafka-consumer-groups.sh ##

### 1. 查看消费者列表`--list` ###

> `sh bin/kafka-consumer-groups.sh --bootstrap-server xxxx:9090 --list`  
> ![在这里插入图片描述][20210625175303377.png]

先调用`MetadataRequest`拿到所有在线Broker列表  
再给每个Broker发送`ListGroupsRequest`请求获取 消费者组数据

### 2. 查看消费者组详情`--describe` ###

`DescribeGroupsRequest`

**查看消费组详情`--group` 或 `--all-groups`**

> **查看指定消费组详情`--group`**  
> `sh bin/kafka-consumer-groups.sh --bootstrap-server xxxxx:9090 --describe --group test2_consumer_group`
> 
> --------------------
> 
> **查看所有消费组详情`--all-groups`**  
> `sh bin/kafka-consumer-groups.sh --bootstrap-server xxxxx:9090 --describe --all-groups`  
> 查看该消费组 消费的所有Topic、及所在分区、最新消费offset、Log最新数据offset、Lag还未消费数量、消费者ID等等信息  
> ![在这里插入图片描述][20210625174905816.png]

**查询消费者成员信息`--members`**

> **所有消费组成员信息**  
> `sh bin/kafka-consumer-groups.sh --describe --all-groups --members --bootstrap-server xxx:9090`  
> **指定消费组成员信息**  
> `sh bin/kafka-consumer-groups.sh --describe --members --group test2_consumer_group --bootstrap-server xxxx:9090`  
> ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY_size_16_color_FFFFFF_t_70]

**查询消费者状态信息`--state`**

> **所有消费组状态信息**  
> `sh bin/kafka-consumer-groups.sh --describe --all-groups --state --bootstrap-server xxxx:9090`  
> **指定消费组状态信息**  
> `sh bin/kafka-consumer-groups.sh --describe --state --group test2_consumer_group --bootstrap-server xxxxx:9090`  
> ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY_size_16_color_FFFFFF_t_70 1]

### 3. 删除消费者组`--delete` ###

`DeleteGroupsRequest`

**删除消费组–delete**

> **删除指定消费组`--group`**  
> `sh bin/kafka-consumer-groups.sh --delete --group test2_consumer_group --bootstrap-server xxxx:9090`  
> **删除所有消费组`--all-groups`**  
> `sh bin/kafka-consumer-groups.sh --delete --all-groups --bootstrap-server xxxx:9090`

**PS: 想要删除消费组前提是这个消费组的所有客户端都停止消费/不在线才能够成功删除;否则会报下面异常**

Error: Deletion of some consumer groups failed:
    * Group 'test2_consumer_group' could not be deleted due to: java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.GroupNotEmptyException: The group is not empty.

### 4. 重置消费组的偏移量 `--reset-offsets` ###

能够执行成功的一个前提是 消费组这会是不可用状态;

下面的示例使用的参数是: `--dry-run` ;这个参数表示预执行,会打印出来将要处理的结果;  
等你想真正执行的时候请换成参数`--excute` ;

下面示例 重置模式都是 `--to-earliest` 重置到最早的;

请根据需要参考下面 **相关重置Offset的模式** 换成其他模式;

**重置指定消费组的偏移量 `--group`**

> **重置指定消费组的所有Topic的偏移量`--all-topic`**  
> `sh bin/kafka-consumer-groups.sh --reset-offsets --to-earliest --group test2_consumer_group --bootstrap-server xxxx:9090 --dry-run --all-topic`  
> **重置指定消费组的指定Topic的偏移量`--topic`**  
> `sh bin/kafka-consumer-groups.sh --reset-offsets --to-earliest --group test2_consumer_group --bootstrap-server xxxx:9090 --dry-run --topic test2`

**重置所有消费组的偏移量 `--all-group`**

> **重置所有消费组的所有Topic的偏移量`--all-topic`**  
> `sh bin/kafka-consumer-groups.sh --reset-offsets --to-earliest --all-group --bootstrap-server xxxx:9090 --dry-run --all-topic`  
> **重置所有消费组中指定Topic的偏移量`--topic`**  
> `sh bin/kafka-consumer-groups.sh --reset-offsets --to-earliest --all-group --bootstrap-server xxxx:9090 --dry-run --topic test2`

`--reset-offsets` 后面需要接**重置的模式**

**相关重置Offset的模式**

<table> 
 <thead> 
  <tr> 
   <th>参数</th> 
   <th>描述</th> 
   <th>例子</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td><strong><code>--to-earliest</code> :</strong></td> 
   <td>重置offset到最开始的那条offset(找到还未被删除最早的那个offset)</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><strong><code>--to-current</code>:</strong></td> 
   <td>直接重置offset到当前的offset，也就是LOE</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><strong><code>--to-latest</code>：</strong></td> 
   <td>重置到最后一个offset</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><strong><code>--to-datetime</code>:</strong></td> 
   <td>重置到指定时间的offset;格式为:<code>YYYY-MM-DDTHH:mm:SS.sss</code>;</td> 
   <td><code>--to-datetime "2021-6-26T00:00:00.000"</code></td> 
  </tr> 
  <tr> 
   <td><code>--to-offset</code></td> 
   <td>重置到指定的offset,但是通常情况下,匹配到多个分区,这里是将匹配到的所有分区都重置到这一个值; 如果 1.目标最大offset&lt;<code>--to-offset</code>, 这个时候重置为目标最大offset；2.目标最小offset&gt;<code>--to-offset</code> ，则重置为最小; 3.否则的话才会重置为<code>--to-offset</code>的目标值; <strong>一般不用这个</strong></td> 
   <td><code>--to-offset 3465</code> <img src="https://img-blog.csdnimg.cn/20210626153528929.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY=,size_16,color_FFFFFF,t_70" alt="在这里插入图片描述"></td> 
  </tr> 
  <tr> 
   <td><code>--shift-by</code></td> 
   <td>按照偏移量增加或者减少多少个offset；正的为往前增加;负的往后退；当然这里也是匹配所有的;</td> 
   <td><code>--shift-by 100</code> 、<code>--shift-by -100</code></td> 
  </tr> 
  <tr> 
   <td><code>--from-file</code></td> 
   <td>根据CVS文档来重置; 这里下面单独讲解</td> 
   <td></td> 
  </tr> 
 </tbody> 
</table>

**`--from-file`着重讲解一下**

> 上面其他的一些模式重置的都是匹配到的所有分区; 不能够每个分区重置到不同的offset；不过\*\*`--from-file`\*\*可以让我们更灵活一点;

1.  先配置cvs文档  
    格式为: Topic:分区号: 重置目标偏移量
    
        test2,0,100
        test2,1,200
        test2,2,300
2.  执行命令
    
    > `sh bin/kafka-consumer-groups.sh --reset-offsets --group test2_consumer_group --bootstrap-server xxxx:9090 --dry-run --from-file config/reset-offset.csv`

### 5. 删除偏移量`delete-offsets` ###

能够执行成功的一个前提是 消费组这会是不可用状态;

偏移量被删除了之后,Consumer Group下次启动的时候,会从头消费;

> `sh bin/kafka-consumer-groups.sh --delete-offsets --group test2_consumer_group2 --bootstrap-server XXXX:9090 --topic test2`

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

相关可选参数

<table> 
 <thead> 
  <tr> 
   <th>参数</th> 
   <th>描述</th> 
   <th>例子</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td><code>--bootstrap-server</code></td> 
   <td>指定连接到的kafka服务;</td> 
   <td>–bootstrap-server localhost:9092</td> 
  </tr> 
  <tr> 
   <td><code>--list</code></td> 
   <td>列出所有消费组名称</td> 
   <td><code>--list</code></td> 
  </tr> 
  <tr> 
   <td><code>--describe</code></td> 
   <td>查询消费者描述信息</td> 
   <td><code>--describe</code></td> 
  </tr> 
  <tr> 
   <td><code>--group</code></td> 
   <td>指定消费组</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--all-groups</code></td> 
   <td>指定所有消费组</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--members</code></td> 
   <td>查询消费组的成员信息</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--state</code></td> 
   <td>查询消费者的状态信息</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--offsets</code></td> 
   <td>在查询消费组描述信息的时候,这个参数会列出消息的偏移量信息; 默认就会有这个参数的;</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>dry-run</code></td> 
   <td>重置偏移量的时候,使用这个参数可以让你预先看到重置情况，这个时候还没有真正的执行,真正执行换成<code>--excute</code>;默认为<code>dry-run</code></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--excute</code></td> 
   <td>真正的执行重置偏移量的操作;</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--to-earliest</code></td> 
   <td>将offset重置到最早</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>to-latest</code></td> 
   <td>将offset重置到最近</td> 
   <td></td> 
  </tr> 
 </tbody> 
</table>

## More ##

Kafka专栏持续更新中…(源码、原理、实战、运维、视频、面试视频)

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

[【kafka运维】Kafka全网最全最详细运维命令合集(精品强烈建议收藏!!!)\_石臻臻的杂货铺-CSDN博客][kafka_Kafka_-CSDN]

[【kafka实战】分区重分配可能出现的问题和排查问题思路(生产环境实战,干货!!!非常干!!!建议收藏)][kafka]

[【kafka异常】kafka 常见异常处理方案(持续更新! 建议收藏)][kafka_kafka _]

[【kafka运维】分区从分配、数据迁移、副本扩缩容 (附教学视频)][kafka_]

[【kafka源码】ReassignPartitionsCommand源码分析(副本扩缩、数据迁移、副本重分配、副本跨路径迁移][kafka_ReassignPartitionsCommand]

[【kafka】点击更多…][kafka 1]

[LogiKM_Kafka]: https://github.com/didi/LogiKM
[20210625175303377.png]: /images/20220829/e644f9ae5e984407a17e1ab0c8d1432a.png
[20210625174905816.png]: /images/20220829/3850d246b3434f22b217becbb6b191a4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY_size_16_color_FFFFFF_t_70]: /images/20220829/82339680f14c49de9199ed37dd3d5b5a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY_size_16_color_FFFFFF_t_70 1]: /images/20220829/3d06464edcc746ecaf895a0d0ae4792e.png
[kafka_Kafka_-CSDN]: https://blog.csdn.net/u010634066/article/details/118215928?spm=1001.2014.3001.5501
[kafka]: https://blog.csdn.net/u010634066/article/details/118631272?spm=1001.2014.3001.5501
[kafka_kafka _]: https://blog.csdn.net/u010634066/article/details/118105676?spm=1001.2014.3001.5501
[kafka_]: https://blog.csdn.net/u010634066/article/details/118028403?spm=1001.2014.3001.5501
[kafka_ReassignPartitionsCommand]: https://blog.csdn.net/u010634066/article/details/118051963
[kafka 1]: https://mp.weixin.qq.com/mp/appmsgalbum?__biz=Mzg4ODY1NTcxNg==&action=getalbum&album_id=1966026980307304450#wechat_redirect