更新:当我打印数据时,即:
考虑以下代码:
public static long Offset = 0L;
FetchRequest req = new FetchRequest(KafkaProperties.topic, 0, Offset,10485760);
ByteBufferMessageSet messageSet = simpleConsumer.fetch(req);
问题是如何获得最后一个偏移量,并将变量Offset
设置为从Kafka读取下一批数据?
更新:当我打印数据时,即:
for (MessageAndOffset messageAndOffset : messageSet) {
System.out.println(messageAndOffset);
}
输出如下:
MessageAndOffset(message(magic = 1, attributes = 0, crc = 2000130375, payload = java.nio.HeapByteBuffer[pos=0 lim=176 cap=176]),296215)
MessageAndOffset(message(magic = 1, attributes = 0, crc = 956398356, payload = java.nio.HeapByteBuffer[pos=0 lim=196 cap=196]),298144)
....
....
MessageAndOffset(message(magic = 1, attributes = 0, crc = 396743887, payload = java.nio.HeapByteBuffer[pos=0 lim=179 cap=179]),299136)
医生说最后一个数字是偏移
MessageAndOffset(message: Message, offset: Long)
也就是说,在上面的情况下,我最后一次读取的偏移量将是299136
这样的东西有帮助吗?有一点不好,那就是它将永远循环。
long offset = 0;
while (true) {
FetchRequest fetchrequest = new FetchRequest(topicName, 0, offset, 10485760);
ByteBufferMessageSet messages = consumer.fetch(fetchrequest);
for (MessageAndOffset msg : messages) {
System.out.println("consumed: " + Utils.toString(msg.message().payload(), "UTF-8"));
offset = msg.offset();
}
}
同样在0.8 Kafka SimpleConsumer的例子中,他们有一些类似的东西
long numRead = 0;
for (MessageAndOffset messageAndOffset : fetchResponse.messageSet(a_topic, a_partition)) {
long currentOffset = messageAndOffset.offset();
if (currentOffset < readOffset) {
System.out.println("Found an old offset: " + currentOffset + " Expecting: " + readOffset);
continue;
}
readOffset = messageAndOffset.nextOffset();
ByteBuffer payload = messageAndOffset.message().payload();
byte[] bytes = new byte[payload.limit()];
payload.get(bytes);
System.out.println(String.valueOf(messageAndOffset.offset()) + ": " + new String(bytes, "UTF-8"));
numRead++;
a_maxReads--;
}
但他们也提到,应用程序希望a_maxread
(要读取的最大消息数)参数作为参数传递,这样我们就不会永远循环。我是卡夫卡的新手,不确定这是否是你想要的。