卡夫卡消费者不能从日食中工作答案

【问题标题】：Kafka consumer does not work from eclipse卡夫卡消费者不能从日食中工作
【发布时间】：2015-09-12 21:13:19
【问题描述】：

Kafka 0.8.2.2.3 和 zookeper 都在 VM 中运行。我能够分别使用 kafka-console-producer.sh 和 kafka-console-consumer.sh 在 VM 中成功运行生产者和消费者。甚至我也能够使用 kafka-console-consumer.sh 从主机消费 Kafka 消息。但是当我尝试使用 eclipse 中的 java 运行消费者时，zookeeper 会记录以下错误

2015-06-26 03:06:26,323 - INFO  [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2181:NIOServerCnxn@1007] - Closed socket connection for client /192.168.1.12:59549 (no session established for client)
2015-06-26 03:07:26,225 - INFO  [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2181:NIOServerCnxnFactory@197] - Accepted socket connection from /192.168.1.12:59617
2015-06-26 03:07:26,226 - WARN  [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2181:NIOServerCnxn@357] - caught end of stream exception
EndOfStreamException: Unable to read additional data from client sessionid 0x0, likely client has closed socket
    at org.apache.zookeeper.server.NIOServerCnxn.doIO(NIOServerCnxn.java:228)
    at org.apache.zookeeper.server.NIOServerCnxnFactory.run(NIOServerCnxnFactory.java:208)
    at java.lang.Thread.run(Thread.java:745)

以下是我的 Kafka 消费代码

package com.truckevent.producer;


import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;


public class KafkaConsumer {

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

        String group = "hello" ;


        Properties props = new Properties();
        props.put("zookeeper.connect", "192.168.1.12:2181");
        props.put("group.id", group);
        props.put("zookeeper.session.timeout.ms", "20000");
        props.put("zookeeper.sync.time.ms", "2030");
        props.put("auto.commit.interval.ms", "10000");
        props.put("auto.offset.reset", "smallest");

        ConsumerConfig cf = new ConsumerConfig(props) ;

        ConsumerConnector consumer = Consumer.createJavaConsumerConnector(cf) ;

        String topic = "event" ;

        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);


        KafkaStream<byte[],byte[]> stream = streams.get(0) ;

        ConsumerIterator<byte[], byte[]> it = stream.iterator();
        int i = 1 ;
        while (it.hasNext()) {

            System.out.println(i + ": " + new String(it.next().message()));
            ++i;
        }
        consumer.shutdown(); 
    }
}

我不知道为什么我不能使用来自 java 代码的消息。 Kafka 运行在 6667 端口，zookeeper 运行在 2181。

【问题讨论】：

你能在这个端口上telnet到这个IP地址吗？
你知道为什么吗？我有同样的问题。
使用 PortQry 检查防火墙。

标签： apache-zookeeper apache-kafka

【解决方案1】：

请检查 zookeeper 是否绑定（netstat -lntp）到 0.0.0.0 或只到 localhost（然后它将只接受来自 vm 本身的连接），这是一个相当新的选项（clientPortBindAddress，检查你是否指定了它，默认情况下，如果您不指定它，它会绑定到所有地址）
它可能连接到您的虚拟机配置。例如。在我使用 vagrant+virtual box 的情况下，我使用 config.vm.network 在主机和来宾机器之间创建专用网络：private_network，ip：192.168.1.12
此外 - 检查您的消费者主程序是否没有抛出一些异常，它可能已连接并会提供一些额外信息（您当前正在将异常抛出到最高级别）

【讨论】：