[大数据]4.1 Consumer示例

1.代码示例

public class ConsumerSample {
    private static final String topicName = "steven";

    public static void main(String[] args) {
        //Consumer配置
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "127.0.0.1:9092");
        props.setProperty("group.id", "test");
        props.setProperty("enable.auto.commit", "true");
        props.setProperty("auto.commit.interval.ms", "1000");
        props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        //Consumer的主对象
        KafkaConsumer<String, String> consumer = new KafkaConsumer(props);
        //消费哪一个topic或者几个topic
        consumer.subscribe(Arrays.asList(topicName));
        while (true) {
            //批量拉取消息
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("partition = %d,offset = %d,key = %s,value = %s%n", record.partition(), record.offset(), record.key(), record.value());
            }
        }
    }
}

2.代码运行结果
(1).控制台输出
运行异步发送示例代码向Kafka发送10条数据，然后再执行ConsumerSample代码进行消费，便可在控制台看见代码中指定的输出信息。停止ConsumerSample后，再次运行ConsumerSample，发现依然会输出以下内容。为了解决重复消费的问题，便需要手动通知offset提交。

partition = 0,offset = 0,key = key-0,value = value-0
partition = 0,offset = 1,key = key-1,value = value-1
partition = 0,offset = 2,key = key-2,value = value-2
partition = 0,offset = 3,key = key-3,value = value-3
partition = 0,offset = 4,key = key-4,value = value-4
partition = 0,offset = 5,key = key-5,value = value-5
partition = 0,offset = 6,key = key-6,value = value-6
partition = 0,offset = 7,key = key-7,value = value-7
partition = 0,offset = 8,key = key-8,value = value-8
partition = 0,offset = 9,key = key-9,value = value-9

(2).终端输出
打开一个cmd终端，在E:\Kafka\kafka_2.12-1.1.0\bin\windows目录下执行kafka-console-consumer.bat --zookeeper localhost:2181 --topic steven，可以看到生产的消息。