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没有内容
3.启动服务3.1启动zookeeper启动zk有两种方式,第一种是使用kafka自己带的一个zk。bin/zookeeper-server-start.shconfig/zookeeper.properties&另一种是使用其它的zookeeper,可以位于本机也可以位于其它地址。这种情况需要修改config下面的sercer.properties里面的zookeeper地址。例如zookeeper.connect=10.202.4.179:21813.2启动kafkabin/kafka-server-start.shconfig/server.properties4.创建topicbin/kafka-topics.sh--create--zookeeper10.202.4.179:2181--replication-factor1--partitions1--topictest创建一个名为test的topic,只有一个副本,一个分区。通过list命令查看刚刚创建的topicbin/kafka-topics.sh-list-zookeeper10.202.4.179:21815.启动producer并发送消息启动producerbin/kafka-console-producer.sh--broker-listlocalhost:9092--topictest启动之后就可以发送消息了比如testhelloboy按Ctrl+C退出发送消息6.启动consumerbin/kafka-console-consumer.sh--zookeeper10.202.4.179:2181--topictest--from-beginning启动consumer之后就可以在console中看到producer发送的消息了可以开启两个终端,一个发送消息,一个接受消息。如果这样都不行的话,查看zookeeper进程和kafka的topic,一步步排查原因吧。
答案来源于网络
版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。