Kafka消费者API无法正常工作-Java 学习之路

我是Kafka的新手 . 我开始在Kafka做我面临的问题请帮助我提前解决这个问题 . 首先，我正在编写生产环境者API，它工作正常，但不显示消费者API消息 .

我的代码是这样的：

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

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


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

        String topic = "Hello-Kafka";
        String group = "myGroup";
        Properties props = new Properties();
        props.put("bootstrap.servers", "XXX.XX.XX.XX:9092");
        props.put("group.id", group);
        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);
        try {

            consumer.subscribe(Arrays.asList(topic));
            System.out.println("Subscribed to topic " + topic);


            ConsumerRecords<String, String> records = consumer.poll(100);

            System.out.println("records ::" + records);
            System.out.println(records.toString());
            for (ConsumerRecord<String, String> record : records) {
                System.out.println("Record::" + record.offset());
                System.out.println(record.key());
                System.out.println(record.value());
            }
            consumer.commitSync();

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            consumer.commitSync();
            consumer.close();
        }
    }
}

回复::

订阅主题Hello-Kafka记录:: org.apache.kafka.clients.consumer.ConsumerRecords@76b0bfab org.apache.kafka.clients.consumer.ConsumerRecords@76b0bfab

这里没有打印偏移，键，值控制不是为了（ConsumerRecord记录：记录）{为循环它自己请帮助我 .

1 回答

您正在尝试打印空记录，因此只有records.toString（）在您的代码中打印，这基本上是类的名称 .
我对您的代码进行了一些更改并使其正常工作 . 看看这是否有帮助 .

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

        String topic = "Hello-Kafka";
        String group = "myGroup";
        Properties props = new Properties();
        props.put("bootstrap.servers", "xx.xx.xx.xx:9092");
        props.put("group.id", group);
        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);
        try {

            consumer.subscribe(Arrays.asList(topic));
            System.out.println("Subscribed to topic " + topic);

            while(true){
                ConsumerRecords<String, String> records = consumer.poll(1000);
                if(records.isEmpty()){

                }
                else{
                System.out.println("records ::" + records);
                System.out.println(records.toString());
                for (ConsumerRecord<String, String> record : records) {
                    System.out.println("Record::" + record.offset());
                    System.out.println(record.key());
                    System.out.println(record.value());
                }
                consumer.commitSync();
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            consumer.commitSync();
            consumer.close();
        }
    }
}

回复于 2024-05-14T01:13:50+08:00

Kafka消费者API无法正常工作

1 回答

相关问题