首页 文章

kafka使用者不接收java中的消息

提问于
浏览 次
1

在我的代码中，消费者订阅现有主题，但没有收到来自主题的消息请帮助我，它等待消息，而在kafka控制台消费者消息收到正确

package kafka;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.Properties;

public class SimpleConsumer {
    public static void main(String[] args) throws Exception {

        //Kafka consumer configuration settings
        String topicName = "test12";
        Properties props = new Properties();

        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "test");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("session.timeout.ms", "30000");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        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
            <String, String>(props);

        //Kafka Consumer subscribes list of topics here.
        consumer.subscribe(Arrays.asList(topicName));

        //print the topic name
        System.out.println("Subscribed to topic " + topicName);
        int i = 0;

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(1000);
            for (ConsumerRecord<String, String> record : records)

                // print the offset,key and value for the consumer records.
                System.out.printf("offset = %d, key = %s, value = %s\n",
                    record.offset(), record.key(), record.value());
        }
    }
}
java apache-kafka kafka-consumer-api

1 回答

  • 1

    可能是偏移位于最新位置的情况 . 你可以尝试从头开始 .

    HashSet<TopicPartition> partitions = new HashSet<TopicPartition>();
        for (TopicPartition partition : partitions) {
            long offset = consumer.position(partition);
            System.out.println(partition.partition() + ": " + offset);
        }
        consumer.seekToBeginning(partitions);
    回复于

相关问题