如何使用Kafka Consumer API中的密鑰讀取數據？

Question

我構建使用下面的代碼的消息...

Producer<String, String> producer = new kafka.javaapi.producer.Producer<String, String>(producerConfig); 
KeyedMessage<String, String> keyedMsg = new KeyedMessage<String, String>(topic, "device-420", "{message:'hello world'}");   
producer.send(keyedMsg); 

而且使用下面的代碼塊消費......因爲我在讀數據的話題明智

//Key = topic name, Value = No. of threads for topic 
Map<String, Integer> topicCount = new HashMap<String, Integer>();  
topicCount.put(topic, 1); 

//ConsumerConnector creates the message stream for each topic 
Map<String, List<KafkaStream<byte[], byte[]>>> consumerStreams = consumerConnector.createMessageStreams(topicCount);   

// Get Kafka stream for topic 
List<KafkaStream<byte[], byte[]>> kStreamList = consumerStreams.get(topic); 

// Iterate stream using ConsumerIterator 
for (final KafkaStream<byte[], byte[]> kStreams : kStreamList) { 
    ConsumerIterator<byte[], byte[]> consumerIte = kStreams.iterator();   
    while (consumerIte.hasNext()) {    
     MessageAndMetadata<byte[], byte[]> msg = consumerIte.next();    
     System.out.println(topic.toUpperCase() + ">" 
       + " Partition:" + msg.partition() 
       + " | Key:"+ new String(msg.key()) 
       + " | Offset:" + msg.offset() 
       + " | Message:"+ new String(msg.message())); 
    } 
} 

一切工作正常。所以我想知道在這個例子中有沒有什麼辦法可以使用消息鍵，即device-420來使用數據？

Answer 1

簡答題：沒有。

卡夫卡的最小粒度是分區。您可以編寫一個僅從單個分區讀取的客戶端。但是，分區可以包含多個密鑰，並且需要佔用此分區中包含的所有密鑰。