Kafka0.8.2消费者

Question

我正在Java中实现一个简单的Kafka消费者。代码如下：

public class TestConsumer {

    public static void main(String []a) throws Exception{
        Properties props = new Properties();
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("partition.assignment.strategy", "round-robin");
        props.put("group.id", "test");
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);

        try{
            consumer.subscribe("ay_sparktopic");
            Map<String, ConsumerRecords<String, String>> msg = consumer.poll(100);
            System.out.println(msg);                
        }catch(Exception e){
            System.out.println("Exception");
        }
    }
}

我在网上查看的任何文档都给出了range或roundrobin作为可能的分配策略，据我所知，groupId是一个自定义名称。不确定这里什么是正确的配置值。

Answer 1

看起来您正在尝试使用只有在Kafka0.9+中才有的新的消费者API。要使用旧的API，必须从kafka.javaapi.consumer.*包而不是新的org.apache.kafka.clients.consumer包导入类。

consumer.subscribe和consumer.poll与新的API相关，所以如果您真的想使用旧的API，您需要相应地更改代码。如果希望使用新的消费者API，则需要运行Kafka0.9或更高版本。