假设你有：

String topic1 = "channel1";
String topic2 = "channel2";
String topic3 = "channel3";

那么，你确实可以做到：

topicCountMap.put(topic1, new Integer(a_numThreads_topic1));
topicCountMap.put(topic2, new Integer(a_numThreads_topic2));
topicCountMap.put(topic3, new Integer(a_numThreads_topic3));

一旦您获得了consumermap（不会更改的代码），您将能够检索每个主题的流：

List<KafkaStream<byte[], byte[]>> topic1_streams = consumerMap.get(topic1);
List<KafkaStream<byte[], byte[]>> topic2_streams = consumerMap.get(topic2);
List<KafkaStream<byte[], byte[]>> topic3_streams = consumerMap.get(topic3);

要从流中消费，您需要创建适当数量的执行器：

executors_topic1 = Executors.newFixedThreadPool(a_numThreads_topic1);
executors_topic2 = Executors.newFixedThreadPool(a_numThreads_topic2);
executors_topic3 = Executors.newFixedThreadPool(a_numThreads_topic3);

最后：

int threadNumber = 0;
for (final KafkaStream stream : topic1_streams) {
    executors_topic1.submit(new ConsumerTest(streams, threadNumber));
    threadNumber++;
}
for (final KafkaStream stream : topic2_streams) {
    executors_topic2.submit(new ConsumerTest(stream, threadNumber));
    threadNumber++;
}
for (final KafkaStream stream : topic3_streams) {
    executor_topic3.submit(new ConsumerTest(stream, threadNumber));
    threadNumber++;
}

当然，这只是给你一个想法。显然，代码可以改进。