文章目录
- 一、消费消息概览
- 1.1、消费示例代码
- 1.2、消费过程
- 二、消费者组
- 2.1、push & pull
- 2.2、消费者组
- 三、调度器Coordinator
- 四、消费者分配策略
- 4.1、引言
- 4.2、分配基本流程
- 4.3、分配策略
- 4.3.1、轮询分配策略
- 4.3.2、轮询分配策略
- 五、消费偏移量
- 5.1、起始偏移量
- 5.2、指定偏移量消费
- 5.3、偏移量提交
- 5.3.1、自动提交
- 5.3.2、手动提交
- 5.4、偏移量的保存
- 六、消费者事务
- 6.1、消费者事务原理
- 6.2、示例代码
如果想了解kafka基础架构、生产者架构和kafka存储消息可以参考 kafka基础、 Kafka进阶_1.生产消息和 kafka进阶_2.存储消息。
一、消费消息概览
1.1、消费示例代码
public class KafkaConsumerTest {
public static void main(String[] args) {
// TODO 创建消费者配置参数集合
Map<String, Object> paramMap = new HashMap<>();
paramMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
paramMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
paramMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
paramMap.put(ConsumerConfig.GROUP_ID_CONFIG, "test");
// TODO 通过配置,创建消费者对象
KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(paramMap);
// TODO 订阅主题
consumer.subscribe(Collections.singletonList("test"));
// TODO 消费数据
final ConsumerRecords<String, String> poll = consumer.poll(Duration.ofMillis(100));
// TODO 遍历数据
for (ConsumerRecord<String, String> record : poll) {
System.out.println( record.value() );
}
// TODO 关闭消费者
consumer.close();
}
}
主要的属性配置:
参数名 | 参数作用 | 类型 | 推荐值 |
---|---|---|---|
bootstrap.servers | 集群地址 | 必须 | |
key.deserializer | 对数据Key进行反序列化的类完整名称 | 必须 | Kafka提供的字符串反序列化类:StringSerializer |
value.deserializer | 对数据Value进行反序列化的类完整名称 | 必须 | Kafka提供的字符串反序列化类:ValueSerializer |
group.id | 消费者组ID,用于标识完整的消费场景,一个组中可以包含多个不同的消费者对象 |
1.2、消费过程
消费者消费数据时,一般情况下,只是设定了订阅的主题名称,那是如何消费到数据的呢。我们这里说一下服务端拉取数据的基本流程:
- 服务端获取到用户拉取数据的请求:Kafka消费客户端会向Broker发送拉取数据的请求FetchRequest,服务端Broker获取到请求后根据请求标记FETCH交给应用处理接口KafkaApis进行处理。
- 通过副本管理器拉取数据:副本管理器需要确定当前拉取数据的分区,然后进行数据的读取操作。
- 判定首选副本:2.4版本前,数据读写的分区都是Leader分区,从2.4版本后,kafka支持Follower副本进行读取。主要原因就是跨机房或者说跨数据中心的场景,为了节约流量资源,可以从当前机房或数据中心的副本中获取数据。这个副本称之为首选副本。
- 拉取分区数据:Kafka的底层读取数据是采用日志段LogSegment对象进行操作的。
- 零拷贝:为了提高数据读取效率,Kafka的底层采用nio提供的FileChannel零拷贝技术,直接从操作系统内核中进行数据传输,提高数据拉取的效率。
二、消费者组
从数据处理的角度来讲,消费者和生产者的处理逻辑都相对比较简单。Producer生产者的基本数据处理逻辑就是向Kafka发送数据,并获取Kafka的数据接收确认响应。
而消费者的基本数据处理逻辑就是向Kafka请求数据,并获取Kafka返回的数据。
逻辑确实很简单,但是Kafka为了能够构建高吞吐,高可靠性,高并发的分布式消息传输系统,所以在很多细节上进行了扩展和改善:比如生产者可以指定分区,可以异步和同步发送数据,可以进行幂等性操作和事务处理。对应的,消费者功能和处理细节也进行了扩展和改善。
2.1、push & pull
Kafka的主题如果就一个分区的话,那么在硬件配置相同的情况下,消费者Consumer消费主题数据的方式没有什么太大的差别。
不过,Kafka为了能够构建高吞吐,高可靠性,高并发的分布式消息传输系统,它的主题是允许多个分区的,那么就会发现不同的消费数据的方式区别还是很大的。
1、如果数据由Kafka进行推送(push),那么多个分区的数据同时推送给消费者进行处理,明显一个消费者的消费能力是有限的,那么消费者无法快速处理数据,就会导致数据的积压,从而导致网络、存储等资源造成极大的压力,影响吞吐量和数据传输效率。
2.如果kafka的分区数据在内部可以存储的时间更长一些,再由消费者根据自己的消费能力向kafka申请(拉取)数据,那么整个数据处理的通道就会更顺畅一些。Kafka的Consumer就采用的这种拉取数据的方式。
2.2、消费者组
消费者可以根据自身的消费能力主动拉取Kafka的数据,但是毕竟自身的消费能力有限,如果主题分区的数据过多,那么消费的时间就会很长。对于kafka来讲,数据就需要长时间的进行存储,那么对Kafka集群资源的压力就非常大。如果希望提高消费者的消费能力,并且减少kafka集群的存储资源压力。所以有必要对消费者进行横向伸缩,从而提高消息消费速率。
不过这么做有一个问题,就是每一个消费者是独立,那么一个消费者就不能消费主题中的全部数据,简单来讲,就是对于某一个消费者个体来讲,主题中的部分数据是没有消费到的,也就会认为数据丢了,这个该如何解决呢?那如果我们将这多个消费者当成一个整体,是不是就可以了呢?这就是所谓的消费者组 Consumer Group。在kafka中每个消费者都对应一个消费组,消费者可以是一个线程,一个进程,一个服务实例,如果kafka想要消费消息,那么需要指定消费哪个topic的消息以及自己的消费组id(groupId)。
三、调度器Coordinator
消费者想要拉取数据,首先必须要加入到一个组中,成为消费组中的一员,同样道理,如果消费者出现了问题,也应该从消费者组中剥离。而这种加入组和退出组的处理,都应该由专门的管理组件进行处理,这个组件在kafka中,我们称之为消费者组调度器(Group Coordinator)。Group Coordinator是Broker上的一个组件,用于管理和调度消费者组的成员、状态、分区分配、偏移量等信息。每个Broker都有一个Group Coordinator对象,负责管理多个消费者组,但每个消费者组只有一个Group Coordinator。
四、消费者分配策略
4.1、引言
消费者想要拉取主题分区的数据,首先必须要加入到一个组中:
但是一个组中有多个消费者的话,那么每一个消费者该如何消费呢,是不是像图中一样的消费策略呢?如果是的话,那假设消费者组中只有2个消费者或有4个消费者,和分区的数量不匹配,怎么办?所以介绍一下Kafka中基本的消费者组中的消费者和分区之间的分配规则:
- 为了避免数据被重复消费,所以主题一个分区的数据只能被组中的一个消费者消费,但是反过来,一个消费者是可以消费多个分区数据的。
- 消费者组中的消费者数量最好不要超出主题分区的数据(资源充裕另说哈),就会导致多出的消费者是无法消费数据的,造成了资源的浪费。
消费者中的每个消费者到底消费哪一个主题分区,这个分配策略其实是由消费者的Leader决定的,这个Leader我们称之为群主。群主是多个消费者中,第一个加入组中的消费者,其他消费者我们称之为Follower,称呼上有点类似与分区的Leader和Follower。
当消费者加入群组的时候,会发送一个JoinGroup请求。群主负责给每一个消费者分配分区。每个消费者只知道自己的分配信息,只有群主知道群组内所有消费者的分配信息。
4.2、分配基本流程
4.3、分配策略
4.3.1、轮询分配策略
4.3.2、轮询分配策略
五、消费偏移量
偏移量offset是消费者消费数据的一个非常重要的属性。默认情况下,消费者如果不指定消费主题数据的偏移量,那么消费者启动消费时,无论当前主题之前存储了多少历史数据,消费者只能从连接成功后当前主题最新的数据偏移位置读取,而无法读取之前的任何数据,如果想要获取之前的数据,就需要设定配置参数或指定数据偏移量。
5.1、起始偏移量
在消费者的配置中,我们可以增加偏移量相关参数auto.offset.reset
,用于从最开始获取主题数据:
public class KafkaConsumerTest {
public static void main(String[] args) {
// TODO 创建消费者配置参数集合
Map<String, Object> paramMap = new HashMap<>();
paramMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
paramMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
paramMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
paramMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
paramMap.put(ConsumerConfig.GROUP_ID_CONFIG, "test");
// TODO 通过配置,创建消费者对象
KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(paramMap);
// TODO 订阅主题
consumer.subscribe(Arrays.asList("test"));
while ( true ) {
// TODO 消费数据
final ConsumerRecords<String, String> poll = consumer.poll(Duration.ofMillis(100));
// TODO 遍历数据
for (ConsumerRecord<String, String> record : poll) {
System.out.println( record );
}
}
}
}
auto.offset.reset有三种取值:
1、earliest
对于同一个消费者组,从头开始消费。就是说如果这个topic有历史消息存在,现在新启动了一个消费者组,且auto.offset.reset=earliest,那将会从头开始消费(未提交偏移量的场合)。
2、latest
对于同一个消费者组,消费者只能消费到连接topic后,新产生的数据(未提交偏移量的场合)。
3、none:生产环境不使用。
5.2、指定偏移量消费
除了上一节所讲的从最开始的偏移量或最后的偏移量读取数据以外,Kafka还支持从指定偏移量的位置开始消费数据:
public class KafkaConsumerOffsetTest {
public static void main(String[] args) {
Map<String, Object> paramMap = new HashMap<>();
paramMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
paramMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
paramMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
paramMap.put(ConsumerConfig.GROUP_ID_CONFIG, "test");
KafkaConsumer<String, String> c = new KafkaConsumer<String, String>(paramMap);
// TODO 订阅主题
c.subscribe(Collections.singletonList("test"));
// TODO 拉取数据,获取基本集群信息
c.poll(Duration.ofMillis(100));
// TODO 根据集群的基本信息配置需要消费的主题及偏移量
final Set<TopicPartition> assignment = c.assignment();
for (TopicPartition topicPartition : assignment) {
if ( topicPartition.topic().equals("test") ) {
c.seek(topicPartition, 0);
}
}
// TODO 拉取数据
while (true) {
final ConsumerRecords<String, String> poll = c.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : poll) {
System.out.println( record.value() );
}
}
}
}
5.3、偏移量提交
生产环境中,消费者可能因为某些原因或故障重新启动消费,那么如果不知道之前消费数据的位置,重启后再消费,就可能重复消费(earliest)或漏消费(latest)。所以Kafka提供了保存消费者偏移量的功能,而这个功能需要由消费者进行提交操作。这样消费者重启后就可以根据之前提交的偏移量进行消费了。
注意:一旦消费者提交了偏移量,那么kafka会优先使用提交的偏移量进行消费。此时,auto.offset.reset参数是不起作用的。
5.3.1、自动提交
所谓的自动提交就是消费者消费完数据后,无需告知kafka当前消费数据的偏移量,而是由消费者客户端API周期性地将消费的偏移量提交到Kafka中。这个周期默认为5000ms
,可以通过配置进行修改。
public class KafkaConsumerCommitAutoTest {
public static void main(String[] args) {
// TODO 创建消费者配置参数集合
Map<String, Object> paramMap = new HashMap<>();
paramMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
// TODO 启用自动提交消费偏移量,默认取值为true
paramMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
// TODO 设置自动提交offset的时间周期为1000ms,默认5000ms
paramMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000);
paramMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
paramMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
paramMap.put(ConsumerConfig.GROUP_ID_CONFIG, "test");
// TODO 通过配置,创建消费者对象
KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(paramMap);
// TODO 订阅主题
consumer.subscribe(Arrays.asList("test"));
while ( true ) {
// TODO 消费数据
final ConsumerRecords<String, String> poll = consumer.poll(Duration.ofMillis(100));
// TODO 遍历数据
for (ConsumerRecord<String, String> record : poll) {
System.out.println( record );
}
}
}
}
5.3.2、手动提交
基于时间周期的偏移量提交,是我们无法控制的,一旦参数设置的不合理,或单位时间内数据量消费的很多,却没有来及的自动提交,那么数据就会重复消费。所以Kafka也支持消费偏移量的手动提交,也就是说当消费者消费完数据后,自行通过API进行提交。不过为了考虑效率和安全,kafka同时提供了异步提交和同步提交两种方式供我们选择。注意:需要禁用自动提交ENABLE_AUTO_COMMIT_CONFIG=false,才能开启手动提交:
1、异步提交
向Kafka发送偏移量offset提交请求后,就可以直接消费下一批数据,因为无需等待kafka的提交确认,所以无法知道当前的偏移量一定提交成功,安全性比较低,但相对,消费性能会提高
public class KafkaConsumerCommitASyncTest {
public static void main(String[] args) {
// TODO 创建消费者配置参数集合
Map<String, Object> paramMap = new HashMap<>();
paramMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
// TODO 禁用自动提交消费偏移量,默认取值为true
paramMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
paramMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
paramMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
paramMap.put(ConsumerConfig.GROUP_ID_CONFIG, "test");
// TODO 通过配置,创建消费者对象
KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(paramMap);
// TODO 订阅主题
consumer.subscribe(Arrays.asList("test"));
while ( true ) {
// TODO 消费数据
final ConsumerRecords<String, String> poll = consumer.poll(Duration.ofMillis(100));
// TODO 遍历处理数据
for (ConsumerRecord<String, String> record : poll) {
System.out.println( record );
}
// TODO 异步提交偏移量
// 此处需要注意,需要在拉取数据完成处理后再提交
// 否则提前提交了,但数据处理失败,下一次消费数据就拉取不到了
consumer.commitAsync();
}
}
}
2、同步提交
必须等待Kafka完成offset提交请求的响应后,才可以消费下一批数据,一旦提交失败,会进行重试处理,尽可能保证偏移量提交成功,但是依然可能因为意外情况导致提交请求失败。此种方式消费效率比较低,但是安全性高。
5.4、偏移量的保存
由于消费者在消费消息的时候可能会由于各种原因而断开消费,当重新启动消费者时我们需要让它接着上次消费的位置offset继续消费,因此消费者需要实时的记录自己以及消费的位置。0.90版本之前,这个信息是记录在zookeeper内的,在0.90之后的版本,offset保存在_consumer_offsets
这个topic内。
每个consumer会定期将自己消费分区的offset提交给_consumer_offsets,提交过去的时候,key是consumerGroupId+topic+分区号
:
六、消费者事务
6.1、消费者事务原理
无论偏移量使用自动提交还是手动提交,特殊场景中数据都有可能会出现重复消费:
如果提前提交偏移量,再处理业务,又可能出现数据丢失的情况:
对于单独的Consumer来讲,事务保证会比较弱,尤其是无法保证提交的信息被精确消费,主要原因就是消费者可以通过偏移量访问信息,而不同的数据文件生命周期不同,同一事务的信息可能会因为重启导致被删除的情况。所以一般情况下,想要完成kafka消费者端的事务处理,需要将数据消费过程和偏移量提交过程进行原子性绑定,也就是说数据处理完了,必须要保证偏移量正确提交,才可以做下一步的操作,如果偏移量提交失败,那么数据就恢复成处理之前的效果。
对于生产者事务而言,消费者消费的数据也会受到限制。默认情况下,消费者只能消费到生产者已提交的数据。如果想要消费到未提交的数据,需要更高消费事务隔离级别。
6.2、示例代码
public class KafkaConsumerTransactionTest {
public static void main(String[] args) {
Map<String, Object> paramMap = new HashMap<>();
paramMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
paramMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
paramMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
// TODO 隔离级别:已提交读,读取已经提交事务成功的数据(默认)
//paramMap.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
// TODO 隔离级别:未提交读,读取已经提交事务成功和未提交事务成功的数据
paramMap.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_uncommitted");
paramMap.put(ConsumerConfig.GROUP_ID_CONFIG, "test");
KafkaConsumer<String, String> c = new KafkaConsumer<String, String>(paramMap);
c.subscribe(Collections.singletonList("test"));
while (true) {
final ConsumerRecords<String, String> poll = c.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : poll) {
System.out.println( record.value() );
}
}
}
}
总结
- 消费者想要拉取主题分区的数据,首先必须要加入到一个组中。
- 一个生产者可以向多个topic推送数据,可以向一个topic中的多个分区推送数据。
- 一个消费者组可以订阅多个topic,一个消费者可以订阅多个topic,一个消费者可以消费多个分区数据,一个分区可以被多个消费者组消费,但是一个分区只能被消费者组中的一个消费者消费。