@
1. 消费者
消费者Consumer负责订阅kafka中的主题。并从主题中拉取消息。每个消费者都有一个对应的消费组。
两个消费组之间互不影响，每一个分区只能被一个消费组中的一个消费者所消费。
我们可以增加（或减少）消费者的个数来提高（或降低）整体的消费能力，但是过多增加消费者个数并不能一致提高消费水平，消费者个数大于分区数，有的消费者不会分配到任何分区。
消息投递的两种方式：
- 点对点：所有的消费者都隶属于一个消费组
- 发布/订阅模式：所有的消费者都属于不同的消费组
消费组是一个逻辑概念，每一个消费者只隶属于一个消费组
消费者并非逻辑上的概念，实际应用，可以是一个线程/进程
2. 客户端开发
一个正常的消费逻辑是：
- 配置消费者客户端参数及创建相应的消费者实例
- 订阅主题
- 拉取消息并消费
- 提交消费位移
- 关闭消费者实例
订阅方式：
- 集合订阅方式subscribe
- 正则表达式subscribe(Pattern)
- 指定分区assign(Collection)
通过subscribe方法订阅主题具有消费者再（Reblance）均衡的功能，在多个消费者的情况下，根据分区分配策略来自动分配各个消费者与分区的关系。
2.1 消息消费
Kafka的消费是基于拉模式的，kafka消息消费是一个不断轮训的过程，消费者所要做的就是重复调用poll()方法。
- 对于Kafka的分区而言，将offset称为偏移量，每条消息都有唯一的offset，用来表示消息在分区中的对应的位置。
- 对于消费者也有offset，将offset称为消费位移，消费者用offset来表示消费到分区中某个消息所在的位置。
对于消费位移来说，必须持久化保存。
每次调用poll()方法，他返回的是还没消费过的消息集，那么就需要记录上一次消费的消费位移，并且需要做持久化保存，例如，对于一个分区，在再均衡触发之后，如果持久化保存消费位移，那么这个新的消费者不知道上一次的消费位移。
那么如何保存消费位移？
旧的消费者客户端，消费位移是保存在zookeeper中。
新的消费者客户端，消费位移是保存在Kafka内部主题中，__consumer_offset，消费者在消费完消息后需要执行消费位移的提交。
那么提交的消费位移是什么？
提交的消费位移是x+1, x表示消费者已经消费了x位置的消息。
位移提交的时间？
位移提交也很有讲究，可能会造成重复消费和消息丢失的现象。
- 拉取到消息就提交，可能会造成消息丢失现象
- 消费完拉取的消息才提交，可能会造成数据重复消费
KAFKA默认是消费者自动定期（由参数设定）提交，自动位移提交也无法做到精确的位移管理，而且会带来重复消费和消息丢失问题。重复消费：在下一次自动提交之前崩溃，会从上一次消费位移拉取，导致重复消费，消息丢失：多线程下，一个线程负责拉取到本地缓存，一个线程负责具体的消费，真正的消费线程崩溃，导致消息丢失。
自动提交也会带来消费重复和消息丢失，怎么办？
Kafka提供了手动提交位移的方式，有时消息消费不是指拉取到消息就算消费成功，需要将数据写入数据库，本地缓存等，才能算消费成功。手动提交更加灵活。
手动提交位移的方式？
手动提交分为 同步提交 和 异步提交。
同步提交方式：也会带来重复消费问题
while(isRunning.get()) {
ConsumerRecords<String, String> records = consumer.poll(1000);
for (ConsumerRecord<String, String> record : records) {
// do some logical processing.
}
consumer.commitSync();
}
更多的时候按照分区的粒度同步提交位移。
long lastConsumedOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
consumer.commitSync(Collections.singletonmap(partition, new OffsetAndMetadata(lastConsumedoffset + 1)))；
异步提交
与commitSync方法相反，异步提交在执行消费者线程不会被阻塞，可能在提交消费位移的结果还未返回之前就开始了新一次拉取，使得消费者的性能得到一定增强。
2.2 指定位移消费
如果消费者找不到消费位移或者位移越界，就会根据消费者客户端参数anto.offset.reset 决定从和位置进行消费。
- earliest 从起始开始消费
- latest 从最新开始消费
- none 不消费
消息的拉取是基于poll()方法，无法精确掌握其消费的起始位置，提供的auto.offset.reset 参数也只是在找不到消费位移或位移越界情况下粗粒度从开头或末尾开始消费，KafkaConsumer中的seek(),让我们追前消费或回溯消费。
2.3 再均衡
再均衡是指分区的所属权从一个消费者转移到另一消费者的行为。他为消费组具体高可用和伸缩性提供保证，但是再均衡发生期间，消费组内的消费者无法读取消息。
再均衡发生期间会产生的问题？
- 首先消费组变得不可用，2当一个分区被重新分配给另一个消费者时，消费者当前的状态也会丢失，比如这个消费者在消费数据后还没来得及提交消费位移，就发生在均衡，会发生重复消费。
怎么解决重复消费问题？
将消费位移暂存到一个局部变量中，正常消费通过commitAsync()方法来异步提交消费位移，在发生再均衡动作前，通过再均衡监听器回调commitSync()方法同步提交消费位移，避免重复消费。
2.4 多线程实现
KafkaProducer是线程安全的，KafkaConsumer却是非线程安全的。
KafkaConsumer定义了一个acquire()方法，acquire()方法是一个轻量级锁，通过Cas保证并发下只有一个线程在操作。
如何进行多线程消费？
第一种：线程封闭，为每个线程实例化一个KafkaConsumer对象，每个线程消费一个或多个分区，一个分区只会被一个线程消费。
第二种：多个线程同时消费同一个分区。（不推荐）
滑动窗口解决
2.5 重要参数
max.poll.records 拉取最大的消息数
heartbeat.interval.ms 心跳到消费者协调器之间的时间，确保消费者保持会话，会触发rebalance
session.timeout.ms 组管理之间检测消费者是否失效的超时时间
max.poll.interval.ms 通过消费组管理消费者时，当超过这个时间还没有发起poll，认为消费者已经离开，将触发再均衡，也许只是处理的慢，假死