zoukankan      html  css  js  c++  java
  • 四、 kafka consumer 配置

    consumer配置

    #指明当前消费进程所属的消费组,一个partition只能被同一个消费组的一个消费者消费(同一个组的consumer不会重复消费同一个消息)
    group.id

    #针对一个partition的fetch request所能拉取的最大消息字节数,必须大于等于Kafka运行的最大消息
    fetch.message.max.bytes  1024 * 1024

    #是否自动周期性提交已经拉取到消费端的消息offset; 如果此值设置为true,consumer会周期性的把当前消费的offset值保存到zookeeper(注意,新版的

    kafka已经不把offset信息保存到zookeeper上了,而是保存到broker上。

    auto.commit.enable true

    consumer端丢失消息的情形

    1: auto.commit.enable=true 在消息处理完成前就提交了offset,那么就有可能造成数据的丢失。由于Kafka consumer默认是自动提交位移的,所以在后台提交位移前一定要保证消息被正常处理了,因此不建议采用很重的处理逻辑,如果处理耗时很长,则建议把逻辑放到另一个线程中去做。为了避免数据丢失,现给出两点建议:

    • enable.auto.commit=false  关闭自动提交位移
    • 在消息被完整处理之后再手动提交位移

    2 如果auto.commit.enable=false也可能出现数据丢失的情况。假设consumer的两个fetcher各自拿了一条数据(这种情形是partition的数量大于consumer group中的consumer的数量,这样就会有一个consumer同时消耗两个partition的数据),并且由两个线程同时处理,这时线程t1处理完partition1的数据,手动提交offset,这里需要着重说明的是,当手动执行commit的时候,实际上是对这个consumer进程所占有的所有partition进行commit,kafka暂时还没有提供更细粒度的commit方式,也就是说,即使t2没有处理完partition2的数据,offset也被t1提交掉了。如果这时consumer crash掉,t2正在处理的这条数据就丢失了。

    • 如果希望能够严格的不丢数据,解决办法有两个:

      1. 手动commit offset,并针对partition_num启同样数目的consumer进程,这样就能保证一个consumer进程占有一个partition,commit offset的时候不会影响别的partition的offset。但这个方法比较局限,因为partition和consumer进程的数目必须严格对应。
      2. 另一个方法同样需要手动commit offset,另外在consumer端再将所有fetch到的数据缓存到queue里,当把queue里所有的数据处理完之后,再批量提交offset,这样就能保证只有处理完的数据才被commit。当然这只是基本思路,实际上操作起来不是这么简单,具体做法以后我再另开一篇。



    #自动提交offset到zookeeper的时间间隔
    auto.commit.interval.ms  60 * 1000

    #消费均衡的重试次数(当新的consumer加入到consumer  group时,consumers集合试图重新平衡分配到每个consumer的partitions数目(即重新做负载均衡)。如果consumers集合改变了,当分配正在执行时(正在做负载均衡时,consumer的数目变化了),这个重新平衡会失败并重入(负载均衡会失败并且重新负载均衡))
    rebalance.max.retries  4

    #消费均衡两次重试之间的时间间隔
    rebalance.backoff.ms 2000

    #当重新去获取partition的leader前需要等待的时间
    refresh.leader.backoff.ms   200

    #如果zookeeper上没有offset合理的初始值情况下获取第一条消息开始的策略smallest|largeset
    auto.offset.reset largest(新的consumer加入进来的时候,从最新的消息消费,还是从最早的消息开始消费)

    #如果其超时,将会可能触发rebalance并认为已经死去
    zookeeper.session.timeout.ms  6000(zookeeper 会话的超时限制。如果consumer在这段时间内没有向zookeeper发送心跳信息,则它会被认为挂掉了,并且reblance将会产生)

    #确认zookeeper连接建立操作客户端能等待的最长时间
    zookeeper.connection.timeout.ms 6000 ---------------------  

  • 相关阅读:
    Linux命令
    Linux目录说明
    python推导式
    python公共方法
    python集合
    python字典
    python元组
    python列表
    python字符串常用操作方法
    C语言编译过程
  • 原文地址:https://www.cnblogs.com/liufei1983/p/9728681.html
Copyright © 2011-2022 走看看