我正在尝试用kafka实现一个多消费者-单个生产者的设置。我创建了一个包含3个分区的主题，如下所示：

./kafka-create-topic.sh --topic stream.main.out --zookeeper 
localhost:2181 --partition 3

生产商设置如下：

props = new Properties();
     props.put("serializer.class", "kafka.serializer.StringEncoder");
     props.put("broker.list","xxx.xxx.x:9092,.....");
     props.put("request.required.acks", "1");
     props.put("topic.metadata.refresh.interval.ms", "1");
     props.put("partitioner.class", "kafka.producer.DefaultPartitioner");
    //props.put("enable.auto.cimmit","true");
    //props.put("auto.commit.interval.ms","10");

现在，如果我启动第一个消费者，它运行良好；启动第二个消费者是问题所在。它启动了，但似乎是等待消息，没有发生任何事情超过10分钟。为什么会这样？起初我尝试在没有多个分区的情况下执行此操作，但后来每个使用者似乎都得到了相同的消息，即使使用相同的组id。

props = new Properties();
        props.put("metadata.broker.list", "192.xxxx:9092,....");
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        props.put("broker.list","192.1xxx:9092,192....");
        props.put("request.required.acks", "1");
        ConsumerConnector consumer = 
        kafka.consumer.Consumer.createJavaConsumerConnector(
            createConsumerConfig(zooKeeper, groupId));

    Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
    topicCountMap.put(topic, new Integer(1));
    Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);
    List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(topic);

    for ( KafkaStream stream : streams) {
         ConsumerIterator<byte[], byte[]> it = stream.iterator();
         while (it.hasNext())