kafka consumer无法获取topic消息。

魅影灬紫苑 发布于 2017/03/24 16:55
阅读 554
收藏 0

Properties properties = new Properties();  
    properties.put("zookeeper.connect", “192.········”);//声明zk  
    properties.put("group.id", "group2xx");// 必须要使用别的组名称, 如果生产者和消费者都在同一组,则不能访问同一组内的topic数据  
    properties.put("auto.offset.reset", "smallest");

    ConsumerConnector consumer = Consumer.createJavaConsumerConnector(new ConsumerConfig(properties));  
    Map<String, Integer> topicCountMap = new HashMap<String, Integer>();  
    topicCountMap.put("zuo", 1); // 一次从主题中获取一个数据  
    Map<String, List<KafkaStream<byte[], byte[]>>>  messageStreams = consumer.createMessageStreams(topicCountMap);  
    KafkaStream<byte[], byte[]> stream = messageStreams.get("zuo").get(0);// 获取每次接收到的这个数据  
    ConsumerIterator<byte[], byte[]> iterator =  stream.iterator();  
    System.out.println("----------");
    while(iterator.hasNext()){
        System.out.println("`#############");
        System.out.println("receive:" + new String(iterator.next().message()));
    }
    
        
}
    

加载中
0
魅影灬紫苑
魅影灬紫苑

是哪里写错了嘛。我在客户端可以看到信息,在java api上面看不到

返回顶部
顶部