Kafka消费者-Java客户端

Question

我在kafka消费者文档中看到了这个注释-

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

事实上，您正在执行a_numThreads=50和executors.newfixedthreadpool(a_numThreads)；是的，这意味着您不能在任何时间点创建超过50个线程，至少不能使用该执行器。

文档所说的是，一个分区只能分配给1个流，如果您不创建50个流，而是创建51个流，后者将一无所获，正如这里所解释的那样