satellite7832
2017-05-25 02:02kafka消费不到数据问题
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();
}
}
- 点赞
- 回答
- 收藏
- 复制链接分享
1条回答
为你推荐
- 从kafka里面消费数据,就是连接不到kafka,在kafka里面生成和消费都没有问题,也没有报错,求大神指教哪里有问题,我也从来没遇到过这个问题,求指教?
- scala
- 1个回答
- 如何在Golang Kafka 10中获取分区的消费者组偏移量
- it技术
- 互联网问答
- IT行业问题
- 计算机技术
- 编程语言问答
- 1个回答
- 如何使用php-rdkafka在kafka中确认消费消息?
- php
- 1个回答
- 救命!!!!!nginx+lua+kafka 做日志收集出现发送消息问题,could not be resolved
- kafka
- dns
- 阿里云
- nginx
- lua
- 2个回答