title: How can I create Multiconsumer for single queue on apache kafka?
body: <p>Hi my scenario is having a queue that a lot of resources put messages into that queue and a lot of consumers read messages and act specific job.</p>
  
  <p>for this scenario I create a topic in Kafka  with this command</p>
  
  <pre><code>bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test1
  </code></pre>
  
  <p>now I develop a java class to consume it </p>
  
  <pre><code>public class ConsumerGroupExample {
  
      private final ConsumerConnector consumer;
      private final String topic;
      private  ExecutorService executor;
  
      public ConsumerGroupExample(String a_zookeeper, String a_groupId, String a_topic) {
          consumer = kafka.consumer.Consumer.createJavaConsumerConnector(
                  createConsumerConfig(a_zookeeper, a_groupId));
          this.topic = a_topic;
      }
  
      public void shutdown() {
          if (consumer != null) consumer.shutdown();
          if (executor != null) executor.shutdown();
          try {
              if (!executor.awaitTermination(5000, TimeUnit.MILLISECONDS)) {
                  System.out.println("Timed out waiting for consumer threads to shut down, exiting uncleanly");
              }
          } catch (InterruptedException e) {
              System.out.println("Interrupted during shutdown, exiting uncleanly");
          }
     }
  
      public void run(int a_numThreads) {
          Map&lt;String, Integer&gt; topicCountMap = new HashMap&lt;String, Integer&gt;();
          topicCountMap.put(topic, new Integer(a_numThreads));
          Map&lt;String, List&lt;KafkaStream&lt;byte[], byte[]&gt;&gt;&gt; consumerMap = consumer.createMessageStreams(topicCountMap);
          List&lt;KafkaStream&lt;byte[], byte[]&gt;&gt; streams = consumerMap.get(topic);
  
          // now launch all the threads
          //
          executor = Executors.newFixedThreadPool(a_numThreads);
  
          // now create an object to consume the messages
          //
          int threadNumber = 0;
          for (final KafkaStream stream : streams) {
              executor.submit(new ConsumerTest(stream, threadNumber));
              threadNumber++;
          }
      }
  
      private static ConsumerConfig createConsumerConfig(String a_zookeeper, String a_groupId) {
          Properties props = new Properties();
          props.put("zookeeper.connect", a_zookeeper);
          props.put("zookeeper.session.timeout.ms", "400");
          props.put("zookeeper.sync.time.ms", "200");
  
          props.put("group.id", a_groupId);
          props.put("num.consumer.fetchers", "2");
          props.put("partition.assignment.strategy", "roundrobin");
          props.put("auto.commit.interval.ms", "1000");
  
          return new ConsumerConfig(props);
      }
  
      public static void main(String[] args) {
  
          String zooKeeper = "tls.navaco.local:2181";
          String groupId = "group1";
          String topic = "test1";
          int threads = 4;
  
          ConsumerGroupExample example = new ConsumerGroupExample(zooKeeper, groupId, topic);
          example.run(threads);
  
          while (true) {}
  
      }
  </code></pre>
  
  <p>and another java class for produce messages</p>
  
  <pre><code>public class TestProducer {
  
      public static void main(String[] args) {
          Properties props = new Properties();
  
          props.put("metadata.broker.list", "tls.navaco.local:9092");
          props.put("serializer.class", "kafka.serializer.StringEncoder");
          props.put("request.required.acks", "1");
          ProducerConfig config = new ProducerConfig(props);
  
          Producer p = new Producer&lt;String, String&gt;(config);
  
          //sending...
          String topic = "test1";
          String message = "Hello Kafka";
          for (int i = 0; i &lt; 1000; i++) {
              KeyedMessage&lt;String, String&gt; keyedMessage = new KeyedMessage&lt;String, String&gt;(topic, message + i);
              p.send(keyedMessage);           
          }
      }
  
  }
  </code></pre>
  
  <p>, and as the Apache document said if the topic wants to act as a queue, the consumer should have the same group.id, and I did it, but when I run 2, 3 or event more consumers, just one of them get the messages and the others didn't do anything.</p>
  
  <p>In fact I want a queue, and the ordering it's not important for me, the thing that important for me is each message just and just consume by one consumer.</p>
  
  <p>I wonder that is it possible to implement it in Kafka or Should I use another product such as ActiveMQ, HornetMQ, .....?</p>