时间戳

首页 > 解决方案 > 为什么我的 Java 使用者不能读取我创建的数据？

java - 为什么我的 Java 使用者不能读取我创建的数据？

问题描述

我正在尝试从我制作的简单生产者那里读取数据。出于某种原因，每当我运行消费者时，它都看不到/产生我产生的任何数据。任何人都可以给我关于下一步做什么的指导吗？

我在下面包含了我的生产者和消费者的代码：

制片人：

public class AvroProducer {

public static void main(String[] args) {

    String bootstrapServers = "localhost:9092";
    String topic = "trackingReportsReceived";

    //create Producer properties
    Properties properties = new Properties();
    properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class.getName());
    properties.setProperty("schema.registry.url", "http://localhost:8081");

    //create the producer
    KafkaProducer<String, trackingReport> producer = new KafkaProducer<>(properties);

    //creating my own event
    trackingReport event = trackingReport.newBuilder()
            .setRemoteEventUID(2)
            .setReceivedPacketUID(2)
            .setRemoteUnitAID(2)
            .setEventTime(2)
            .setEventLocationStampUID(3)
            .setEventLatitude(2)
            .setEventLongitude(2)
            .setEventOdometer(3)
            .setEventSpeed(3)
            .setEventCourse(3)
            .build();


    //create a producer record
    ProducerRecord<String, trackingReport> eventRecord = new ProducerRecord<>(topic, event);

    //send data - asynchronous
    producer.send(eventRecord, new Callback() {
        @Override
        public void onCompletion(RecordMetadata recordMetadata, Exception e) {
            if (e == null) {
                    System.out.println("Success!");
                    System.out.println(recordMetadata.toString());
                } else {
                    e.printStackTrace();
                }
            }
        });


    //flush data
    producer.flush();
    //flush and close producer
    producer.close();

消费者：

public class AvroConsumer {

public static void main(String[] args) {

    final Logger logger = LoggerFactory.getLogger(AvroConsumer.class);

    String bootstrapServers = "localhost:9092";

    //create Consumer properties
    Properties properties = new Properties();
    properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    properties.put(ConsumerConfig.GROUP_ID_CONFIG, "consumer");
    properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
    properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

//        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
    properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class.getName());
    properties.put("schema.registry.url", "http://localhost:8081");
    properties.put("specific.avro.reader", "true");

    //create the consumer
    KafkaConsumer<String, trackingReport> consumer = new KafkaConsumer<>(properties);
    String topic = "trackingReportsReceived";

    consumer.subscribe(Collections.singletonList(topic));

    System.out.println("Waiting for data...");

//        try {

        while (true) {
            ConsumerRecords<String, trackingReport> records = consumer.poll(100);
            for (ConsumerRecord<String, trackingReport> record : records) {
                trackingReport trackingrep = record.value();
                System.out.println(trackingrep);
            }
            consumer.commitSync();
        }

//        } catch (Exception e) {
//            logger.error("Exception occured while consuming messages...", e);
//        } finally {
//            consumer.close();
//        }


}
}

注意生产者工作，但消费者不工作。

标签： javaapache-kafkakafka-consumer-api

解决方案

使用控制台使用者脚本，您是否使用了与您的 java 使用者中相同的组 ID？

如果您这样做是为了验证，请尝试在您的代码中使用新的消费者组。

如果它有效，则意味着控制台使用者读取主题中的所有数据，因此具有此组 id 的使用者提交最后的当前偏移量，当您启动具有相同组 id 的 java 使用者时，他尝试读取该偏移量是最后一个.. 所以没有要阅读的消息。

为了验证您是否也可以先启动 java 使用者，然后再启动生产者，如果您看到消息，那么控制台和 java 使用者具有相同的组 ID。

推荐阅读