- 论坛徽章:
- 0
|
- Properties props = new Properties();
- props.put("bootstrap.servers", bootstrapServers);
- props.put("group.id", groupId);
- props.put("enable.auto.commit", false);
- props.put("auto.commit.interval.ms", "1000");
- props.put("session.timeout.ms", "30000");
- props.put("max.poll.records", maxPollRecords);
- props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
- props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
- KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
- //consumer.subscribe(Arrays.asList(topic));
- TopicPartition p = new TopicPartition(topic,2);
- consumer.assign(Arrays.asList(p));
- consumer.seek(p,488430);
- //while (true) {
- ConsumerRecords<String, String> records = consumer.poll(100);
- for (ConsumerRecord<String, String> record : records) {
- String V = record.value();
- }
复制代码
查不出数据recods没有内容
|
|