我正在与Kafka合作，并试图通过本文建立消费者群体。唯一的区别是我已经创建了自己的抽象类，处理程序来简化设计。
下面是我的抽象课：

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 三人的主题都一样。
error:-
每当我运行这个，看起来只有一个消费者在消费群体的工作和其他两个不工作。我在控制台上看到了这两个例外：

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.id 以及 group.id 对该消费群体中的所有三个消费者具有相同的价值。
以下是我的设计细节：
我的 KafkaConsumerA 将有三个消费者在一个消费群体中，每个消费者将致力于 topicA .
我的 KafkaConsumerB （类似于kafkaconsumera）将有两个消费者在一个不同的消费群体中，并且每个消费者都将致力于 topicB .
而这两个消费者 KafkaConsumerA 以及 KafkaConsumerB 将在同一个盒子上运行，不同的消费群体相互独立。