美文网首页
kafka consumer 使用总结

kafka consumer 使用总结

作者: 大数据技术进阶 | 来源:发表于2018-06-14 12:26 被阅读0次

    Kafka的consumer是以pull的形式获取消息数据的,consumer提供两种版本,即high level 和low level API。

    1 consumer和partition

    1)如果consumer比partition多,是浪费,因为kafka的设计是在一个partition上是不允许并发的,所以consumer数不要大于partition数
    2)如果consumer比partition少,一个consumer会对应于多个partitions,这里主要合理分配consumer数和partition数,否则会导致partition里面的数据被取的不均匀,最好partiton数目是consumer数目的整数倍,所以partition数目很重要,比如取24,就很容易设定consumer数目
    3)如果consumer从多个partition读到数据,不保证数据间的顺序性,kafka只保证在一个partition上数据是有序的,但多个partition,根据你读的顺序会有不同
    4)增减consumer,broker,partition会导致rebalance,所以rebalance后consumer对应的partition会发生变化
    5)High-level接口中获取不到数据的时候是会block的。
    6)突然停止Consumer以及Broker会导致消息重复读的情况,为了避免这种情况在shutdown之前通过Thread.sleep(10000)让Consumer有时间将offset同步到zookeeper

    2 auto.offset.reset

    测试代码如下:

    public class kafkaConsumer extends Thread{
    
        private String topic;  
    
        public kafkaConsumer(String topic){  
            super();  
            this.topic = topic;  
        }  
    
    
        @Override  
        public void run() {  
            ConsumerConnector consumer = createConsumer();  
            Map<String, Integer> topicCountMap = new HashMap<String, Integer>();  
            topicCountMap.put(topic, 1); // 一次从主题中获取一个数据  
             Map<String, List<KafkaStream<byte[], byte[]>>>  messageStreams = consumer.createMessageStreams(topicCountMap);  
             KafkaStream<byte[], byte[]> stream = messageStreams.get(topic).get(0);// 获取每次接收到的这个数据  
             ConsumerIterator<byte[], byte[]> iterator =  stream.iterator();  
             while(iterator.hasNext()){  
                 String message = new String(iterator.next().message());  
                 System.out.println("接收到: " + message);  
             }  
        }  
    
        private ConsumerConnector createConsumer() {  
            Properties properties = new Properties();  
            properties.put("zookeeper.connect", "ip1:2181,ip2:2181,ip3:2181");//声明zk  
            properties.put("group.id", "group03");
            return Consumer.createJavaConsumerConnector(new ConsumerConfig(properties));  
         }  
    
    
        public static void main(String[] args) {  
            new kafkaConsumer("user").start();// 使用kafka集群中创建好的topic:user      
        }  
    }
    

    auto.offset.reset 默认值为largest,那么auto.offset.reset 有什么作用呢?auto.offset.reset定义了Consumer在ZooKeeper中发现没有初始的offset时或者发现offset非法时定义Comsumer的行为,常见的配置有:
    smallest : 自动把offset设为最小的offset;
    largest : 自动把offset设为最大的offset;
    anything else: 抛出异常;
    遇到过这种情况:先produce一些数据,然后停止produce数据的线程——〉然后再用consumer上面的代码消费数据,发现无数据可消费!
    其原因在于:初始的offset默认是非法的,而auto.offset.reset 默认值为largest,表示自动把offset设为最大的offset,由于此时没有生产者向kafka push数据,当然没有数据可以消费了。如果此时有生产者向kafka push数据,那么该代码可以从最新位置消费数据。
    如果在代码中增加如下配置:

    properties.put("auto.offset.reset", "smallest"); 
    

    那么在停止生产者线程之后,再启动消费者线程可以消费之前produce的数据。
    3 high-level的Consumer工具
    3.1 kafka.tools.ConsumerOffsetChecker
    使用如下命令查看当前group的offset情况

    ./kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --group group03
    
    111.png

    或者指定topic

    ./kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --topic user01 --group group03
    

    如上图所示pid表示topic的partition号,上图中topic为user0的partiton数量为10;现在只启动生产者线程一段时间后,再次运行上面的命令,发现如下:


    222.png

    3.2 kafka.tools.UpdateOffsetsInZK

    ./kafka-run-class.sh kafka.tools.UpdateOffsetsInZK earliest config/consumer.properties  user01
    

    该命令的三个参数
    [earliest | latest],表示将offset置到哪里
    consumer.properties ,这里是配置文件的路径
    topic,topic名,这里是user01

    相关文章

      网友评论

          本文标题:kafka consumer 使用总结

          本文链接:https://www.haomeiwen.com/subject/pkzeeftx.html