消费者没有在Apache Kafka中收到消息-Java 学习之路

我正在构建一个Apache Kafka消费者来订阅另一个已经运行的Kafka . 现在，我的问题是当我的制作人将消息推送到服务器时...我的消费者没有收到它们 . Here I give Producer code,

Properties properties = new Properties();
        properties.put("metadata.broker.list","Running kafka ip addr:9092");
        properties.put("serializer.class","kafka.serializer.StringEncoder");
        ProducerConfig producerConfig = new ProducerConfig(properties);
        kafka.javaapi.producer.Producer<String,String> producer = new kafka.javaapi.producer.Producer<String, String>(producerConfig);
        String filePath="filepath";
        File rootFile= new File(filePath);
        Collection<File> allFiles = FileUtils.listFiles(rootFile, CanReadFileFilter.CAN_READ, TrueFileFilter.INSTANCE);
        for(File file : allFiles) {
            StringBuilder sb = new StringBuilder();
            sb.append(file);
            KeyedMessage<String, String> message =new KeyedMessage<String, String>(TOPIC,sb.toString());
            System.out.println("sending msg from producer.."+sb.toString());
            producer.send(message);
        }
           producer.close();

Here Consumer code,

properties.put("bootstrap.servers","Running zookeaper ip addr:2181");
         properties.put("group.id","test-group");
         properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
         properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
         properties.put("enable.auto.commit", "false");

         KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
         consumer.subscribe(Collections.singletonList(topicName)); 
         while (true) {
                ConsumerRecords<String, String> records = consumer.poll(100);
                for (ConsumerRecord<String, String> record : records)
                {
                    System.out.println("topic = "+record.topic());
                    System.out.println("topic = "+record.partition());
                    System.out.println("topic = "+record.offset());
                }
                try {
                  consumer.commitSync(); 
                } catch (CommitFailedException e) {
                    System.out.printf("commit failed", e) ;
                }
            }

我使用这种依赖：

<dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka_2.11</artifactId>
        <version>0.10.0.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>0.10.1.0</version>
    </dependency>

I get all information from that link:
https://kafka.apache.org/0100/javadoc/index.html?org/apache/kafka/clients/consumer/KafkaConsumer.html

当我们运行消费者时，我们没有收到消费者方面的任何通知 . 请给我任何想法 .

2 回答

0

对于制片人：

properties.put（“metadata.broker.list”，“Running kafka ip addr：9092”）;

我猜，这应该是“bootstrap.servers” .

对于消费者：

properties.put（“bootstrap.servers”，“运行zookeaper ip addr：2181”）;

bootstrap.servers 必须指向经纪人，而不是ZK .

“问题”是，如果指定的主机/端口没有代理，则消费者将只等待代理但不会失败 .

回复于 2024-04-29T05:12:54+08:00
0
我是Kafka和Java的新手，但我想建议以下方法
- 使用以下命令 /usr/bin/kafka-avro-console-consumer --new-consumer --bootstrap-server localhost:9092 --topic KumarTopic --from-beginning 验证生产环境者是否实际写入主题 .
- 如果是，您'll probably need to focus on your consumer code. Confluent'的指南非常有帮助 .
回复于 2024-04-29T05:12:54+08:00

消费者没有在Apache Kafka中收到消息

2 回答

相关问题