【发布时间】:2016-10-26 04:07:36
【问题描述】:
我可以在命令行上针对 Kafka 位置安装发送和接收消息。我还可以通过 Java 代码发送消息。这些消息显示在 Kafka 命令提示符中。我还有一个用于 Kafka 消费者的 Java 代码。代码昨天收到消息。然而,今天早上它没有收到任何消息。代码没有改变。我想知道属性配置是否不太正确。这是我的配置:
制作人:
bootstrap.servers - localhost:9092
group.id - test
key.serializer - StringSerializer.class.getName()
value.serializer - StringSerializer.class.getName()
ProducerRecord 设置为
ProducerRecord<String, String>("test", "mykey", "myvalue")
消费者:
zookeeper.connect - "localhost:2181"
group.id - "test"
zookeeper.session.timeout.ms - 500
zookeeper.sync.time.ms - 250
auto.commit.interval.ms - 1000
key.deserializer - org.apache.kafka.common.serialization.StringDeserializer
value.deserializer - org.apache.kafka.common.serialization.StringDeserializer
对于 Java 代码:
Map<String, Integer> topicCount = new HashMap<>();
topicCount.put("test", 1);
Map<String, List<KafkaStream<byte[], byte[]>>> consumerStreams = consumer
.createMessageStreams(topicCount);
List<KafkaStream<byte[], byte[]>> streams = consumerStreams.get(topic);
缺少什么?
【问题讨论】:
标签: apache-kafka kafka-consumer-api