我有一个简单的 java 生产者,如下所示
public class Producer
{
private final static String TOPIC = "my-example-topi8";
private final static String BOOTSTRAP_SERVERS = "localhost:8092";
public static void main( String[] args ) throws Exception {
Producer<String, byte[]> producer = createProducer();
for(int i=0;i<3000;i++) {
String msg = "Test Message-" + i;
final ProducerRecord<String, byte[]> record = new ProducerRecord<String, byte[]>(TOPIC, "key" + i, msg.getBytes());
producer.send(record).get();
System.out.println("Sent message " + msg);
}
producer.close();
}
private static Producer<String, byte[]> createProducer() {
Properties props = new Properties();
props.put("metadata.broker.list", BOOTSTRAP_SERVERS);
props.put("bootstrap.servers", BOOTSTRAP_SERVERS);
props.put("client.id", "AppFromJava");
props.put("serializer.class", "kafka.serializer.DefaultEncoder");
props.put("key.serializer.class", "kafka.serializer.StringEncoder");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("compression.codec", "snappy");
props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
return new KafkaProducer<String, byte[]>(props);
}
}
我正在尝试读取如下数据
public class Consumer
{
private final static String TOPIC = "my-example-topi8";
private final static String BOOTSTRAP_SERVERS = "localhost:8092";
public static void main( String[] args ) throws Exception {
Consumer<String, byte[]> consumer = createConsumer();
start(consumer);
}
static void start(Consumer<String, byte[]> consumer) throws InterruptedException {
final int giveUp = 10;
int noRecordsCount = 0;
int stopCount = 1000;
while (true) {
final ConsumerRecords<String, byte[]> consumerRecords = consumer.poll(1000);
if (consumerRecords.count()==0) {
noRecordsCount++;
if (noRecordsCount > giveUp) break;
else continue;
}
consumerRecords.forEach(record -> {
// Process the record System.out.printf("\nConsumer Record:(%s, %s, %s)", record.key(), new String(record.value()), record.topic());
});
consumer.commitSync();
break;
}
consumer.close();
System.out.println("DONE");
}
private static Consumer<String, byte[]> createConsumer() {
final Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
BOOTSTRAP_SERVERS);
props.put(ConsumerConfig.GROUP_ID_CONFIG,
"KafkaExampleConsumer");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
ByteArrayDeserializer.class.getName());
props.put(ConsumerConfig.CLIENT_ID_CONFIG, "1234");
props.put("enable.auto.commit", "false");
// Create the consumer using props.
final Consumer<String, byte[]> consumer = new KafkaConsumer(props);
consumer.subscribe(Collections.singletonList(TOPIC));
return consumer;
}
}
但消费者没有阅读来自 kafka 的任何消息。如果我在最开始添加以下内容start()
consumer.poll(0);
consumer.seekToBeginning(consumer.assignment());
然后消费者开始阅读该主题。但是每次消费者重新启动时,它都会从我不想要的主题开头读取消息。如果我在启动 Consumer 时添加以下配置
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
然后它从主题读取消息,但如果消费者在处理所有消息之前重新启动,那么它不会读取未处理的消息。
有人可以让我知道出了什么问题以及如何解决这个问题吗?
Kafka Broker 和 Zookeeper 使用默认配置运行。