当前位置:首页 > java > 正文内容

spring boot 删除kafka没有消费者的消费组

root2个月前 (03-21)java1072

背景: 服务使用k8s 部署,服务逻辑是广播消费,即每个服务(pod)都会起一个消费组
因为每个pod都是一样的,无法固定消费组。同时每次更新或者重启pod时消费组就会增多。
时间旧了消费组就会消息堆积,消费组也会越来越多。

在项目中启动一个定时任务删除无消费者的消费组

直接看代码

package org.osc.scan.codesearch.service.impl;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.ConsumerGroupListing;
import org.apache.kafka.clients.admin.ListConsumerGroupsResult;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.config.SaslConfigs;
import org.osc.scan.codesearch.service.DeleteKafkaGroups;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;

@Slf4j
@Service
public class DeleteKafkaGroupsImpl implements DeleteKafkaGroups {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Value("${spring.kafka.properties.security.protocol}")
    private String protocol;

    @Value("${spring.kafka.properties.sasl.mechanism}")
    private String saslMechanism;

    @Value("${spring.kafka.properties.sasl.jaas.config}")
    private String jaasConfig;

    @Override
    public void deleteGroupIds() {

        Map<String, Object> props = new HashMap<>(1);
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,bootstrapServers);
        props.put(AdminClientConfig.SECURITY_PROTOCOL_CONFIG, protocol);
        props.put(SaslConfigs.SASL_MECHANISM, saslMechanism);
        props.put(SaslConfigs.SASL_JAAS_CONFIG, jaasConfig);
        // 创建一个kafka的admin客户端,执行高级操作
        AdminClient adminClient = AdminClient.create(props);
        // 获取消费组
        ListConsumerGroupsResult listConsumerGroupsResult = adminClient.listConsumerGroups();
        try {
            // 获取所有消费组
            Collection<ConsumerGroupListing> consumerGroupListings = listConsumerGroupsResult.all().get();
            // 遍历消费组
            for (ConsumerGroupListing consumerGroupListing : consumerGroupListings) {
                String groupId = consumerGroupListing.groupId();
                // 根据消费组的前缀(避免删除其他服务正在使用的消费组) 和 消费组没有消费者
                if (groupId.contains("LatchGroup-") && adminClient.describeConsumerGroups(Arrays.asList(groupId)).all().get().get(groupId).members().size() == 0){
                    log.info("找到要删除的groupID:{}", groupId);
                    // 执行删除操作
                    KafkaFuture<Void> resultFuture = adminClient.deleteConsumerGroups(Arrays.asList(groupId)).all();
                    try {
                    // 执行
                        resultFuture.get();
                        log.info("成功删除groupID:{}", groupId);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        adminClient.close();


    }
}


扫描二维码推送至手机访问。

版权声明:本文由一叶知秋发布,如需转载请注明出处。

本文链接:https://zhiqiu.top/?id=258

分享给朋友:
返回列表

上一篇:java pom 引入的依赖中 标签

没有最新的文章了...

相关文章

java-Springboot的几个重要注解@controller、@service、 @repository、@component

1、@controller 控制器(注入服务)用于标注控制层,相当于struts中的action层2、@service 服务(注入dao)用于标注服务层,主要用来进行业务的逻辑处理3、@repository(实现dao访问)用于标注数据访问...

java 处理json 字符串

假设有一个实体类public class User{   private int id;   private String name;&...

pom.xml文件的标签含义

<?xml version="1.0" encoding="UTF-8"?>声明xml的版本<project xmlns="http://maven.apache.org/...

IDEA 设置项目为maven

IDEA 设置项目为maven

1.右键项目,选择Add Frameworks support... 2.选择maven,点击OK即可...

java mybatis Parameter index out of range (5 > number of parameters, which is 4

java mybatis Parameter index out of range (5 > number of parameters, which is 4

该报错在修改mapper的xml之后出现的发现是因为注释的问题导致的在xml中注释已经要谨慎。...

java 小知识总结

StringUtils常用的判断空字符串有两个方法:isEmpty和isBlank】上源码// isEmpty public static boolean isEmpty(String s...