卡夫卡消费者：受控阅读主题

Question

我有以下 kafka 消费者代码，其中 3 个线程正在从具有 3 个分区的 kafka 主题中读取。

有什么办法，只有在线程当前正在处理的消息被处理后，才会从kafka主题中读取新消息。

例如，假设主题中有 100 条消息，那么有什么方法可以一次读取并处理 3 条消息。现在，当这 3 条消息被处理后，则只应读取接下来的 3 条消息，依此类推。

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(3);

// 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

如果 ConsumerTest 内部的迭代器正在同步处理消息，那么一次只会消耗 3 条消息。 enable.auto.commit 默认为 true。确保不要将其设置为 false，否则需要添加提交偏移量的逻辑。

前-

 ConsumerIterator<byte[], byte[]> streamIterator= stream.iterator(); 
 while (streamIterator.hasNext()) { 
   String kafkaMsg= new String(streamIterator.next().message()); 
 } 

Answer 2

好吧，默认情况下，消费者彼此不了解，因此无法“同步”他们的工作。您可以做的是将您的三条消息包装成一条消息（从而保证它们都将按顺序得到答复）或者可能引入更多（“子”）主题。

另一种可能性（如果您确实需要保证您的三个消息将被单个消费者消费）可能是您的所有消费者同步他们的工作，或者可能通知一个跟踪您的工作的控制器。

但是感觉就像你“做错了”，实际上队列中的消息是无状态的，只有它们在主题中的顺序决定了它们的“处理顺序”。处理消息的时间无关紧要。