赞
踩
一、概念理解
Kafka是一个分布式、支持分区的(partition)、多副本的(replica),基于zookeeper协调的分布式消息系统,它的最大的特性就是可以实时的处理大量数据。
1、Kafka的特性
2、发布订阅模式
一个或多个生产者产生的消息能够被多个消费者同时消费的情况
三、Kafka的架构原理
上面简单的介绍了为什么需要消息队列以及消息队列通信的两种模式,下面主角介绍Kafka。Kafka是一种高吞吐量的分布式发布订阅消息系统,它可以处理消费者规模的网站中的所有动作流数据,具有高性能、持久化、多副本备份、横向扩展能力。。
1、基础架构与名词解释
1.方便扩展:因为一个topic可以有多个partition,所以我们可以通过扩展机器去轻松的应对日益增长的数据量。
2.提高并发:以partition为读写单位,可以多个消费者同时消费数据,提高了消息的处理效率。
kafka如果某个topic有多个partition,producer又怎么知道该将数据发往哪个partition呢?kafka中有几个原则:
1.partition在写入的时候可以指定需要写入的partition,如果有指定,则写入对应的partition。
2.如果没有指定partition,但是设置了数据的key,则会根据key的值hash出一个partition。
3.如果既没指定partition,又没有设置key,则会轮询选出一个partition。
保证消息不丢失是一个消息队列中间件的基本保证,那producer在向kafka写入消息的时候,怎么保证消息不丢失呢?其实上面的写入流程图中有描述出来,那就是通过ACK应答机制!在生产者向队列写入数据的时候可以设置参数来确定是否确认kafka接收到数据,这个参数可设置的值为0、1、all。
(2)Message结构
上面说到log文件就实际是存储message的地方,我们在producer往kafka写入的也是一条一条的message,那存储在log中的message是什么样子的呢?消息主要包含消息体、消息大小、offset、压缩类型……等等!我们重点需要知道的是下面三个:
在保存数据的小节里面,我们聊到了partition划分为多组segment,每个segment又包含.log、.index、.timeindex文件,存放的每条message包含offset、消息大小、消息体……我们多次提到segment和offset,查找消息的时候是怎么利用segment+offset配合查找的呢?假如现在需要查找一个offset为368801的message是什么样的过程呢?我们先看看下面的图:
1.先找到offset的368801message所在的segment文件(利用二分法查找),这里找到的就是在第二个segment文件。
2.打开找到的segment中的.index文件(也就是368796.index文件,该文件起始偏移量为368796+1,我们要查找的offset为368801的message在该index内的偏移量为368796+5=368801,所以这里要查找的相对offset为5)。由于该文件采用的是稀疏索引的方式存储着相对offset及对应message物理偏移量的关系,所以直接找相对offset为5的索引找不到,这里同样利用二分法查找相对offset小于或者等于指定的相对offset的索引条目中最大的那个相对offset,所以找到的是相对offset为4的这个索引。
3.根据找到的相对offset为4的索引确定message存储的物理偏移位置为256。打开数据文件,从位置为256的那个地方开始顺序扫描直到找到offset为368801的那条Message。
这套机制是建立在offset为有序的基础上,利用segment+有序offset+稀疏索引+二分查找+顺序查找等多种手段来高效的查找数据!至此,消费者就能拿到需要处理的数据进行处理了。那每个消费者消费到的offset已经直接维护在kafk集群的__consumer_offsets这个topic中,key是group.id+topic+分区号,value就是当前offset的值。
四、消费者配置
消费者实例<=partion数量
3.1 offset
Offset,消息位移,它表示分区中每条消息的位置信息,是一个单调递增且不变的值。换句话说,offset可以用来唯一的标识分区中每一条记录,消费者消费完一条消息记录之后,需要提交offset来告诉Kafka Broker自己消费到哪里了。
(1)自动提交:
Kafka为了使我们能够专注于自己的业务逻辑,提供了自动提交offset的功能,这也是默认配置项。
我们需要关注以下两个配置参数:
enable.auto.commit:是否开启自动提交offset功能,默认是true
auto.commit.interval.ms:自动提交offset的时间间隔,默认是5s
自动提交可能存在消息重复消费或丢失
一个典型的自动提交的边界场景是分区再均衡场景。假设我们采用默认 5 秒的自动提交时间间隔,在本分区最近一次提交后 3 秒发生了故障,再均衡之后,新的消费者从本分区的消费位点开始读取并处理消息。由于消费位点是故障前 3 秒前自动提交的,在这 3 秒之间读取的消息及其影响的消费位点没有被提交,因此这些数据将被重复处理。
自动提交的配置下,每一次 poll 调用都会提交上一次 poll 移动到的消费位点,在调用消费者的 close 方法时也会触发自动提交。通常来说,自动提交的消费位点总是不大于消费者实际处理的消息。但是,如果在轮询时拉取到一批消息,并在处理完所有消息之前抛出异常,就有可能导致自动提交时按照这批消息处理过的假设进行提交,从而导致部分消息被跳过,不被处理的情形
(2)手动提交
虽然自动提交offset带来了很大的便利,但是在消息的可靠性上不太容易掌控,因此Kafka也提供了手动提交offset这个功能。
(3)指定offset消费方式
Kafka针对offset的消费方式提供了三种类型:earliest | latest | none,默认是latest,即从最新的offset开始消费。
(1)earliest:自动将偏移量 重置为最早的,–fromfromfrom。
(2)latest(默认值):自动将偏移量重置为最新偏移量。
(3)none :如果未找到消费者组的先前偏移量,则向抛出异常。
3.2 session 超时和心跳监测
session.timeout.ms:consumer 的 session 超时时间,默认 10s。超过这个时间将会认为 consumer 挂了。将 reblance (重新平衡)消费者。
heartbeat.interval.ms:心跳监测时间,默认值 3s。consumer 每 3s发送一次心跳给协调者(broker)。
3.3 拉取大消息
消费过程是由客户端主动去服务端拉取消息的,在拉取大消息时,需要注意控制拉取速度,注意修改配置:
max.poll.records:默认值:500。如果单条消息超过1 MB,建议设置为1。
fetch.max.bytes:设置比单条消息的大小略大一点。
max.partition.fetch.bytes:设置比单条消息的大小略大一点。
3.4 提高消费速度
(1)可以考虑给Kafka增加Topic的分区数,并同步增加消费者Consumer的实例数
(2)增加并发消费线程
concurrency: 并发线程数,表示一个消费者实例启动的Kafka消费者线程的数量,这多个线程可以并行消费来自一个或多个分区的消息
(3)提高每批次拉取的数量
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。