Kafka服务器包含消息，但消费者无法接收任何消息

Question

我使用一个生产者在本地主机上运行的Kafka服务器中输入了一些消息。Zookeeper也在本地主机上。我使用了这里给出的ConsumerGroupExample。

但是，消费者似乎没有收到任何消息！kafka-console-consumer.sh 脚本可以提取所有这些消息，但代码不能。怎么了？消费者代码与该页面上给出的代码完全相同。

zookeeper="localhost:2181", group id = "test-a", topic = "test". 

这是我发布消息的同一主题。以下是制作人的代码：

package test;

import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import java.util.Properties;
import kafka.javaapi.producer.Producer;

public class KakfaProducer {
    public static void main(String[] args) {
        Properties prop = new Properties();
        prop.put("zookeeper.connect", "localhost:2181");
        prop.put("metadata.broker.list", "localhost:9092");
        prop.put("serializer.class", "kafka.serializer.StringEncoder");
        ProducerConfig producerConfig = new ProducerConfig(prop);
        Producer<String, String> producer = new<String, String> Producer(producerConfig);
        Task t = new Task(producer);
        Thread newThread = new Thread(t);
        newThread.start();
    }
}

class Task implements Runnable {
    Producer<String, String> producer;

    public Task(Producer producer) {
        this.producer = producer;
    }

    public void run() {
        long num = 1;
        while (true) {
            String topic = "test";
            KeyedMessage<String, String> message = new<String, String> KeyedMessage(topic, "Hello Test message " + num);
            producer.send(message);
            synchronized (this) {
                num++;
            }
            if (num % 1000 == 0) {
                System.out.println("Total messages sent by Thread-" + Thread.currentThread().getId() + num);
            }
        }
    }
}

Answer 1

尝试将 group.id 更改为其他内容并进行测试。

当您使用 kafka-console-conumser.sh 测试使用者时，它会使用主题中的数据并更改偏移量。这就是您的使用者代码无法使用相同的 group.id 使用的原因。

此外，当您更改组时。id，添加到您的消费者：

props.put("auto.offset.reset", "earliest");

这将确保您的新消费者群体从一开始就使用数据。

您的消费者没有收到任何数据:

>

您已使用同一组。id，并且它有一个新的偏移量。它将仅在Broker中有新数据时使用。（更改组.id并再次测试。）

尝试 auto.offset.reset=最早。使用者将从一开始就读取消息，您可以使用它来测试您的使用者。还有一些方法可以从某个偏移量读取。seek（）、seekToBeginning（） 和 seekToEnd（）。