kafka 消费者offset记录位置和方式

我们大家都知道，kafka消费者在会保存其消费的进度，也就是offset，存储的位置根据选用的kafka api不同而不同。

首先来说说消费者如果是根据javaapi来消费，也就是【kafka.javaapi.consumer.ConsumerConnector】，我们会配置参数【zookeeper.connect】来消费。这种情况下，消费者的offset会更新到zookeeper的【consumers/{group}/offsets/{topic}/{partition}】目录下，例如：

[zk: localhost(CONNECTED) 0] get /kafka/consumers/zoo-consumer-group/offsets/my-topic/0

5662

cZxid = 0x20006d28a

ctime = Wed Apr 12 18:20:51 CST 2017

mZxid = 0x30132b0ed

mtime = Tue Aug 22 18:53:22 CST 2017

pZxid = 0x20006d28a

cversion = 0

dataVersion = 5758

aclVersion = 0

ephemeralOwner = 0x0

dataLength = 4

numChildren = 0

如果是根据kafka默认的api来消费，即【org.apache.kafka.clients.consumer.KafkaConsumer】，我们会配置参数【bootstrap.servers】来消费。而其消费者的offset会更新到一个kafka自带的topic【__consumer_offsets】下面，查看当前group的消费进度，则要依靠kafka自带的工具【kafka-consumer-offset-checker】，例如：

[root@localhost data]# kafka-consumer-offset-checker --zookeeper localhost :2181/kafka --group test-consumer-group  --topic stable-test

[2017-08-22 19:24:24,222] WARN WARNING: ConsumerOffsetChecker is deprecated and will be dropped in releases following 0.9.0. Use ConsumerGroupCommand instead. (kafka.tools.ConsumerOffsetChecker$)

Group          Topic                          Pid Offset          logSize        Lag            Owner

test-consumer-group stable-test                    0  601808          601808          0              none

test-consumer-group stable-test                    1  602826          602828          2              none

test-consumer-group stable-test                    2  602136          602136          0              none

上面结果的说明：

Group : 消费者组

Topic : topic的名字

Pid : partition的ID

Offset : kafka消费者在对应分区上已经消费的消息数【位置】

logSize : 已经写到该分区的消息数【位置】

Lag : 还有多少消息未读取（Lag = logSize - Offset）

Owner : 分区创建在哪个broker

offset更新的方式，不区分是用的哪种api，大致分为两类：

自动提交，设置enable.auto.commit=true，更新的频率根据参数【auto.commit.interval.ms】来定。这种方式也被称为【at most once】，fetch到消息后就可以更新offset，无论是否消费成功。

手动提交，设置enable.auto.commit=false，这种方式称为【at least once】。fetch到消息后，等消费完成再调用方法【consumer.commitSync()】，手动更新offset；如果消费失败，则offset也不会更新，此条消息会被重复消费一次。

Kafka如何保证数据不重复消费，不丢失数据

不重复消费：

1.幂等操作，重复消费不会产生问题

2. dstream.foreachRDD {(rdd, time) =

  rdd.foreachPartition { partitionIterator =>

    val partitionId = TaskContext.get.partitionId()

val uniqueId = generateUniqueId(time.milliseconds,partitionId)将uniqueID存入数据库中

    //use this uniqueId to transationally commit the data in partitionIterator

 }

}

对每个partitionID，产生一个uniqueID,.只有这个partition的数据被完全消费，才算成功，否则失败回滚。下次若重复执行，就skip

不丢失数据：丢失情况： 

1.生产者数据不丢失

      同步模式：配置=1（只有Leader收到，-1所有副本成功，0不等待）。leader partition挂了，数据就会丢失。

                  解决：设置为-1保证produce写入所有副本算成功

　　　　　　　　　　producer.type=sync

　request.required.acks=-1

      异步模式，当缓冲区满了，如果配置为0（没有收到确认，一满就丢弃），数据立刻丢弃

                解决：不限制阻塞超时时间。就是一满生产者就阻塞

　　　　　　　　　　producer.type=async

　　　　　　　　　　request.required.acks=1

　　　　　　　　　　queue.buffering.max.ms=5000

　　　　　　　　　　queue.buffering.max.messages=10000

queue.enqueue.timeout.ms = -1

　　　　　　　　　　batch.num.messages=200

2.消费者数据不丢失 ：流计算，基本数据源不适用。高级数据源以kafka为例，由2种方式：receiver(开启WAL,失败可恢复）和director（checkpoint保证）

3.   若是storm在消费，开启storm的ackfail机制；若不是storm,数据处理完更新offset,低级API手动控制offset

4.   Kafka发送数据过快，导致服务器网卡流量暴增。或磁盘过忙，出现丢包。

　　　　　　1》  首先，对kafka进行限速，

　　　　　　2》  其次启用重试机制，使重试间隔变长。

　　　　　　3》  Kafka设置ack=all,即需要处于ISR（副本列表）的分区都确认，才算发送成功。    rops.put("compression.type", "gzip");

        　　　　　　 props.put("linger.ms", "50");

        　　　　　　 props.put("acks", "all")表示至少成功发送一次;

        　　　　　　 props.put("retries ", 30);

        　　　　　　 props.put("reconnect.backoff.ms ", 20000);

       　　　　　　  props.put("retry.backoff.ms", 20000)

　5.消费者速度很慢，导致一个session周期（0.1版本是默认30s）内未完成消费。导致心跳机制检测报告出问题。

   导致消费了的数据未及时提交offset.配置由可能是自动提交

            问题场景：1.offset为自动提交，正在消费数据，kill消费者线程，下次重复消费

　　　　　　　　 2.设置自动提交，关闭kafka,close之前，调用consumer.unsubscribed()则由可能部分offset没有提交。

     　　　　　　   3.消费程序和业务逻辑在一个线程，导致offset提交超时，