kafka-consumer-api - 卡夫卡消费者没有收到消息

Question

我是卡夫卡的新手。我在互联网上阅读了很多说明来制作 Kafka Producer 和 Kafka Consumer。我成功地完成了前者，它可以将消息发送到 Kafka 集群。但是，我没有完成后一个。请帮助我解决这个问题。我看到我的问题喜欢 StackOverflow 上的一些帖子，但我想更清楚地描述。
我在 Virtual Box 的 Ubuntu 服务器上运行 Kafka 和 Zookeeper。使用 1 个 Kafka 集群和 1 个 Zookeeper 集群的最简单配置(几乎是默认值)。

1.当我将Kafka的命令行用于生产者和消费者时，例如:

* Case 1: It works. I can see the word: Hello, World on the screen

$~/kafka/bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic helloKafka --partitions 1 --replication-factor 1.
$echo "Hello, World" | ~/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic TutorialTopic > /dev/null.
$~/kafka/bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic TutorialTopic --from-beginning.

2.当我使用Kafka的Producer和命令行进行消费者时，例如:

* Case 2: It works. I can see the messages which sent from the Producer on the screen

$~/kafka/bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic helloKafka --partitions 1 --replication-factor 1.
$~/kafka/bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic TutorialTopic --from-beginning.
$java -cp target/Kafka_Producer_Program-0.0.1-SNAPSHOT.jar AddLab_Producer

3.当我使用生产者和消费者时，比如:

* Case 3: Only Producer works perfectly. The Consumer runs but does not shows any messages. 

$~/kafka/bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic helloKafka --partitions 1 --replication-factor 1.
$java -cp target/Kafka_Producer_Program-0.0.1-SNAPSHOT.jar AddLab_Producer
$java -cp target/Kafka_Consumer_Program-0.0.1-SNAPSHOT.jar AddLab_Consumer

这是我的生产者和消费者代码。实际上，我是从 Kafka 的一些说明网站复制的。

*制作人计划

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
public class AddLab_Producer {
    public static void main(String args[]) throws InterruptedException, ExecutionException {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);

        boolean sync = false;
        String topic = args[0];
        String key = "mykey";

        for (int i = 1; i <= 3; i++) {
            String value = args[1] + " " + i;
            ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(topic, value);
            if (sync) {
                producer.send(producerRecord).get();
            } else {
                producer.send(producerRecord);
            }
        }
        producer.close();
    }
}

*消费者计划

import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class AddLab_Consumer {

    public static class KafkaPartitionConsumer implements Runnable {

        private int tnum ;
        private KafkaStream kfs ;

        public KafkaPartitionConsumer(int id, KafkaStream ks) {
            tnum = id ;
            kfs = ks ;
        }   
        public void run() {
            // TODO Auto-generated method stub
            System.out.println("This is thread " + tnum) ;

            ConsumerIterator<byte[], byte[]> it = kfs.iterator();
                int i = 1 ;
                while (it.hasNext()) {
                    System.out.println(tnum + " " + i + ": " + new String(it.next().message()));
                    ++i ;
                }       
        }
    }

    public static class MultiKafka {    
        public void run() {
        }   
    }

    public static void main(String[] args) {

        Properties props = new Properties();
        props.put("zookeeper.connect", "localhost:2181");
        props.put("group.id", "mygroupid2");
        props.put("zookeeper.session.timeout.ms", "413");
        props.put("zookeeper.sync.time.ms", "203");
        props.put("auto.commit.interval.ms", "1000");

        ConsumerConfig cf = new ConsumerConfig(props) ;    
        ConsumerConnector consumer = Consumer.createJavaConsumerConnector(cf) ;      
        String topic = "mytopic" ;     
        Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
        topicCountMap.put(topic, new Integer(1));
        Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);
        List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(topic);

        ExecutorService executor = Executors.newFixedThreadPool(1); ;

        int threadnum = 0 ;     
        for(KafkaStream<byte[],byte[]> stream  : streams) { 
            executor.execute(new KafkaPartitionConsumer(threadnum,stream));
            ++threadnum ;
        }
    }
}

*我的POM.xml文件

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>HelloJava</groupId>
    <artifactId>HelloJava</artifactId>
    <version>0.0.1-SNAPSHOT</version>

    <dependencies>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka_2.10</artifactId>
            <version>0.9.0.0</version>
        </dependency>

        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>3.8.1</version>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <build>
        <sourceDirectory>src</sourceDirectory>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>1.4</version>
                <configuration>
                    <createDependencyReducedPom>true</createDependencyReducedPom>
                    <filters>
                        <filter>
                            <artifact>*:*</artifact>
                            <excludes>
                                <exclude>META-INF/*.SF</exclude>
                                <exclude>META-INF/*.DSA</exclude>
                                <exclude>META-INF/*.RSA</exclude>
                            </excludes>
                        </filter>
                    </filters>
                </configuration>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

我非常感谢您的帮助。非常感谢。

Screen of Consumer. It seem runs but can not receive any messages from Producer

Answer 1

我遇到了和你一样的问题。经过长时间的尝试，这就是答案。
有两种类型的 kafka 新消费者 api，您可以选择一种。

cousumer.assign(...)

consumer.subscribe(..)

并使用如下:

    // set these properites or you should run consumer first than run producer
    props.put("enable.auto.commit", "false");
    props.put("auto.offset.reset", "earliest");

    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

    boolean assign = false;
    if(assign) {
        TopicPartition tp = new TopicPartition(topic, 0);
        List<TopicPartition> tps = Arrays.asList(tp);
        consumer.assign(tps);
        consumer.seekToBeginning(tps);
    }else {
        consumer.subscribe(Arrays.asList(topic));
    }

http://kafka.apache.org/documentation.html#newconsumerconfigs
如果您使用旧的消费者 api，则属性配置几乎相同。
如果要查看消费者消费前产生的消息，记得添加以下两段代码:

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

希望这会帮助其他人。