Kafka权威指南

Kafka

消费者——从

Kafka

读取数据

｜

67

...

try {

// 循环，直到按下Ctrl+C键，关闭的钩子会在退出时进行清理

while (true) {

ConsumerRecords<String, String> records =

movingAvg.consumer.poll(1000);

System.out.println(System.currentTimeMillis() + "

-- waiting for data...");

for (ConsumerRecord<String, String> record :

records) {

System.out.printf("offset = %d, key = %s,

value = %s\n",

record.offset(), record.key(),

record.value());

}

for (TopicPartition tp: consumer.assignment())

System.out.println("Committing offset at

position:" +

consumer.position(tp));

movingAvg.consumer.commitSync();

}

} catch (WakeupException e) {

// 忽略关闭异常

➋

} finally {

consumer.close(); ...