完成迭代后，调用不会从 while 循环中出来

Question

我有这个 Apache kafka 消费者的示例代码

public class WorkerThread implements Callable<Object> {
    ConsumerConnector consumerConnector;
    private static ArrayList<Object> list = new ArrayList<>();

    @Override
    public Object call() throws Exception {
        processMsg();
        return list;
    }

    private void processMsg() {
        StringMessage msg = new StringMessage();
        Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
        topicCountMap.put(KafkaConstants.LOB_TOPIC, new Integer(1));
        Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumerConnector
                .createMessageStreams(topicCountMap);
        KafkaStream<byte[], byte[]> stream = consumerMap.get(KafkaConstants.LOB_TOPIC).get(0);
        ConsumerIterator<byte[], byte[]> itr = stream.iterator();

            while (itr.hasNext()) {
                msg.setMessage(new String((itr.next().message())));
                list.add(msg);
                System.out.println("&&&" + msg);
            }

        System.out.println("++++++++++++");
    }
}

在上面的示例中，我的"++++++" sysout 无法正常工作。任何人都知道原因..？

Answer 1

KafkaStream 正在阻塞。

见

http://grokbase.com/t/kafka/users/131nb67cvk/non-blocking-kafka-stream-iterators