首页 > 代码库 > kafka 并发数配置过程中踩到的坑 InstanceAlreadyExistsException

kafka 并发数配置过程中踩到的坑 InstanceAlreadyExistsException

2017-07-05 13:09:15.460 [kafka_spout:7-MultipleThreadSpoutExecutors] WARN o.a.kafka.common.utils.AppInfoParser - Error registering AppInfo mbean

javax.management.InstanceAlreadyExistsException: kafka.consumer:type=app-info,id=gx-test-20170629
at com.sun.jmx.mbeanserver.Repository.addMBean(Repository.java:437)
at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerWithRepository(DefaultMBeanServerInterceptor.java:1898)
at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerDynamicMBean(DefaultMBeanServerInterceptor.java:966)
at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerObject(DefaultMBeanServerInterceptor.java:900)
at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerMBean(DefaultMBeanServerInterceptor.java:324)
at com.sun.jmx.mbeanserver.JmxMBeanServer.registerMBean(JmxMBeanServer.java:522)
at org.apache.kafka.common.utils.AppInfoParser.registerAppInfo(AppInfoParser.java:57)
at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:640)
at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:512)
at org.springframework.kafka.core.DefaultKafkaConsumerFactory.createKafkaConsumer(DefaultKafkaConsumerFactory.java:73)
at org.springframework.kafka.core.DefaultKafkaConsumerFactory.createConsumer(DefaultKafkaConsumerFactory.java:69)
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.<init>(KafkaMessageListenerContainer.java:284)
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.<init>(KafkaMessageListenerContainer.java:222)
at org.springframework.kafka.listener.KafkaMessageListenerContainer.doStart(KafkaMessageListenerContainer.java:179)
at org.springframework.kafka.listener.AbstractMessageListenerContainer.start(AbstractMessageListenerContainer.java:204)
at org.springframework.kafka.listener.ConcurrentMessageListenerContainer.doStart(ConcurrentMessageListenerContainer.java:126)
at org.springframework.kafka.listener.AbstractMessageListenerContainer.start(AbstractMessageListenerContainer.java:204)
at com.mapbar.stream.qingqi.core.spout.KafkaSpout.open(KafkaSpout.java:74)
at com.alibaba.jstorm.task.execute.spout.SpoutExecutors.init(SpoutExecutors.java:142)
at com.alibaba.jstorm.task.execute.spout.MultipleThreadSpoutExecutors.init(MultipleThreadSpoutExecutors.java:64)
at com.alibaba.jstorm.task.execute.BaseExecutors.initWrapper(BaseExecutors.java:154)
at com.alibaba.jstorm.task.execute.spout.MultipleThreadSpoutExecutors.run(MultipleThreadSpoutExecutors.java:76)
at com.alibaba.jstorm.callback.AsyncLoopRunnable.run(AsyncLoopRunnable.java:95)
at java.lang.Thread.run(Thread.java:745)

上面是本人在使用spring kafka中所遇到的问题,针对此问题做一个记录,整理到此处

出现上述问题的原因:

ConcurrentMessageListenerContainer factory = new ConcurrentMessageListenerContainer(cf, containerProps);
factory.setConcurrency(kafkaConfig.getConcurrencySize());
如果使用了ConcurrentMessageListenerContainer 的实现,并且配置了并发度大于1,同时配置了kafka的 client.id属性则会出现上述问题,而当你配置为1的时候不会出现上述log
解决方式:不配置client.id这一项,kakfa中会默认为多个线程生成id
详细解析:
org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:640)
从error log中可以看出

技术分享

       在调用AppInfoParser.registerAppInfo方法时出现的异常

  

at com.sun.jmx.mbeanserver.Repository.addMBean(Repository.java:437)
at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerWithRepository(DefaultMBeanServerInterceptor.java:1898)
at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerDynamicMBean(DefaultMBeanServerInterceptor.java:966)
at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerObject(DefaultMBeanServerInterceptor.java:900)
at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerMBean(DefaultMBeanServerInterceptor.java:324)
at com.sun.jmx.mbeanserver.JmxMBeanServer.registerMBean(JmxMBeanServer.java:522)

  从堆栈信息中定位到 Repository.addMBean()

截取其中一部分代码

技术分享

上图中可以得知,一个clientId对应一个,不可重复

最后找到KafkaConsumer,发现是因为自己配置了client.id导致的,如果不配置的话

技术分享

会为每一个线程生成一个clientid,"consumer" +  自增id,原子性递增

看到这里,发现了一个问题,如果自己不配置client.id的话,那从config里取出来的数据直接判断的length,猜测是有默认配置,当自己不配置的时候给赋值为"" 空串,于是又查了一下代码来验证

创建consumer对象的时候,会创建ConsumerConfig 配置,new ConsumerConfig()会调用父类构造方法
ConsumerConfig(Map<?, ?> props) {
super(CONFIG, props);
}
AbstractConfig

技术分享

 

ConsumerConfig中有static静态初始化块,来初始化 ConfigDef

技术分享

把client.id赋值为""



kafka 并发数配置过程中踩到的坑 InstanceAlreadyExistsException