Kafka 消费者 - Java 实现

Question

我遇到了这个link，用于在 Java 中实现 KafkaConsumer。下面列出的代码读取流并处理消息。一旦启动，此代码会继续收听传入的消息吗？如果是这样，它如何继续运行并继续消费消息？

public void run(int a_numThreads) {
    Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
    topicCountMap.put(topic, new Integer(a_numThreads));
    Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);
    List<KafkaStream<byte[], byte[]>> 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++;
    }
}

Answer 1

代码一旦启动就会一直监听传入的消息，因为它使用ThreadPool（线程池中的线程数=a_numThreads）并且线程池中的每个线程都执行一个消费者（ConsumerTest） .

如果您仔细查看ConsumerTest 类，您会发现有一个无限循环。 it.hasNext() 处于阻塞状态（请参阅 ConsumerIterator），因此消费者将始终等待下一条消息：

public void run() {
        ConsumerIterator<byte[], byte[]> it = m_stream.iterator();
        while (it.hasNext())
            System.out.println("Thread " + m_threadNumber + ": " + new String(it.next().message()));
        System.out.println("Shutting down Thread: " + m_threadNumber);
}

话虽如此，如果在指定的时间间隔后没有消息可供消费，则可以使用属性consumer.timeout.ms Throw 设置超时异常，但默认值为 - 1（无超时），因此默认情况下，消费者将一直持续消费消息，直到消费者和执行者关闭（参见示例中的shutdown 方法）。