专业建设金融行业网站的公司,邱县手机网站建设,深圳高端设计装饰公司,建筑模板价格现在是多少的背景
kafka广播消息的时候为了保证groupId不重复#xff0c;再创建的时间采用前缀时间戳的形式#xff0c;这样可以保证每次启动的时候是创建的新的#xff0c;但是
会出现一个问题#xff1a;就是每次停机或者重启都会新建一个应用实例#xff0c;关闭应用后并不会删除…背景
kafka广播消息的时候为了保证groupId不重复再创建的时间采用前缀时间戳的形式这样可以保证每次启动的时候是创建的新的但是
会出现一个问题就是每次停机或者重启都会新建一个应用实例关闭应用后并不会删除kafka下面的消费组导致消费组越来越多目前
我们有promethes监控kafka消息偏移一直没有消费的消费组就会进行报警
解决思路
既然是没有删除消费组就通过优雅停机应用关闭前采用java的api操作kafka消费组进行删除
代码实现
1编写类实现DisposableBean接口实现destroy方法注意每个项目定义的id会不一样此例子中 id “cfgs-broadcast”
package com.simo.vsim.cfgs.init;import com.alibaba.nacos.api.config.annotation.NacosValue;
import lombok.Data;
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.DeleteConsumerGroupsResult;
import org.apache.kafka.common.KafkaFuture;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.listener.MessageListenerContainer;
import org.springframework.stereotype.Component;import javax.annotation.Resource;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;Data
Component
Slf4j
public class ApplicationListen implements InitializingBean, DisposableBean {Resourceprivate KafkaListenerEndpointRegistry registry;NacosValue(value ${spring.kafka.bootstrap-servers}, autoRefreshed true)private String servers;Overridepublic void destroy() {MessageListenerContainer listenerContainer registry.getListenerContainer(cfgs-broadcast);String groupId listenerContainer.getGroupId();MapString, Object props new HashMap(1);props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,servers);AdminClient adminClient AdminClient.create(props);DeleteConsumerGroupsResult deleteConsumerGroupsResult adminClient.deleteConsumerGroups(Arrays.asList(groupId));KafkaFuture resultFuture deleteConsumerGroupsResult.all();try {resultFuture.get();log.info(kafka关闭消费组groupId);} catch (InterruptedException e) {e.printStackTrace();} catch (Exception e) {e.printStackTrace();}adminClient.close();}Overridepublic void afterPropertiesSet() {}
}2接收kafka广播消息的时候指定容器id,用于第一步通过id进行删除id “cfgs-broadcast”
/*** groupId不一样代表广播模式earliest 可能重复消费latest可能漏消费* param message* param ack*/
KafkaListener(containerFactory manualImmediateListenerContainerFactory , topics {${kafka.topic.cfgs-broadcast}},properties {auto.offset.resetlatest},groupId cfgs-broadcast- #{T(java.lang.System).currentTimeMillis()},idIsGroup false,id cfgs-broadcast)
public void onMessageManualBroadcast(ListObject message, Acknowledgment ack){message.forEach(item - handleMsg(2,item));//直接提交offsetack.acknowledge();
}效果
1正常启动有这个消费组cfgs-broadcast-1696754926097
2重新启动通过日志显示已经删除k8s默认是优雅停机 如果是iead直接关闭下不要一下子点击两下停止点击一次是优雅停机连续点击2次就是kill -9的效果就无法看到效果 