Kafka——》消费组管理kafka-consumer-groups

痛定思痛。 2024-03-31 15:32 102阅读 0赞

> 推荐链接：  
> [总结——》【Java】][Java]  
> [总结——》【Mysql】][Mysql]  
> [总结——》【Redis】][Redis]  
> [总结——》【Kafka】][Kafka]  
> [总结——》【Spring】][Spring]  
> [总结——》【SpringBoot】][SpringBoot]  
> [总结——》【MyBatis、MyBatis-Plus】][MyBatis_MyBatis-Plus]

#### Kafka——》消费组管理kafka-consumer-groups ####

*  list：查看消费者组列表
 *  describe：查看消费者组详情
 *  state：查看消费者组状态
 *  members：查看消费者组成员
 *  delete：删除消费者组
 *  reset-offsets：重置消费组的偏移量
 *  delete-offsets：删除消费组的偏移量

> KAFKA\_HOME=/opt/app/install/kafka

<table> 
 <thead> 
  <tr> 
   <th align="left"><strong>参数</strong></th> 
   <th align="left"><strong>描述</strong></th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left"><code>--bootstrap-server</code></td> 
   <td align="left">kafka服务地址</td> 
  </tr> 
  <tr> 
   <td align="left"><code>--list</code></td> 
   <td align="left">查看消费者组列表</td> 
  </tr> 
  <tr> 
   <td align="left"><code>--group</code></td> 
   <td align="left">指定消费者组</td> 
  </tr> 
  <tr> 
   <td align="left"><code>--all-groups</code></td> 
   <td align="left">所有消费者组</td> 
  </tr> 
  <tr> 
   <td align="left"><code>--describe</code></td> 
   <td align="left">查看消费者组详情</td> 
  </tr> 
  <tr> 
   <td align="left"><code>--state</code></td> 
   <td align="left">查看消费者组状态</td> 
  </tr> 
  <tr> 
   <td align="left"><code>--members</code></td> 
   <td align="left">查看消费者组成员</td> 
  </tr> 
  <tr> 
   <td align="left"><code>--delete</code></td> 
   <td align="left">删除消费者组</td> 
  </tr> 
  <tr> 
   <td align="left"><code>--reset-offsets</code></td> 
   <td align="left">重置消费组的偏移量</td> 
  </tr> 
  <tr> 
   <td align="left"><code>--delete-offsets</code></td> 
   <td align="left">删除消费组的偏移量</td> 
  </tr> 
  <tr> 
   <td align="left"><code>--dry-run</code></td> 
   <td align="left">预先执行重置偏移量</td> 
  </tr> 
  <tr> 
   <td align="left"><code>--excute</code></td> 
   <td align="left">真正执行重置偏移量</td> 
  </tr> 
  <tr> 
   <td align="left"><code>--to-earliest</code></td> 
   <td align="left">将offset重置到最早</td> 
  </tr> 
  <tr> 
   <td align="left"><code>--to-latest</code></td> 
   <td align="left">将offset重置到最近</td> 
  </tr> 
 </tbody> 
</table>

## list：查看消费者组列表 ##

# 查看消费者组列表
    bin/kafka-consumer-groups.sh --bootstrap-server x.x.x.x:9092 --list

## describe：查看消费者组详情 ##

# 查看所有消费者组详情
    bin/kafka-consumer-groups.sh --bootstrap-server 10.116.192.27:9092 --describe --all-groups
        
    # 查看指定消费者组详情
    bin/kafka-consumer-groups.sh --bootstrap-server 10.116.192.27:9092 --describe --group listenerForSyncEsfCommunity1

<table> 
 <thead> 
  <tr> 
   <th align="left"><strong>结果列</strong></th> 
   <th align="left"><strong>GROUP</strong></th> 
   <th align="left"><strong>TOPIC</strong></th> 
   <th align="left"><strong>PARTITION</strong></th> 
   <th align="left"><strong>CURRENT-OFFSET</strong></th> 
   <th align="left"><strong>LOG-END-OFFSET</strong></th> 
   <th align="left"><strong>LAG</strong></th> 
   <th align="left"><strong>CONSUMER-ID</strong></th> 
   <th align="left"><strong>HOST</strong></th> 
   <th align="left"><strong>CLIENT-ID</strong></th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">描述</td> 
   <td align="left">消费组</td> 
   <td align="left">主题</td> 
   <td align="left">分区编号</td> 
   <td align="left">当前offset</td> 
   <td align="left">最新offset</td> 
   <td align="left">消息滞后（未消费）数量</td> 
   <td align="left">消费者ID</td> 
   <td align="left">主机</td> 
   <td align="left">客户端ID</td> 
  </tr> 
 </tbody> 
</table>

## state：查看消费者组状态 ##

# 查看所有消费者组状态 
    bin/kafka-consumer-groups.sh --bootstrap-server 10.116.192.27:9092 --describe --all-groups --state
    
    # 查看指定消费者组状态
    bin/kafka-consumer-groups.sh --bootstrap-server 10.116.192.27:9092 --describe --group listenerForSyncEsfCommunity1 --state

<table> 
 <thead> 
  <tr> 
   <th align="left"><strong>结果列</strong></th> 
   <th align="left"><strong>GROUP</strong></th> 
   <th align="left"><strong>COORDINATOR (ID)</strong></th> 
   <th align="left"><strong>ASSIGNMENT-STRATEGY</strong></th> 
   <th align="left"><strong>STATE</strong></th> 
   <th align="left"><strong>#MEMBERS</strong></th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">描述</td> 
   <td align="left">消费组</td> 
   <td align="left">协调者ID</td> 
   <td align="left">分配策略</td> 
   <td align="left">状态<br>- Stable：有消费者成员<br>- Empty：没有消费者成员<br>- Dead<br>- PreparingRebalance<br>- CompletingRebalance<br></td> 
   <td align="left">成员数量</td> 
  </tr> 
 </tbody> 
</table>

## members：查看消费者组成员 ##

# 查看所有消费者组成员
    bin/kafka-consumer-groups.sh --bootstrap-server 10.116.192.27:9092 --describe --all-groups --members
    
    # 查看指定消费者组成员    
    bin/kafka-consumer-groups.sh --bootstrap-server 10.116.192.27:9092 --describe --group listenerForSyncEsfCommunity1 --members

<table> 
 <thead> 
  <tr> 
   <th align="left"><strong>结果列</strong></th> 
   <th align="left"><strong>GROUP</strong></th> 
   <th align="left"><strong>CONSUMER-ID</strong></th> 
   <th align="left"><strong>HOST</strong></th> 
   <th align="left"><strong>CLIENT-ID</strong></th> 
   <th align="left"><strong>#PARTITION</strong></th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">描述</td> 
   <td align="left">消费组</td> 
   <td align="left">消费者ID</td> 
   <td align="left">主机</td> 
   <td align="left">客户端ID</td> 
   <td align="left">分区</td> 
  </tr> 
 </tbody> 
</table>

## delete：删除消费者组 ##

# 删除所有消费者组
    bin/kafka-consumer-groups.sh --bootstrap-server 10.122.139.40:9092 --delete --all-groups
    
    # 删除指定消费者组
    bin/kafka-consumer-groups.sh --bootstrap-server 10.122.139.40:9092 --delete --group listenerForSyncEsfCommunity1
    
        
    
    # 只有这个消费组的所有客户端都停止消费/不在线才能够成功删除，否则会报下面异常
    Error: Deletion of some consumer groups failed:
    * Group 'listenerForSyncEsfCommunity1' could not be deleted due to: java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.GroupNotEmptyException: The group is not empty.

## reset-offsets：重置消费组的偏移量 ##

<table> 
 <thead> 
  <tr> 
   <th align="left"><strong>重置Offset的模式</strong></th> 
   <th align="left"><strong>描述</strong></th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left"><code>--to-earliest</code></td> 
   <td align="left">重置到最早offset</td> 
  </tr> 
  <tr> 
   <td align="left"><code>--to-latest</code></td> 
   <td align="left">重置到最近offset</td> 
  </tr> 
  <tr> 
   <td align="left"><code>--to-current</code></td> 
   <td align="left">重置到当前offset</td> 
  </tr> 
  <tr> 
   <td align="left"><code>--to-datetime</code></td> 
   <td align="left">重置到指定时间offset<br>格式：YYYY-MM-DDTHH:mm:SS.sss<br>示例：2021-6-26T00:00:00.000</td> 
  </tr> 
  <tr> 
   <td align="left"><code>--to-offset</code></td> 
   <td align="left">重置到指定offset（多个分区都重置，一般不用这个）</td> 
  </tr> 
  <tr> 
   <td align="left"><code>--shift-by</code></td> 
   <td align="left">按照偏移量增加或者减少offset<br>- 正数：往前增加<br>- 负数：往后减少<br></td> 
  </tr> 
  <tr> 
   <td align="left"><code>--from-file</code></td> 
   <td align="left">根据CVS文档来重置</td> 
  </tr> 
  <tr> 
   <td align="left"><code>--dry-run</code></td> 
   <td align="left">预先执行重置偏移量</td> 
  </tr> 
  <tr> 
   <td align="left"><code>--excute</code></td> 
   <td align="left">真正执行重置偏移量</td> 
  </tr> 
 </tbody> 
</table>

# 重置指定消费组的所有Topic的偏移量
    bin/kafka-consumer-groups.sh --bootstrap-server 10.122.139.40:9092 --reset-offsets --to-earliest --group listenerForSyncEsfCommunity1 --all-topic --dry-run
    
    # 重置指定消费组的指定Topic的偏移量
    bin/kafka-consumer-groups.sh --bootstrap-server 10.122.139.40:9092 --reset-offsets --to-earliest --group listenerForSyncEsfCommunity1 --topic test_topic --dry-run
    
    # 重置所有消费组的所有Topic的偏移量
    bin/kafka-consumer-groups.sh --bootstrap-server 10.122.139.40:9092 --reset-offsets --to-earliest --all-group --all-topic --dry-run 
    
    # 重置所有消费组的指定Topic的偏移量
    bin/kafka-consumer-groups.sh --bootstrap-server 10.122.139.40:9092 --reset-offsets --to-earliest --all-group --topic test_topic --dry-run
    
    
    
    # 只有这个消费组不可用状态才能重置成功，否则会报下面异常：
    Error: Assignments can only be reset if the group 'listenerForSyncEsfCommunity1' is inactive, but the current state is Stable.
    
    TOPIC                          PARTITION  NEW-OFFSET

## delete-offsets：删除消费组的偏移量 ##

# 删除指定消费组的指定Topic的偏移量（下一次从头消费）
    bin/kafka-consumer-groups.sh --bootstrap-server 10.122.139.40:9092 --delete-offsets --group listenerForSyncEsfCommunity1 --topic test_topic
    
    # 只有这个消费组不可用状态才能删除成功，否则会报异常

[Java]: https://blog.csdn.net/weixin_43453386/article/details/84788317
[Mysql]: https://blog.csdn.net/weixin_43453386/article/details/88667709
[Redis]: https://blog.csdn.net/weixin_43453386/article/details/127966762
[Kafka]: https://blog.csdn.net/weixin_43453386/article/details/128189472
[Spring]: https://blog.csdn.net/weixin_43453386/article/details/124900806
[SpringBoot]: https://blog.csdn.net/weixin_43453386/article/details/84788714
[MyBatis_MyBatis-Plus]: https://blog.csdn.net/weixin_43453386/article/details/84788053