0

我正在按照 MapR 的教程制作基本的 Kafka 生产者/消费者：

https://github.com/mapr-demos/kafka-sample-programs

我正在使用带有 Kafka 0.9.2 的 Hortonworks Sandbox VM 2.4 版这是我的 Producer 代码：

package com.whatever.kafka.basic;

...

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.ProducerConfig;

public class BasicProducer {

    public static void main(String[] args) throws IOException {

        /* comment-out snippet from MapR example; hard-code properties instead
        KafkaProducer<String, String> producer;
        try (InputStream props = Resources.getResource("producer.props").openStream()) {
            Properties properties = new Properties();
            properties.load(props);
            producer = new KafkaProducer<String, String>(properties);
        }
        */

        Properties props = new Properties();
        props.put("bootstrap.servers", "sandbox.hortonworks.com:6667");
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("auto.commit.interval.ms", 1000);
        props.put("linger.ms", 0);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("block.on.buffer.full", true);

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

        try {
            for (int i = 0; i < 1000000; i++) {

                System.out.println("Test 1"); // prints to console and stops on the next line

                producer.send(new ProducerRecord<String, String>(
                    "fast-messages", String.format("{\"type\":\"marker\", \"t\":%.3f, \"k\":%d}", System.nanoTime() * 1e-9, i)  
                ));

                System.out.println("Test 2");

                if (i % 250 == 0) {
                    producer.send(new ProducerRecord<String, String>(
                            "fast-messages", String.format("{\"type\":\"marker\", \"t\":%.3f, \"k\":%d}", System.nanoTime() * 1e-9, i)
                    ));
                    producer.send(new ProducerRecord<String, String>(
                            "summary-markers", String.format("{\"type\":\"other\", \"t\":%.3f, \"k\":%d}", System.nanoTime() * 1e-9, i)
                    )); 
                    producer.flush();
                    System.out.println("Sent msg number " + i);
                }
            }
            producer.close();
        } catch (Throwable throwable) {
            System.out.printf("%s", throwable.getStackTrace());
        } finally {
            producer.close();
        }
    } 
}

我已经启动了 Zookeeper 和 Kafka，并按照说明创建了主题：

$ bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic fast-messages
$ bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic summary-markers

但是当我运行编译的应用程序时

$ ./kafka-example producer

我得到这个输出：

log4j:WARN No appenders could be found for logger (org.apache.kafka.clients.producer.ProducerConfig)
log4j:WARN Please initialize the log4j system properly
Test 1

它卡在sysout 日志producer.send之后的第一行。Test 1也没有抛出异常，它只是在打印出该Test 1行时停止。当我使用 Eclipse 使用换行符对其进行调试时，它也停止在该行上，显然没有发生任何事情。

Kafka 日志中的fast-messages和summary-markers目录已创建，但没有写入任何内容。

任何想法？

4

0 回答 0