kafka集群搭建正常，通过console都能正常生产和消费消息，但是通过JAVA程序就是读取不到消息，更换group都尝试过了
package test;
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 extends Thread {
private String topic;
public KafkaConsumer(String topic){
super();
this.topic=topic;
}
@Override
public void run(){
//通过properties设置了Consumer的参数，并且创建了连接器，连接到Kafaka
ConsumerConnector consumer = createConsumer();
//Map作用指定获取的topic以及partition
Map<String,Integer> topicCountMap = new HashMap<String,Integer>();
topicCountMap.put(topic, 3);
//consumer连接器获取消息
Map<String,List<KafkaStream<byte[],byte[]>>> messageStreams = consumer.createMessageStreams(topicCountMap);
//获取对应的topic中的某一个partition中的数据
KafkaStream<byte[],byte[]> kafkaStream = messageStreams.get(topic).get(0);
ConsumerIterator<byte[], byte[]> iterator = kafkaStream.iterator();
while(iterator.hasNext()){
byte[] message = iterator.next().message();
System.out.println("message is:"+new String(message));
}
}
private ConsumerConnector createConsumer(){
Properties properties = new Properties();
properties.put("zookeeper.connect", "XXX:2181");
properties.put("auto.offset.reset", "smallest");//读取旧数据
properties.put("group.id", "333fcdcd");
return Consumer.createJavaConsumerConnector(new ConsumerConfig(properties));
}
public static void main(String[] args) {
new KafkaConsumer("testtest").start();
}
}
