2016-11-30 419 views
0

我正在与Kafka合作,并尝试通过关注此article来设置消费者群组。唯一的区别是我创建了自己的抽象类,处理器使设计更简单。来自kafka消费者的InstanceAlreadyExistsException

下面是我的抽象类:

public abstract class Consumer implements Runnable { 
    private final Properties consumerProps; 
    private final String consumerName; 

    public Consumer(String consumerName, Properties consumerProps) { 
    this.consumerName = consumerName; 
    this.consumerProps = consumerProps; 
    } 

    protected abstract void shutdown(); 

    protected abstract void run(String consumerName, Properties consumerProps); 

    @Override 
    public final void run() { 
    run(consumerName, consumerProps); 
    } 
} 

下面是我KafkaConsumerA延伸上面抽象类:

public class KafkaConsumerA extends Consumer { 
    private KafkaConsumer<byte[], DataHolder> consumer; 

    public KafkaConsumerA(String consumerName, Properties consumerProps) { 
    super(consumerName, consumerProps); 
    } 

    @Override 
    public void shutdown() { 
    consumer.wakeup(); 
    } 

    @Override 
    protected void run(String consumerName, Properties consumerProps) { 
    // exception comes from below line from two of the threads and the remaining one thread works fine. 
    consumer = new KafkaConsumer<>(consumerProps); 
    List<String> topics = getTopicsBasisOnConsumerName(consumerName); 
    try { 
     consumer.subscribe(topics); 
     // Setup the schema config 
     Map<String, Object> config = new HashMap<>(); 
     config.put("urls", "https://abc.qa.host.com"); 

     GenericRecordDomainDataDecoder decoder = new GenericRecordDomainDataDecoder(config); 
     while (true) { 
     ConsumerRecords<byte[], DataHolder> records = consumer.poll(200); 
     for (ConsumerRecord<byte[], DataHolder> record : records) { 
      Map<String, Object> data = new HashMap<>(); 
      data.put("partition", record.partition()); 
      data.put("offset", record.offset()); 
      data.put("value", record.value()); 
      System.out 
       .println((Thread.currentThread().getId() % 3) + 1 + ": " + decoder.decode(record.value())); 
     } 
     } 
    } catch (WakeupException ex) { 
     ex.printStackTrace(); 
    } catch (Exception ex) { 
     ex.printStackTrace(); 
    } finally { 
     consumer.close(); 
    } 
    } 
} 

,下面将我的处理程序类:

// looks like something is wrong in this class 
public final class ConsumerHandler { 
    private final ExecutorService executorServiceProcess; 
    private final Consumer consumer; 
    private final List<Consumer> consumers = new ArrayList<>(); 

    public ConsumerHandler(Consumer consumer, int poolSize) { 
    this.executorServiceProcess = Executors.newFixedThreadPool(poolSize); 
    this.consumer = consumer; 
    for (int i = 0; i < poolSize; i++) { 
     consumers.add(consumer); 
     executorServiceProcess.submit(consumer); 
    } 
    } 

    public void shutdown() { 
    Runtime.getRuntime().addShutdownHook(new Thread() { 
     @Override 
     public void run() { 
     for (Consumer consumer : consumers) { 
      consumer.shutdown(); 
     } 
     executorServiceProcess.shutdown(); 
     try { 
      executorServiceProcess.awaitTermination(1000, TimeUnit.MILLISECONDS); 
     } catch (InterruptedException ex) { 
      Thread.currentThread().interrupt(); 
     } 
     } 
    }); 
    } 
} 

这里我是从主类的消费者群体中启动我的所有消费者:

public static void main(String[] args) { 
    ConsumerHandler handlerA = 
     new ConsumerHandler(new KafkaConsumerA("KafkaConsumerA", getConsumerProps()), 3); 
    // run KafkaConsumerB here 

    handlerA.shutdown(); 
    // shutdown KafkaConsumerB here 
    } 

所以用这个 - 我的计划是建立一个消费群体有三个消费者KafkaConsumerA和所有三个订阅相同的主题。

错误: -

每当我跑这个,貌似只有一个消费群体在消费工作和其他两名不起作用。我从控制台上看到这个例外:

javax.management.InstanceAlreadyExistsException: kafka.consumer:type=app-info,id=a97716e0-0e05-4938-8fa1-6b872cf24e34 
    at com.sun.jmx.mbeanserver.Repository.addMBean(Repository.java:437) ~[na:1.7.0_79] 
    at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerWithRepository(DefaultMBeanServerInterceptor.java:1898) ~[na:1.7.0_79] 
    at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerDynamicMBean(DefaultMBeanServerInterceptor.java:966) ~[na:1.7.0_79] 
    at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerObject(DefaultMBeanServerInterceptor.java:900) ~[na:1.7.0_79] 
    at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerMBean(DefaultMBeanServerInterceptor.java:324) ~[na:1.7.0_79] 
    at com.sun.jmx.mbeanserver.JmxMBeanServer.registerMBean(JmxMBeanServer.java:522) ~[na:1.7.0_79] 
    at org.apache.kafka.common.utils.AppInfoParser.registerAppInfo(AppInfoParser.java:58) ~[kafka-clients-0.10.0.0-SASL.jar:na] 
    at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:694) [kafka-clients-0.10.0.0-SASL.jar:na] 
    at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:587) [kafka-clients-0.10.0.0-SASL.jar:na] 
    at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:569) [kafka-clients-0.10.0.0-SASL.jar:na] 

我在这里做什么? getConsumerProps()方法返回属性对象,其中client.idgroup.id与该消费者组中的所有三个消费者具有相同的值。

下面是我的设计细节:

  • KafkaConsumerA将有3名消费者在消费群体和每一个消费者将在topicA工作。
  • 我的KafkaConsumerB(与KafkaConsumerA类似)将有两个消费者在不同的消费群体中,并且这些消费者中的每一个都将在topicB上工作。

而这两个消费者KafkaConsumerAKafkaConsumerB将运行在相互独立的不同消费者组的相同盒子上。

回答

3

Kafka正在尝试注册MBeans进行应用程序监视,并且正在使用client.id这样做。正如您所说,您的抽象类中注入了属性,并为每个消费者注入相同的client.idgroup.idA。但是,你有不同的客户,所以你应该给他们自己client.id,但保持相同group.id。这会将不同的客户/消费者注册到相同的消费者组中,并使它们一起工作,但不会与MBeans注册发生冲突。

+0

我明白了。谢谢(你的)信息。我会试试看。另外,你认为我的设计和逻辑是正确的,我正在尝试做什么。我在我的问题的底部提到了它。或者你看到任何可以完成的设计改进? – user1950349

+0

这是一个非常合理的设置,但我当然不知道你的域名。 –

+1

我的想法是在同一个JVM上运行多个kafka使用者。每个卡夫卡消费者都有多个线程。例如:'KafkaConsumerA'将有三个使用者在一个使用topicA的消费群中。 'KafkaConsumerB'将有2个消费者在另一个使用'topicB'的消费群体中。这两个'KafkaConsumerA'和'KafkaConsumerB'都将在单个JVM上运行。 – user1950349

相关问题