我是 Apache Kafka 的新手,正在探索 SimpleConsumer 以读取来自该主题的消息。
我使用下面的代码来做同样的事情,
FetchRequestBuilder builder = new FetchRequestBuilder();
FetchRequest fetchRequest = builder.addFetch(topic, partitionId, offset, 1024).build();
FetchResponse fetchResponse;
try {
fetchResponse = consumer.fetch(fetchRequest);
} catch (Exception e) {}
这会读取特定分区中的所有可用消息;我想设置要阅读的最大消息数。在这个阶段有没有办法做到这一点?当队列中有大量消息时,我不希望所有消息都落在 JVM 堆中。
另一个问题,
以下代码返回一个 ByteBufferMessageSet。
fetchResponse.messageSet(topic, partitionId);
这是否意味着,并非所有可用消息都实际进入内存?